FabledScribe/src/scribe/services/family_adoption.py

"""Family canon's adoption ledger (milestone 463 step 4).

Promotion (services/family.py) decides that an idea is canon for some
platforms. This module decides what each project on those platforms does
about it, and keeps the answers honest as the canon moves.

ASSESSMENT — one project, one idea, one of four outcomes, judged in order:

  exempt  → adopted/variant/owed are never reached: the idea does not apply.
  variant → it applies, and the project departs for a FACT about itself.
  adopted → it applies and the project does it; the evidence says where.
  owed    → none of the above. A task is filed in the project to do it.

No person approves an assessment. Consistency comes from precedent: every
assessment records the answers it was checked against — the same idea's
answers in other projects, and this project's answers to the nearest ideas —
found by the engine itself, so "which precedents were consulted" is a fact
about the call. Asking the same question twice with the same answer records
nothing the second time.

OWED → TASK. An `owed` answer files a task in the OWING project — tagged to
the System matching the idea's, naming the gap and the reference
implementation for that project's language. Nothing here edits a repository:
the work is filed where its own project will pick it up. When the answer
moves on, the task follows: `adopted` closes it as done, `exempt` or
`variant` cancels it, `owed` again reopens it. Each change is logged on the
task with the reason.

RECHECK. An answer records the canon version it was given against. When the
idea's version moves (re-promotion, revision, conflict resolution), every
answer given against another version reads `needs_recheck` — DERIVED by
comparing the two numbers, never a stored flag that could go stale.

CONFLICT. When two projects solve the same idea differently and each thinks
its way is the right one, the conflict order decides — the first ground that
applies wins, and every ground above it must be said not to:

  1. a stance the operator stated;
  2. the approach that covers a recorded failure (the other becomes owed);
  3. both right under different conditions (the canon splits by condition);
  4. the most recently verified and most complete.

The losing side's reasoning is folded into the idea's note — as a trap, an
alternative, or the branch for its condition — and is never dropped. The
substance changed, so the version moves.

SHAPES (step 5). The shape ledger answers too: a shape classified against an
idea's reference — in any language — is the project's code saying `adopted`
(an instance) or `variant` (with the shape's reason). The two ledgers never
disagree: a classification moves the row, an assessment or undo the shapes
contradict is refused, an answer the shapes gave is withdrawn with them, and
a conflict resolution re-judges the variant shapes on both sides. The shape
proposer offers an idea's reference across projects and languages by
meaning; its rule and the measurement behind it sit above
shape_ledger._FAMILY_FLOOR.
"""
from __future__ import annotations

import logging
from typing import Iterable

from sqlalchemy import func, select

from scribe.models import async_session
from scribe.models.code_shape import CodeShape
from scribe.models.family import (
    FamilyAdoption, FamilyDecision, FamilyIdea, FamilyIdeaPlatform, FamilyIdeaReference,
    Platform, ProjectPlatform,
)
from scribe.models.note import Note, TaskStatus
from scribe.models.project import Project
from scribe.models.system import RecordSystem, System
from scribe.services import access
from scribe.services import family as family_svc

logger = logging.getLogger(__name__)

# --- the assessment and the conflict order, as product text --------------------
#
# Judged on every install (rules 115, 119); the tool docstrings quote them.

OUTCOMES = (
    {
        "key": "exempt",
        "title": "Does not apply",
        "test": (
            "The idea's 'when it applies' is false for this project. The reason "
            "names the fact about this project that makes it false."
        ),
    },
    {
        "key": "variant",
        "title": "Applies, and the project departs",
        "test": (
            "It applies, and the project departs for a reason that names a FACT "
            "about itself the canon did not account for. A preference, a taste, "
            "or 'we already did it another way' is not a reason: that is owed — "
            "or, if this project's way is better, a conflict to resolve."
        ),
    },
    {
        "key": "adopted",
        "title": "Applies, and the project does it",
        "test": (
            "It applies and the project does it. The evidence names where — a "
            "file, a commit, a task, a CI run."
        ),
    },
    {
        "key": "owed",
        "title": "Applies, and is not done yet",
        "test": (
            "It applies, nothing above excuses it, and the project does not do "
            "it yet. A task is filed in this project to do it."
        ),
    },
)
OUTCOME_KEYS = tuple(o["key"] for o in OUTCOMES)

CONFLICT_ORDER = (
    {
        "key": "operator_stance",
        "title": "A stance the operator stated",
        "test": (
            "One side follows a stance the operator stated — a rule, a "
            "preference, a recorded decision. It wins over anything inferred. "
            "Name the stance in the evidence."
        ),
        "fold_as": "alternative",
    },
    {
        "key": "covers_failure",
        "title": "Covers a recorded failure",
        "test": (
            "One side covers a failure that is on record — an incident, an "
            "issue, a lesson — and the other does not. Name the failure in the "
            "evidence. The other side becomes owed."
        ),
        "fold_as": "trap",
    },
    {
        "key": "split_by_condition",
        "title": "Both right, under different conditions",
        "test": (
            "Each side is right under a condition the other does not meet. The "
            "canon splits: each approach is canon where its condition holds. "
            "State both conditions."
        ),
        "fold_as": "condition",
    },
    {
        "key": "most_recent_complete",
        "title": "Most recently verified and most complete",
        "test": (
            "None of the above decides. The side verified most recently, and "
            "covering the most, wins. Name the verification in the evidence."
        ),
        "fold_as": "alternative",
    },
)
CONFLICT_KEYS = tuple(c["key"] for c in CONFLICT_ORDER)
_GROUND = {c["key"]: c for c in CONFLICT_ORDER}

_OPEN_TASK = (TaskStatus.todo.value, TaskStatus.in_progress.value)


def _clean(items) -> list[str]:
    return [str(e).strip() for e in items or [] if str(e).strip()]


# --- the pure checks ---------------------------------------------------------------

def assessment_problems(*, outcome: str, reason: str, evidence: list | None) -> list[str]:
    """What an assessment is missing before it can be recorded. Pure.

    Every outcome needs a reason — it is what the next assessment of a
    similar idea is checked against. `adopted` also needs evidence naming
    where the project does it.
    """
    if outcome not in OUTCOME_KEYS:
        return [f"outcome must be one of: {', '.join(OUTCOME_KEYS)}"]
    problems = []
    if not (reason or "").strip():
        problems.append({
            "exempt": "exempt needs a reason naming the fact that makes 'when it applies' false here",
            "variant": "variant needs a reason naming the fact about this project the canon missed",
            "adopted": "adopted needs a reason",
            "owed": "owed needs a reason naming the gap",
        }[outcome])
    if outcome == "adopted" and not _clean(evidence):
        problems.append("adopted needs evidence naming where the project does it")
    return problems


