"""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.
"""
from __future__ import annotations

import logging
from sqlalchemy import func, select

from scribe.models import async_session
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]


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 {}

    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,
            }
            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,
                          system_ids: list[int] | None) -> 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 = system_ids or 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,
                          system_ids: list[int] | None = None) -> 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, system_ids)
        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,
    system_ids: list[int] | None = None, decided_via: str = "agent",
) -> 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, or the idea
    is not canon or does not reach the project.

    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)
        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}, 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, system_ids=system_ids)
    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}
        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.
    """
    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)
    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,
    }