def conflict_problems(
    *, ground: str, grounds_checked: dict | None, evidence: list | None,
    conditions: dict | None, fold: str,
) -> list[str]:
    """What a conflict resolution is missing before it can be recorded. Pure.

    The order is enforced, not suggested: every ground ABOVE the deciding one
    must carry a sentence saying why it did not decide. Grounds 1, 2 and 4
    each need evidence (the stance, the failure, the verification); a split
    needs both conditions; and the losing side's reasoning is required,
    because it is folded into the idea rather than dropped.
    """
    if ground not in CONFLICT_KEYS:
        return [f"ground must be one of, in order: {', '.join(CONFLICT_KEYS)}"]
    checked = grounds_checked or {}
    problems = [
        f"'{k}' comes before '{ground}' in the conflict order — say why it did not "
        f"decide (grounds_checked['{k}'])"
        for k in CONFLICT_KEYS[:CONFLICT_KEYS.index(ground)]
        if not str(checked.get(k) or "").strip()
    ]
    if ground == "split_by_condition":
        conds = conditions or {}
        if not str(conds.get("canon") or "").strip() or not str(conds.get("other") or "").strip():
            problems.append(
                "a split needs both conditions: conditions={'canon': …, 'other': …}")
    elif not _clean(evidence):
        problems.append({
            "operator_stance": "name the operator's stance in evidence (a rule, a preference, a decision)",
            "covers_failure": "name the recorded failure in evidence (an incident, an issue, a lesson)",
            "most_recent_complete": "name the verification in evidence (a CI run, a device check, a date)",
        }[ground])
    if not (fold or "").strip():
        problems.append(
            "the losing side's reasoning is folded into the idea, never dropped — `fold` is required")
    return problems


def needs_recheck(row_status: str, row_version: int | None, idea_status: str,
                  idea_version: int) -> bool:
    """An answer given against a canon version that is not the current one.
    Unassessed rows have nothing to recheck; a retired idea asks nothing."""
    return (
        idea_status == "canon" and row_status != "unassessed"
        and row_version is not None and row_version != idea_version
    )


def _row_state(row: FamilyAdoption) -> dict:
    """An answer as a decision's before/after. No ids (the #3182 trap): the
    owed task is the row's column, not part of the snapshot."""
    return {"status": row.status, "reason": row.reason or "", "canon_version": row.canon_version}


# --- reads -------------------------------------------------------------------------

async def _row(session, project_id: int, idea_id: int) -> FamilyAdoption | None:
    return (await session.execute(
        select(FamilyAdoption).where(
            FamilyAdoption.project_id == project_id, FamilyAdoption.idea_id == idea_id)
    )).scalars().first()


async def _is_member(session, project_id: int, idea_id: int) -> bool:
    return bool(await session.scalar(
        select(func.count()).select_from(ProjectPlatform)
        .join(FamilyIdeaPlatform, FamilyIdeaPlatform.platform_id == ProjectPlatform.platform_id)
        .where(
            ProjectPlatform.project_id == project_id,
            FamilyIdeaPlatform.note_id == idea_id,
            ProjectPlatform.state.in_(family_svc.MEMBER_STATES),
        )
    ))


async def _references(session, idea_id: int) -> list[dict]:
    rows = (await session.execute(
        select(Note)
        .join(FamilyIdeaReference, FamilyIdeaReference.snippet_id == Note.id)
        .where(FamilyIdeaReference.idea_id == idea_id, Note.deleted_at.is_(None))
        .order_by(Note.id.asc())
    )).scalars().all()
    return [
        {"id": n.id, "title": n.title,
         "language": str((n.data or {}).get("language") or "").strip().lower()}
        for n in rows
    ]


async def _project_languages(session, project_id: int) -> list[str]:
    """The languages of the snippets recorded in a project, most used first —
    what "the reference for this project's language" is matched against."""
    lang = Note.data["language"].astext
    rows = (await session.execute(
        select(func.lower(lang), func.count())
        .where(Note.project_id == project_id, Note.note_type == "snippet",
               Note.deleted_at.is_(None), lang.is_not(None), lang != "")
        .group_by(func.lower(lang)).order_by(func.count().desc())
    )).all()
    return [r[0] for r in rows]


# --- the shape ledger as evidence (milestone 463 step 5) ----------------------------
#
# A shape classified against an idea's reference IS an answer to the idea: an
# `instance` — or the reference's own `canonical` location — says the project
# does it; a `variant` says it departs, for the reason the shape carries. The
# two ledgers must never disagree, and the shapes are the stronger evidence
# (they name the code), so:
#
#   - classifying a shape moves the adoption row to what the shapes say,
#     decided_via "system", the shapes as evidence;
#   - an assessment or an undo that would contradict the shapes is refused —
#     reclassify the shapes if they are wrong;
#   - an answer the shape ledger gave is withdrawn to `unassessed` when the
#     shapes that gave it are withdrawn. An answer given on other evidence is
#     left alone: no shape is not the same as `owed`.

SHAPE_SOURCE = "shape_ledger"
_SHAPE_ADOPTS = ("canonical", "instance")

# Which reference a shape is judged against when it is classified against an
# idea: the one in the shape's language when there is one.
_EXTS_BY_LANGUAGE = {
    "python": (".py", ".pyi"), "go": (".go",), "kotlin": (".kt", ".kts"),
    "java": (".java",), "dart": (".dart",), "swift": (".swift",), "rust": (".rs",),
    "typescript": (".ts", ".tsx"), "javascript": (".js", ".jsx", ".mjs", ".cjs"),
    "vue": (".vue",), "svelte": (".svelte",), "css": (".css",), "scss": (".scss",),
    "bash": (".sh", ".bash"), "sh": (".sh",), "sql": (".sql",),
}


def path_speaks(path: str, language: str) -> bool:
    """Is a file at ``path`` written in ``language`` (a snippet's recorded one)?"""
    exts = _EXTS_BY_LANGUAGE.get((language or "").strip().lower())
    return bool(exts) and (path or "").strip().lower().endswith(exts)


def shape_outcome(statuses: Iterable[str]) -> str | None:
    """What a project's shapes say about an idea: `adopted` when any shape is
    an instance of (or is) a reference, `variant` when the only ones are
    variants, None when no shape speaks. None is silence, never `owed`."""
    seen = set(statuses)
    if seen & set(_SHAPE_ADOPTS):
        return "adopted"
    if "variant" in seen:
        return "variant"
    return None


def shapes_say(rows) -> dict | None:
    """The answer a project's evidence rows give — outcome, reason, evidence
    lines and the shapes themselves — or None when no shape speaks. The
    `adopted` reason is fixed text, so classifying one more instance does not
    log a new decision; a `variant` carries the first variant shape's why."""
    outcome = shape_outcome(r.status for r in rows)
    if outcome is None:
        return None
    speaking = [r for r in rows if (r.status in _SHAPE_ADOPTS) == (outcome == "adopted")]
    if outcome == "adopted":
        reason = "the shape ledger classifies this project's code as the idea's reference implementation"
    else:
        first = speaking[0]
        reason = (f"the shape ledger classifies {first.path}::{first.symbol} as a variant "
                  f"of #{first.snippet_id}: {(first.reason or '').strip()}")
    return {
        "outcome": outcome,
        "reason": reason,
        "evidence": [f"{r.path}::{r.symbol}" for r in speaking],
        "shapes": [{"id": r.id, "path": r.path, "symbol": r.symbol, "status": r.status,
                    "snippet_id": r.snippet_id} for r in speaking],
    }


async def _shape_evidence(session, project_id: int, idea_id: int) -> list[CodeShape]:
    """The project's live shapes judged against any of the idea's references."""
    refs = select(FamilyIdeaReference.snippet_id).where(FamilyIdeaReference.idea_id == idea_id)
    return list((await session.execute(
        select(CodeShape).where(
            CodeShape.project_id == project_id,
            CodeShape.vanished_at.is_(None),
            CodeShape.snippet_id.in_(refs),
            CodeShape.status.in_(_SHAPE_ADOPTS + ("variant",)),
        ).order_by(CodeShape.path, CodeShape.symbol)
    )).scalars().all())


def _contradiction(outcome: str, said: dict | None, idea_id: int) -> str | None:
    if said is None or outcome == said["outcome"]:
        return None
    named = ", ".join(said["evidence"][:3]) + (" …" if len(said["evidence"]) > 3 else "")
    return (
        f"the shape ledger already answers #{idea_id} in this project as {said['outcome']} "
        f"({named}). The two ledgers never disagree: reclassify those shapes "
        "(classify_shapes) if they are wrong, and the answer follows")


async def reference_for(user_id: int, idea_id: int, path: str) -> int:
    """The reference snippet a shape at ``path`` is judged against when it is
    classified against a family idea: the one in the path's language when
    there is one, else the first. An idea is shared across languages, so a
    Python shape can be an instance of an idea whose only reference is Go.
    Raises ValueError when the idea is not a family idea the caller can read
    or has no reference they can read."""
    if not await access.can_read_note(user_id, idea_id):
        raise ValueError(f"#{idea_id} is not a family idea you can read")
    async with async_session() as session:
        if await session.get(FamilyIdea, idea_id) is None:
            raise ValueError(f"#{idea_id} is not a family idea")
        refs = await _references(session, idea_id)
    readable = [r for r in refs if await access.can_read_note(user_id, r["id"])]
    if not readable:
        raise ValueError(
            f"family idea #{idea_id} has no reference implementation you can read — "
            "set_family_references first, or classify against a snippet_id")
    for r in readable:
        if path_speaks(path, r["language"]):
            return r["id"]
    return readable[0]["id"]


async def family_reference_ids(project_id: int) -> set[int]:
    """The reference snippets of every canon idea that reaches the project —
    what the shape proposer may match across projects and languages."""
    async with async_session() as session:
        return set((await session.execute(
            select(FamilyIdeaReference.snippet_id)
            .join(FamilyIdea, FamilyIdea.note_id == FamilyIdeaReference.idea_id)
            .join(FamilyIdeaPlatform, FamilyIdeaPlatform.note_id == FamilyIdea.note_id)
            .join(ProjectPlatform, ProjectPlatform.platform_id == FamilyIdeaPlatform.platform_id)
            .where(
                FamilyIdea.status == "canon",
                ProjectPlatform.project_id == project_id,
                ProjectPlatform.state.in_(family_svc.MEMBER_STATES),
            )
        )).scalars().all())


async def _latest_source(session, project_id: int, idea_id: int) -> str:
    latest = (await session.execute(
        select(FamilyDecision).where(
            FamilyDecision.idea_id == idea_id, FamilyDecision.project_id == project_id)
        .order_by(FamilyDecision.id.desc()).limit(1)
    )).scalars().first()
    return str(((latest.evidence if latest is not None else None) or {}).get("source") or "")


async def _withdraw_shape_answer(user_id: int, project_id: int, idea_id: int) -> FamilyDecision:
    reason = ("the shapes that answered this idea were withdrawn from the shape ledger; "
              "it is unassessed again")
    async with async_session() as session:
        row = await _row(session, project_id, idea_id)
        before = _row_state(row)
        row.status = "unassessed"
        row.reason = None
        row.canon_version = None
        row.assessed_at = None
        row.decided_via = None
        row.updated_at = family_svc._now()
        decision = family_svc._log(
            session, idea_id=idea_id, project_id=project_id, action="assess",
            reason=reason, before=before, after=_row_state(row),
            evidence={"source": SHAPE_SOURCE, "shapes": []}, precedent_ids=None,
            decided_via="system", user_id=user_id,
        )
        await session.commit()
        await session.refresh(decision)
        row_id = row.id
    await _sync_owed_task(user_id, row_id, reason=reason)
    return decision


async def sync_from_shapes(user_id: int, project_id: int,
                           snippet_ids: Iterable[int] | None = None) -> list[dict]:
    """Bring the project's adoption rows into line with its shape ledger, for
    the canon ideas whose references are among ``snippet_ids`` — or every
    canon idea that reaches the project when None (the coverage refresh,
    which also sees shapes vanish).

    A row that already gives the shapes' outcome at the current version is
    left alone, whatever evidence it was given on. Returns one entry per
    answer that moved: {idea_id, status, decision_id}."""
    query = select(FamilyIdea.note_id).where(FamilyIdea.status == "canon")
    if snippet_ids is not None:
        wanted = {int(s) for s in snippet_ids if s}
        if not wanted:
            return []
        query = query.where(FamilyIdea.note_id.in_(
            select(FamilyIdeaReference.idea_id).where(FamilyIdeaReference.snippet_id.in_(wanted))))
    async with async_session() as session:
        idea_ids = [i for i in (await session.execute(query)).scalars().all()
                    if await _is_member(session, project_id, i)]
    moved: list[dict] = []
    for idea_id in sorted(idea_ids):
        if not await access.can_read_note(user_id, idea_id):
            continue
        async with async_session() as session:
            idea = await session.get(FamilyIdea, idea_id)
            said = shapes_say(await _shape_evidence(session, project_id, idea_id))
            row = await _row(session, project_id, idea_id)
            current = _row_state(row) if row is not None else None
            source = await _latest_source(session, project_id, idea_id)
        if said is not None:
            if current and current["status"] == said["outcome"] \
                    and current["canon_version"] == idea.canon_version:
                continue
            result = await assess(
                user_id, project_id, idea_id, outcome=said["outcome"], reason=said["reason"],
                evidence=said["evidence"], decided_via="system", source=SHAPE_SOURCE,
                shapes=said["shapes"],
            )
            if result["changed"]:
                moved.append({"idea_id": idea_id, "status": said["outcome"],
                              "decision_id": result["decision"]["id"]})
        elif current and current["status"] in ("adopted", "variant") and source == SHAPE_SOURCE:
            decision = await _withdraw_shape_answer(user_id, project_id, idea_id)
            moved.append({"idea_id": idea_id, "status": "unassessed", "decision_id": decision.id})
    return moved


async def assessment_precedents(user_id: int, project_id: int, idea_id: int,
                                limit: int = 5) -> list[dict]:
    """What an assessment should be consistent with: the latest answer to
    THIS idea in each other project the caller can read, then this project's
    latest answers to the ideas nearest this one by meaning."""
    async with async_session() as session:
        same = (await session.execute(
            select(func.max(FamilyDecision.id))
            .where(FamilyDecision.idea_id == idea_id, FamilyDecision.action == "assess",
                   FamilyDecision.project_id.is_not(None),
                   FamilyDecision.project_id != project_id)
            .group_by(FamilyDecision.project_id)
        )).scalars().all()
    near = await family_svc.nearest_ideas(user_id, idea_id, limit=3)
    near_ids = [n.id for _, n in near]
    async with async_session() as session:
        mine = (await session.execute(
            select(func.max(FamilyDecision.id))
            .where(FamilyDecision.idea_id.in_(near_ids), FamilyDecision.action == "assess",
                   FamilyDecision.project_id == project_id)
            .group_by(FamilyDecision.idea_id)
        )).scalars().all() if near_ids else []
        rows = (await session.execute(
            select(FamilyDecision, Note.title, Project.title)
            .join(Note, Note.id == FamilyDecision.idea_id)
            .join(Project, Project.id == FamilyDecision.project_id)
            .where(FamilyDecision.id.in_(list(same) + list(mine)))
        )).all()
    scores = {n.id: round(float(s), 3) for s, n in near}
    out_same, out_near = [], []
    for d, idea_title, project_title in rows:
        if not await access.can_read_project(user_id, d.project_id):
            continue
        item = d.to_dict()
        item.update({"idea_title": idea_title, "project_title": project_title})
        if d.idea_id == idea_id:
            item["relation"] = "same idea, another project"
            out_same.append(item)
        else:
            item["relation"] = "nearest idea, this project"
            item["similarity"] = scores.get(d.idea_id)
            out_near.append(item)
    out_same.sort(key=lambda x: -x["id"])
    out_near.sort(key=lambda x: -(x.get("similarity") or 0))
    return (out_same + out_near)[:limit]


async def adoption_matrix(
    user_id: int, *, platform: str | None = None, project_id: int | None = None,
) -> dict:
    """Projects × canon ideas: every answer, and every project an idea
    reaches that has none yet.

    A cell exists where the project is on one of the idea's platforms, or
    has an answer from before the idea's scope changed. A member project with
    no row reads `unassessed` with `reached: False` — the promoter could not
    write it, so nobody has been asked. Projects and ideas the caller cannot
    read are left out.
    """
    async with async_session() as session:
        query = (
            select(FamilyIdea, Note.title, Note.note_type, Note.status)
            .join(Note, Note.id == FamilyIdea.note_id)
            .where(FamilyIdea.status == "canon", Note.deleted_at.is_(None),
                   access.readable_notes_clause(user_id))
        )
        if platform:
            query = query.where(FamilyIdea.note_id.in_(
                select(FamilyIdeaPlatform.note_id)
                .join(Platform, Platform.id == FamilyIdeaPlatform.platform_id)
                .where(Platform.slug == platform)
            ))
        ideas = (await session.execute(query.order_by(Note.title.asc()))).all()
        idea_ids = [i.note_id for i, *_ in ideas]
        idea_platforms: dict[int, dict[int, str]] = {i: {} for i in idea_ids}
        for nid, pid, slug in (await session.execute(
            select(FamilyIdeaPlatform.note_id, Platform.id, Platform.slug)
            .join(Platform, Platform.id == FamilyIdeaPlatform.platform_id)
            .where(FamilyIdeaPlatform.note_id.in_(idea_ids))
            .order_by(Platform.order_index.asc(), Platform.slug.asc())
        )).all():
            idea_platforms[nid][pid] = slug
        platform_ids = {pid for m in idea_platforms.values() for pid in m}
        membership: dict[int, set[int]] = {}
        for pid, plat in (await session.execute(
            select(ProjectPlatform.project_id, ProjectPlatform.platform_id)
            .where(ProjectPlatform.platform_id.in_(platform_ids),
                   ProjectPlatform.state.in_(family_svc.MEMBER_STATES))
        )).all():
            membership.setdefault(pid, set()).add(plat)
        rows = (await session.execute(
            select(FamilyAdoption).where(FamilyAdoption.idea_id.in_(idea_ids))
        )).scalars().all()
        project_ids = set(membership) | {r.project_id for r in rows}
        if project_id:
            project_ids &= {project_id}
        projects = (await session.execute(
            select(Project.id, Project.title)
            .where(Project.id.in_(project_ids), Project.deleted_at.is_(None))
            .order_by(Project.title.asc())
        )).all()
        task_ids = [r.owed_task_id for r in rows if r.owed_task_id]
        tasks = {
            t.id: t for t in (await session.execute(
                select(Note).where(Note.id.in_(task_ids), Note.deleted_at.is_(None))
            )).scalars().all()
        } if task_ids else {}
        # The code that answers each cell (step 5): live shapes classified
        # against one of the idea's references.
        evidence: dict[tuple[int, int], list[dict]] = {}
        for iid, shape in (await session.execute(
            select(FamilyIdeaReference.idea_id, CodeShape)
            .join(CodeShape, CodeShape.snippet_id == FamilyIdeaReference.snippet_id)
            .where(FamilyIdeaReference.idea_id.in_(idea_ids),
                   CodeShape.project_id.in_(list(project_ids)),
                   CodeShape.vanished_at.is_(None),
                   CodeShape.status.in_(_SHAPE_ADOPTS + ("variant",)))
            .order_by(CodeShape.path, CodeShape.symbol)
        )).all() if idea_ids and project_ids else []:
            evidence.setdefault((shape.project_id, iid), []).append(
                {"path": shape.path, "symbol": shape.symbol, "status": shape.status})

    readable = [(pid, title) for pid, title in projects
                if await access.can_read_project(user_id, pid)]
    by_pair = {(r.project_id, r.idea_id): r for r in rows}
    idea_meta = {i.note_id: (i, title) for i, title, *_ in ideas}
    cells = []
    for pid, ptitle in readable:
        for iid in idea_ids:
            idea, ititle = idea_meta[iid]
            row = by_pair.get((pid, iid))
            reached = bool(membership.get(pid, set()) & set(idea_platforms[iid]))
            if row is None and not reached:
                continue
            cell = {
                "project_id": pid, "project_title": ptitle,
                "idea_id": iid, "idea_title": ititle,
                "idea_version": idea.canon_version,
                "in_scope": reached,
                "reached": row is not None,
                "status": row.status if row else "unassessed",
                "reason": (row.reason or "") if row else "",
                "canon_version": row.canon_version if row else None,
                "assessed_at": row.to_dict()["assessed_at"] if row else None,
                "decided_via": row.decided_via if row else None,
                "needs_recheck": bool(row) and needs_recheck(
                    row.status, row.canon_version, idea.status, idea.canon_version),
                "owed_task": None,
                "shapes": evidence.get((pid, iid), []),
            }
            task = tasks.get(row.owed_task_id) if row and row.owed_task_id else None
            if task is not None:
                cell["owed_task"] = {"id": task.id, "title": task.title, "status": task.status}
            cells.append(cell)

    seen_projects = {c["project_id"] for c in cells}
    return {
        "ideas": [
            {
                "note_id": i.note_id, "title": title, "note_type": note_type or "note",
                "is_task": note_status is not None, "canon_version": i.canon_version,
                "applies_when": i.applies_when or "",
                "platforms": list(idea_platforms[i.note_id].values()),
            }
            for i, title, note_type, note_status in ideas
        ],
        "projects": [
            {"id": pid, "title": title,
             "platforms": sorted({s for iid in idea_ids
                                  for p, s in idea_platforms[iid].items()
                                  if p in membership.get(pid, set())})}
            for pid, title in readable if pid in seen_projects
        ],
        "cells": cells,
        "outcomes": list(OUTCOMES),
    }


async def list_adoptions(
    user_id: int, *, project_id: int | None = None, idea_id: int | None = None,
    status: str | None = None, recheck_only: bool = False, platform: str | None = None,
) -> list[dict]:
    """The ledger as rows — the matrix's cells, filtered."""
    matrix = await adoption_matrix(user_id, platform=platform, project_id=project_id)
    return [
        c for c in matrix["cells"]
        if (not idea_id or c["idea_id"] == idea_id)
        and (not status or c["status"] == status)
        and (not recheck_only or c["needs_recheck"])
    ]


async def get_adoption(user_id: int, project_id: int, idea_id: int) -> dict:
    """Everything one assessment reads: the idea, this project's current
    answer, the four outcomes, the precedents, and the reference
    implementations with this project's languages."""
    if not await access.can_read_project(user_id, project_id):
        raise ValueError(f"project {project_id} not found")
    idea = await family_svc.get_idea(user_id, idea_id)
    if idea is None:
        raise ValueError(f"#{idea_id} is not a family idea you can read")
    rows = await list_adoptions(user_id, project_id=project_id, idea_id=idea_id)
    async with async_session() as session:
        refs = await _references(session, idea_id)
        langs = await _project_languages(session, project_id)
    idea.pop("decisions", None)
    return {
        "idea": idea,
        "adoption": rows[0] if rows else None,
        "outcomes": list(OUTCOMES),
        "precedents": await assessment_precedents(user_id, project_id, idea_id),
        "references": refs,
        "project_languages": langs,
    }


# --- the owed task -----------------------------------------------------------------

def _reference_line(refs: list[dict], langs: list[str], idea_id: int) -> str:
    def fmt(r):
        return f"#{r['id']} “{r['title']}”" + (f" ({r['language']})" if r["language"] else "")

    if not refs:
        return (f"No reference implementation is recorded yet — build from the idea's "
                f"note, #{idea_id}.")
    mine = [r for r in refs if r["language"] and r["language"] in langs]
    if mine:
        return "; ".join(fmt(r) for r in mine)
    said = ", ".join(langs) if langs else "not yet known"
    return (f"None is recorded in this project's language ({said}). The idea is what "
            f"transfers; the ones that exist: {'; '.join(fmt(r) for r in refs)}.")


async def _matching_systems(session, project_id: int, note_ids: list[int]) -> list[int]:
    """The project's Systems matching the idea's: the same canonical area as
    one the idea (or a reference) is filed under, or that very System when
    the idea was written in this project."""
    tagged = (await session.execute(
        select(System.id, System.project_id, System.canonical_id)
        .join(RecordSystem, RecordSystem.system_id == System.id)
        .where(RecordSystem.note_id.in_(note_ids), System.deleted_at.is_(None))
    )).all()
    direct = [sid for sid, pid, _ in tagged if pid == project_id]
    canonical = {cid for _, _, cid in tagged if cid is not None}
    mapped = (await session.execute(
        select(System.id).where(
            System.project_id == project_id, System.canonical_id.in_(canonical),
            System.deleted_at.is_(None))
        .order_by(System.order_index.asc(), System.id.asc())
    )).scalars().all() if canonical else []
    return list(dict.fromkeys(direct + list(mapped)))


async def _file_owed_task(user_id: int, row_id: int, reason: str) -> Note:
    from scribe.services import notes as notes_svc
    from scribe.services import systems as systems_svc

    async with async_session() as session:
        row = await session.get(FamilyAdoption, row_id)
        idea = await session.get(FamilyIdea, row.idea_id)
        note = await session.get(Note, row.idea_id)
        refs = await _references(session, row.idea_id)
        langs = await _project_languages(session, row.project_id)
        platforms = await family_svc._platform_slugs(session, row.idea_id)
        systems = await _matching_systems(
            session, row.project_id, [row.idea_id] + [r["id"] for r in refs])
        project_id, idea_id, version = row.project_id, row.idea_id, idea.canon_version
    body = (
        f"Family idea #{idea_id} “{note.title}” is canon for "
        f"{', '.join(platforms) or 'its platforms'}, applies to this project, and is "
        f"not done here yet.\n\n"
        f"**When it applies:** {idea.applies_when or '—'}\n\n"
        f"**The gap:** {reason.strip()}\n\n"
        f"**Start from:** {_reference_line(refs, langs, idea_id)}\n\n"
        "Build the idea in this project's own language and conventions: what the "
        "family shares is the idea, its traps and its checklist, not the code. When "
        "it is done, assess it again as `adopted` (assess_family_adoption) with where "
        "it lives as evidence; that closes this task. If it turns out not to apply, "
        "or this project has a reason to depart that is a fact about itself, assess "
        "it `exempt` or `variant` instead.\n\n"
        f"_Filed by the family adoption ledger against canon version {version}._"
    )
    task = await notes_svc.create_note(
        user_id, title=f"Adopt the family idea “{note.title}”"[:500], body=body,
        project_id=project_id, status=TaskStatus.todo.value, task_kind="work",
    )
    if systems:
        await systems_svc.set_record_systems(user_id, task.id, systems)
    return task


async def _set_task_status(user_id: int, task: Note, status: str, log: str) -> None:
    from scribe.services import notes as notes_svc
    from scribe.services import task_logs

    # The writer is checked; the write goes through the owner's door, which
    # is the one that keeps versions, claims and the embedding in step.
    await notes_svc.update_note(task.user_id, task.id, status=status)
    await task_logs.create_log(user_id, task.id, log)


async def _sync_owed_task(user_id: int, row_id: int, *, reason: str) -> dict | None:
    """Make the owed task agree with the answer. Idempotent: an `owed` row
    with an open task, or a settled row with a closed one, is left alone.
    A task the caller cannot write is left alone too, and said so."""
    async with async_session() as session:
        row = await session.get(FamilyAdoption, row_id)
        task = await session.get(Note, row.owed_task_id) if row.owed_task_id else None
        if task is not None and task.deleted_at is not None:
            task = None
        status, idea_id = row.status, row.idea_id

    def ref(t: Note, s: str, action: str | None = None) -> dict:
        out = {"id": t.id, "title": t.title, "status": s}
        if action:
            out["action"] = action
        return out

    if status == "owed":
        if task is not None and task.status in _OPEN_TASK:
            return ref(task, task.status)
        if task is not None and await access.can_write_note(user_id, task.id):
            await _set_task_status(
                user_id, task, TaskStatus.todo.value,
                f"Reopened: family idea #{idea_id} was assessed owed again — {reason}")
            return ref(task, TaskStatus.todo.value, "reopened")
        task = await _file_owed_task(user_id, row_id, reason)
        async with async_session() as session:
            row = await session.get(FamilyAdoption, row_id)
            row.owed_task_id = task.id
            await session.commit()
        return ref(task, task.status, "filed")

    if task is None:
        return None
    if task.status not in _OPEN_TASK:
        return ref(task, task.status)
    if not await access.can_write_note(user_id, task.id):
        return ref(task, task.status, "left open — no write access to the task")
    new = TaskStatus.done.value if status == "adopted" else TaskStatus.cancelled.value
    if status == "unassessed":
        line = f"Family idea #{idea_id}'s owed answer was undone — {reason}"
    else:
        line = f"Family idea #{idea_id} was assessed {status} in this project — {reason}"
    await _set_task_status(user_id, task, new, line)
    return ref(task, new, "closed")


# --- writes -------------------------------------------------------------------------

async def _row_for_write(session, project_id: int, idea_id: int) -> FamilyAdoption:
    row = await _row(session, project_id, idea_id)
    if row is not None:
        return row
    if not await _is_member(session, project_id, idea_id):
        raise ValueError(
            f"project {project_id} is not on any of #{idea_id}'s platforms, so the idea "
            "does not reach it — answer the project's platforms first if it should")
    row = FamilyAdoption(project_id=project_id, idea_id=idea_id, status="unassessed")
    session.add(row)
    await session.flush()
    return row


def _answer(row: FamilyAdoption, *, status: str, reason: str, version: int,
            decided_via: str) -> None:
    row.status = status
    row.reason = reason.strip() or None
    row.canon_version = version
    row.assessed_at = family_svc._now()
    row.decided_via = decided_via
    row.updated_at = family_svc._now()


async def assess(
    user_id: int, project_id: int, idea_id: int, *, outcome: str, reason: str,
    evidence: list[str] | None = None, precedent_ids: list[int] | None = None,
    decided_via: str = "agent", source: str = "", shapes: list[dict] | None = None,
) -> dict:
    """Record one project's answer to one canon idea.

    Raises ValueError before writing anything when the answer is malformed
    (assessment_problems), the caller cannot write the project, the idea
    is not canon or does not reach the project, or the project's shape
    ledger already answers the idea otherwise (the two never disagree).

    ``source``/``shapes`` are the shape ledger's own door (sync_from_shapes):
    the decision's evidence then names the shapes it was read from.

    The same answer given again — same outcome, reason and canon version —
    records nothing (`changed: False`); it still makes sure an owed answer
    has an open task. The response carries the precedents consulted, and
    `owed_task` when there is one.
    """
    evidence = _clean(evidence)
    problems = assessment_problems(outcome=outcome, reason=reason, evidence=evidence)
    if problems:
        raise ValueError("; ".join(problems))
    if not await access.can_write_project(user_id, project_id):
        raise ValueError(f"project {project_id} not found or no write access")
    if not await access.can_read_note(user_id, idea_id):
        raise ValueError(f"#{idea_id} is not a family idea you can read")
    consulted = await assessment_precedents(user_id, project_id, idea_id)
    named = await family_svc._existing_decision_ids(precedent_ids or [])
    precedent_list = list(dict.fromkeys(named + [p["id"] for p in consulted]))

    async with async_session() as session:
        idea = await session.get(FamilyIdea, idea_id)
        if idea is None or idea.status != "canon":
            raise ValueError(f"#{idea_id} is not family canon — only canon is assessed")
        row = await _row_for_write(session, project_id, idea_id)
        if source != SHAPE_SOURCE:
            refused = _contradiction(
                outcome, shapes_say(await _shape_evidence(session, project_id, idea_id)), idea_id)
            if refused:
                raise ValueError(refused)
        before = _row_state(row)
        after = {"status": outcome, "reason": reason.strip(), "canon_version": idea.canon_version}
        changed = before != after
        decision = None
        if changed:
            _answer(row, status=outcome, reason=reason, version=idea.canon_version,
                    decided_via=decided_via)
            decision = family_svc._log(
                session, idea_id=idea_id, project_id=project_id, action="assess",
                reason=reason, before=before, after=_row_state(row),
                evidence=({"evidence": evidence, "source": source, "shapes": shapes or []}
                          if source else {"evidence": evidence}),
                precedent_ids=precedent_list, decided_via=decided_via, user_id=user_id,
            )
        await session.commit()
        if decision is not None:
            await session.refresh(decision)
        row_id = row.id
    task = await _sync_owed_task(user_id, row_id, reason=reason)
    rows = await list_adoptions(user_id, project_id=project_id, idea_id=idea_id)
    return {
        "changed": changed,
        "adoption": rows[0] if rows else None,
        "decision": decision.to_dict() if decision is not None else None,
        "precedents": consulted,
        "owed_task": task,
    }


async def undo_assessment(user_id: int, target: FamilyDecision, *, reason: str,
                          decided_via: str = "agent") -> dict:
    """Reverse one project's answer, restoring the row it recorded as
    `before`. Only the latest decision on that (idea, project) answer can be
    undone. The owed task follows the restored answer."""
    if not await access.can_write_project(user_id, target.project_id):
        raise ValueError(f"decision {target.id} not found or no write access")
    async with async_session() as session:
        latest = (await family_svc._latest_pair_decisions(
            session, {(target.idea_id, target.project_id)})).get((target.idea_id, target.project_id))
        if latest != target.id:
            raise ValueError(
                f"decision {target.id} cannot be undone: decision {latest} came after it — "
                "undo that one first")
        if not family_svc._can_undo(target):
            raise ValueError(f"decision {target.id} cannot be undone: it changed nothing")
        row = await _row(session, target.project_id, target.idea_id)
        if row is None:
            raise ValueError(f"decision {target.id} cannot be undone: the answer it changed is gone")
        before = _row_state(row)
        prior = target.before or {"status": "unassessed", "reason": "", "canon_version": None}
        refused = _contradiction(
            prior["status"], shapes_say(await _shape_evidence(session, target.project_id, target.idea_id)),
            target.idea_id)
        if refused:
            raise ValueError(f"decision {target.id} cannot be undone: {refused}")
        row.status = prior["status"]
        row.reason = (prior.get("reason") or "").strip() or None
        row.canon_version = prior.get("canon_version")
        if row.status == "unassessed":
            row.assessed_at = None
            row.decided_via = None
        else:
            row.assessed_at = family_svc._now()
            row.decided_via = decided_via
        row.updated_at = family_svc._now()
        decision = family_svc._log(
            session, idea_id=target.idea_id, project_id=target.project_id, action="undo",
            reason=reason, before=before, after=_row_state(row),
            evidence={"undid_action": target.action}, precedent_ids=[target.id],
            decided_via=decided_via, user_id=user_id,
        )
        await session.commit()
        await session.refresh(decision)
        row_id = row.id
    task = await _sync_owed_task(user_id, row_id, reason=f"undo of decision {target.id}: {reason}")
    async with async_session() as session:
        idea = await session.get(FamilyIdea, target.idea_id)
        row = await session.get(FamilyAdoption, row_id)
        adoption = row.to_dict()
    return {"idea": idea.to_dict(), "adoption": adoption,
            "decision": decision.to_dict(), "owed_task": task}


async def set_references(user_id: int, idea_id: int, snippet_ids: list[int]) -> list[dict]:
    """Replace an idea's reference implementations — the snippets an owed
    task points a project at, one per language. Each must be a snippet the
    caller can read; the idea must be theirs to write."""
    if not await access.can_write_note(user_id, idea_id):
        raise ValueError(f"note {idea_id} not found or no write access")
    wanted = list(dict.fromkeys(int(i) for i in snippet_ids or [] if i))
    async with async_session() as session:
        if await session.get(FamilyIdea, idea_id) is None:
            raise ValueError(f"#{idea_id} is not a family idea")
        found = {n.id: n for n in (await session.execute(
            select(Note).where(Note.id.in_(wanted), Note.deleted_at.is_(None))
        )).scalars().all()} if wanted else {}
        bad = [i for i in wanted
               if i not in found or (found[i].note_type or "") != "snippet"]
        if bad:
            raise ValueError(f"not a snippet: {', '.join(map(str, bad))}")
        for i in wanted:
            if not await access.can_read_note(user_id, i):
                raise ValueError(f"not a snippet: {i}")
        current = set((await session.execute(
            select(FamilyIdeaReference.snippet_id).where(FamilyIdeaReference.idea_id == idea_id)
        )).scalars().all())
        for sid in current - set(wanted):
            await session.delete(await session.get(FamilyIdeaReference, (idea_id, sid)))
        for sid in wanted:
            if sid not in current:
                session.add(FamilyIdeaReference(idea_id=idea_id, snippet_id=sid))
        await session.commit()
        return await _references(session, idea_id)


def _fold_section(*, ground: str, fold: str, conditions: dict | None,
                  canon_title: str, other_title: str, when: str) -> str:
    g = _GROUND[ground]
    if g["fold_as"] == "condition":
        conds = conditions or {}
        return (
            f"### When {conds['other'].strip()} — {other_title}'s approach\n\n"
            f"{fold.strip()}\n\n"
            f"When {conds['canon'].strip()}, {canon_title}'s approach above is the canon. "
            f"_(Family conflict, {when}: {g['title'].lower()}.)_"
        )
    heading = "Trap" if g["fold_as"] == "trap" else "Alternative"
    return (
        f"### {heading} — {other_title}'s approach\n\n{fold.strip()}\n\n"
        f"_(Family conflict, {when}: decided by {g['title'].lower()}; "
        f"{canon_title}'s approach is the canon.)_"
    )


async def resolve_conflict(
    user_id: int, idea_id: int, *, canon_project_id: int, other_project_id: int,
    ground: str, grounds_checked: dict | None, fold: str, evidence: list[str] | None,
    reason: str, conditions: dict | None = None, precedent_ids: list[int] | None = None,
    decided_via: str = "agent",
) -> dict:
    """Settle two projects that solve the same canon idea differently, by
    the conflict order.

    `canon_project_id` is the side whose approach the idea's note states
    after this — if that is not what the note says now, rewrite the note
    first (update_note); the note is the canon. `other_project_id` is the
    side that loses, or in a split, the side whose condition is the branch.

    Effects, in order: the losing side's reasoning (`fold`) is appended to
    the idea's note — as a trap, an alternative, or the branch for its
    condition; the version moves (a `revise` decision carrying the ground and
    the grounds checked above it); the canon side is answered `adopted`, and
    the other side `owed` (with a task filed) or, in a split, `adopted` under
    its condition. Each answer is its own `assess` decision, naming the
    revision as its precedent. Last, either side's shapes classified as a
    variant of the idea are re-judged to agree with its new answer
    (`shapes_rejudged`), so the shape ledger and this one do not disagree.
    """
    from scribe.services import notes as notes_svc

    evidence = _clean(evidence)
    problems = conflict_problems(ground=ground, grounds_checked=grounds_checked,
                                 evidence=evidence, conditions=conditions, fold=fold)
    if canon_project_id == other_project_id:
        problems.append("a conflict is between two different projects")
    if not (reason or "").strip():
        problems.append("a resolution needs a reason")
    if problems:
        raise ValueError("; ".join(problems))
    if not await access.can_write_note(user_id, idea_id):
        raise ValueError(f"note {idea_id} not found or no write access")
    for pid in (canon_project_id, other_project_id):
        if not await access.can_write_project(user_id, pid):
            raise ValueError(f"project {pid} not found or no write access")
    named = await family_svc._existing_decision_ids(precedent_ids or [])
    consulted = await family_svc.precedents(user_id, idea_id)
    precedent_list = list(dict.fromkeys(named + [p["id"] for p in consulted]))

    async with async_session() as session:
        idea = await session.get(FamilyIdea, idea_id)
        if idea is None or idea.status != "canon":
            raise ValueError(f"#{idea_id} is not family canon — only canon has conflicts")
        for pid in (canon_project_id, other_project_id):
            if await _row(session, pid, idea_id) is None and not await _is_member(session, pid, idea_id):
                raise ValueError(f"#{idea_id} does not reach project {pid}")
        titles = dict((await session.execute(
            select(Project.id, Project.title)
            .where(Project.id.in_([canon_project_id, other_project_id]))
        )).all())
        note = await session.get(Note, idea_id)
        owner, body = note.user_id, note.body or ""

    # The losing side's reasoning goes into the canon FIRST: whatever fails
    # after this, the reasoning is not dropped.
    section = _fold_section(
        ground=ground, fold=fold, conditions=conditions,
        canon_title=titles[canon_project_id], other_title=titles[other_project_id],
        when=family_svc._now().date().isoformat(),
    )
    await notes_svc.update_note(owner, idea_id, body=f"{body.rstrip()}\n\n{section}\n")

    split = ground == "split_by_condition"
    other_outcome = "adopted" if split else "owed"
    g = _GROUND[ground]
    async with async_session() as session:
        idea = await session.get(FamilyIdea, idea_id)
        before = await family_svc._snapshot(session, idea)
        idea.canon_version = await family_svc.next_version(session, idea)
        idea.updated_at = family_svc._now()
        await session.flush()
        revision = family_svc._log(
            session, idea_id=idea_id, action="revise", reason=reason, before=before,
            after=await family_svc._snapshot(session, idea),
            evidence={
                "conflict": {
                    "ground": ground,
                    "grounds_checked": {k: str((grounds_checked or {}).get(k) or "").strip()
                                        for k in CONFLICT_KEYS[:CONFLICT_KEYS.index(ground)]},
                    "canon": titles[canon_project_id],
                    "other": titles[other_project_id],
                    "folded_as": g["fold_as"],
                    "conditions": conditions if split else None,
                },
                "evidence": evidence,
            },
            precedent_ids=precedent_list, decided_via=decided_via, user_id=user_id,
        )
        await session.flush()
        row_ids = {}
        answers = (
            (canon_project_id, "adopted",
             f"Canon after a family conflict ({g['title'].lower()}): {reason.strip()}"),
            (other_project_id, other_outcome,
             (f"Canon where {conditions['other'].strip()} (split by condition): {reason.strip()}"
              if split else
              f"Lost a family conflict ({g['title'].lower()}) — adopt the canon: {reason.strip()}")),
        )
        for pid, outcome, why in answers:
            row = await _row_for_write(session, pid, idea_id)
            row_before = _row_state(row)
            _answer(row, status=outcome, reason=why, version=idea.canon_version,
                    decided_via=decided_via)
            family_svc._log(
                session, idea_id=idea_id, project_id=pid, action="assess", reason=why,
                before=row_before, after=_row_state(row), evidence={"evidence": evidence},
                precedent_ids=[revision.id], decided_via=decided_via, user_id=user_id,
            )
            row_ids[pid] = row.id
        await session.commit()
        await session.refresh(revision)
    # The shapes follow the resolution, so the two ledgers still agree.
    rejudged = {pid: await _shapes_follow(user_id, pid, idea_id, outcome, why)
                for pid, outcome, why in answers}
    task = await _sync_owed_task(user_id, row_ids[other_project_id], reason=answers[1][2])
    await _sync_owed_task(user_id, row_ids[canon_project_id], reason=answers[0][2])
    return {
        "idea": idea.to_dict(),
        "decision": revision.to_dict(),
        "folded_as": g["fold_as"],
        "adoptions": await list_adoptions(user_id, idea_id=idea_id),
        "owed_task": task,
        "shapes_rejudged": {str(pid): n for pid, n in rejudged.items() if n},
    }


async def _shapes_follow(user_id: int, project_id: int, idea_id: int, outcome: str,
                         why: str) -> int:
    """Re-judge a project's variant shapes of the idea to agree with the
    answer a conflict resolution just gave it: `adopted` makes them instances
    of the reference they departed from (in a split, their condition is now
    canon); `owed` returns them to the shape todo, to be rebuilt to the
    canon the owed task names. Returns how many rows were re-judged."""
    from scribe.services import shape_ledger

    async with async_session() as session:
        rows = [r for r in await _shape_evidence(session, project_id, idea_id)
                if r.status == "variant"]
    if not rows:
        return 0
    status = "instance" if outcome == "adopted" else "unclassified"
    result = await shape_ledger.classify_shapes(user_id, project_id, [
        {"path": r.path, "symbol": r.symbol, "kind": r.kind, "status": status,
         "snippet_id": r.snippet_id, "reason": why}
        for r in rows
    ], via="agent")
    return int(result.get("classified") or 0)


# --- delivery (milestone 463 step 6) -------------------------------------------------
#
# The ledger reaches a session three ways: a count on entering a project, the
# ideas themselves by retrieval, and the owed answers filed while a task was
# open, handed back when it closes.

def family_line(project_id: int, cells: list[dict]) -> dict | None:
    """The `family` line enter_project carries: what this project has not
    answered, owes, or answered against an older canon version. None when
    all three are zero — the key is attached only when it asks something.

    Shown on EVERY entry (the step settled this), not only when a session
    touches a platform: entering is when the agent chooses what to work on,
    an owed task already appears in open_tasks, and an unanswered idea has
    nowhere else to be seen. Counts only, each with the call that lists it."""
    unassessed = sum(1 for c in cells if c["in_scope"] and c["status"] == "unassessed")
    owed = sum(1 for c in cells if c["status"] == "owed")
    recheck = sum(1 for c in cells if c["needs_recheck"])
    if not (unassessed or owed or recheck):
        return None
    parts, calls = [], {}
    if unassessed:
        parts.append(f"{unassessed} unassessed")
        calls["unassessed"] = f'list_family_adoptions(project_id={project_id}, status="unassessed")'
    if owed:
        parts.append(f"{owed} owed")
        calls["owed"] = f'list_family_adoptions(project_id={project_id}, status="owed")'
    if recheck:
        parts.append(f"{recheck} to recheck")
        calls["needs_recheck"] = f"list_family_adoptions(project_id={project_id}, needs_recheck=true)"
    return {
        "line": "family canon on this project's platforms: " + " · ".join(parts),
        "unassessed": unassessed, "owed": owed, "needs_recheck": recheck,
        "calls": calls,
        "answer_with": ("get_family_adoption(project_id, idea_id) reads one with its "
                        "precedents; assess_family_adoption answers it — or classify_shapes "
                        "the code against it (idea_id=…). The family-canon skill has the order."),
    }


async def family_readout(user_id: int, project_id: int) -> dict | None:
    matrix = await adoption_matrix(user_id, project_id=project_id)
    return family_line(project_id, matrix["cells"])


async def family_reach_ids(project_id: int, session=None) -> set[int]:
    """The family records a project's retrieval reaches beyond its own: every
    canon idea on a platform the project is a member of, and each idea's
    reference implementations in the project's languages — all of them when
    none is in those languages, so a cross-language idea still brings its
    reference. Readability is the search's own scope, not decided here."""
    if session is None:
        async with async_session() as own:
            return await family_reach_ids(project_id, own)
    ideas = set((await session.execute(
        select(FamilyIdea.note_id)
        .join(FamilyIdeaPlatform, FamilyIdeaPlatform.note_id == FamilyIdea.note_id)
        .join(ProjectPlatform, ProjectPlatform.platform_id == FamilyIdeaPlatform.platform_id)
        .where(
            FamilyIdea.status == "canon",
            ProjectPlatform.project_id == project_id,
            ProjectPlatform.state.in_(family_svc.MEMBER_STATES),
        )
    )).scalars().all())
    if not ideas:
        return set()
    langs = set(await _project_languages(session, project_id))
    refs: dict[int, list[tuple[int, str]]] = {}
    for iid, nid, lang in (await session.execute(
        select(FamilyIdeaReference.idea_id, Note.id, func.lower(Note.data["language"].astext))
        .join(Note, Note.id == FamilyIdeaReference.snippet_id)
        .where(FamilyIdeaReference.idea_id.in_(ideas), Note.deleted_at.is_(None))
    )).all():
        refs.setdefault(iid, []).append((nid, lang or ""))
    reach = set(ideas)
    for found in refs.values():
        local = [nid for nid, lang in found if lang in langs]
        reach.update(local or [nid for nid, _ in found])
    return reach


async def owed_since(user_id: int, since) -> list[dict]:
    """The owed answers this user recorded since ``since`` that are still
    owed — work filed into a project (often another one) that the report
    closing this task should name."""
    if since is None:
        return []
    async with async_session() as session:
        rows = (await session.execute(
            select(FamilyAdoption, Note.title, Project.title)
            .join(FamilyDecision, (FamilyDecision.idea_id == FamilyAdoption.idea_id)
                  & (FamilyDecision.project_id == FamilyAdoption.project_id))
            .join(Note, Note.id == FamilyAdoption.idea_id)
            .join(Project, Project.id == FamilyAdoption.project_id)
            .where(
                FamilyDecision.user_id == user_id,
                FamilyDecision.created_at >= since,
                FamilyDecision.after["status"].astext == "owed",
                FamilyAdoption.status == "owed",
            )
            .distinct()
            .order_by(Project.title, Note.title)
        )).all()
    return [
        {"idea_id": row.idea_id, "idea_title": idea_title,
         "project_id": row.project_id, "project_title": project_title,
         "owed_task_id": row.owed_task_id}
        for row, idea_title, project_title in rows
        if await access.can_read_project(user_id, row.project_id)
    ]


OWED_CUE = ("Owed family adoptions were filed while this task was open (`family_owed`) — "
            "name each in the report: it is work now waiting in that project.")


async def attach_owed_adoptions(user_id: int, data: dict, note) -> None:
    """Ride the owed answers filed since the task started on its closing
    response — fail-open: a decoration never breaks the close it rides on."""
    try:
        owed = await owed_since(user_id, getattr(note, "started_at", None)
                                or getattr(note, "created_at", None))
        if owed:
            data["family_owed"] = owed
            if "report_back" in data:
                data["report_back"] = f"{data['report_back']} {OWED_CUE}"
    except Exception:
        logger.warning("owed adoptions unreadable for task %s", getattr(note, "id", None),
                       exc_info=True)