"""The shape ledger's write side — sync and mechanical marking.

The coverage walk (services/coverage.py) is the only feed that sees every
shape, so it is the ledger's sync point: each refresh upserts one repo's
extracted shapes — new shapes arrive `unclassified` (THE todo state, note
2786), surviving shapes bump their last-seen marker, vanished shapes get
stamped rather than deleted (history is the point). Classifications survive
recompute by construction: the upsert never touches a judgment, with two
deliberate exceptions —

  - a judgment whose snippet target is gone (SET NULL on snippet deletion)
    is re-filed as unclassified so it rejoins the todo instead of dangling;
  - a MECHANICALLY-stamped canonical row whose snippet location no longer
    covers it falls back to unclassified. Only mechanical stamps self-heal;
    an agent's judgment is never unwound by machinery.

`location_covers` is the one covering predicate — the same doctrine the
recorded-location drift check uses — shared by the sync's canonical marking
and by anything else that must decide whether a recorded location speaks for
an extracted shape.
"""
from __future__ import annotations

import difflib
import logging
import posixpath
import re
from datetime import datetime, timedelta, timezone
from typing import Iterable, Mapping, NamedTuple

from sqlalchemy import select

from grimoire.models import async_session
from grimoire.models.code_shape import (
    REASON_CODES, CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse,
)
from grimoire.models.base import iso

logger = logging.getLogger(__name__)

# Statuses whose meaning requires a snippet target.
_NEEDS_TARGET = ("canonical", "instance", "variant")


def _norm_symbol(kind_or_symbol: str) -> str:
    # CSS shapes and recorded CSS symbols may or may not carry the leading
    # dot; compare without it so ".btn-primary" and "btn-primary" agree.
    return kind_or_symbol.lstrip(".").strip()


def location_covers(loc_path: str, loc_symbol: str, path: str, name: str) -> bool:
    """Does a recorded (path, symbol) location speak for this shape?

    Symbol-less locations never cover a shape — a whole-file record makes no
    claim about any particular definition inside it. Path semantics are the
    drift check's own: exact file, or the recorded path is a directory the
    file lives under.
    """
    if not (loc_symbol or "").strip():
        return False
    if _norm_symbol(loc_symbol) != _norm_symbol(name):
        return False
    if not loc_path:
        # Symbol-only record: the symbol match is all the claim there is.
        return True
    from grimoire.services.snippets import _path_touches

    return _path_touches(loc_path, path)


# The rows the machine may still speak about: nobody's judgment stands on
# them. `scoped` is the sync's own by-construction stamp — the
# proposer, derive grouping, divergence, hook evidence and sweeps treat it
# like the todo; only the human todo (`unclassified`) excludes it.
_MECHANICAL_TODO = ("unclassified", "scoped")
_SCOPED_REASON = (
    "by construction: a Vue component's scoped <style> rule / <script setup> "
    "function — unreachable from any other file (stamped by the coverage sync)"
)


async def sync_repo_shapes(
    project_id: int,
    repo_key: str,
    shapes: list,
    *,
    seen_marker: str,
) -> None:
    """Upsert one repo's extracted shapes into the ledger.

    ``shapes`` are (path, kind, name) triples, or the richer ArchiveShape
    records whose 4th/5th fields — signature, body_sha — refresh the
    row's content fingerprint, and whose 7th says the shape is a
    one-off by construction: such rows are stamped `scoped` (mechanical)
    while unjudged, and un-stamped if a later tree makes them reachable. ``seen_marker`` is the commit the archive was
    read at when the forge can say, else the ref name — provenance sugar;
    the row timestamps carry the when.
    """
    now = datetime.now(timezone.utc)
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.repo_key == repo_key,
                )
            )
        ).scalars().all()
        by_key = {(r.path, r.symbol, r.kind): r for r in rows}
        seen: set[tuple[str, str, str]] = set()
        for shape in shapes:
            path, kind, name = shape[0], shape[1], shape[2]
            signature = shape[3] if len(shape) > 3 else ""
            body_sha = shape[4] if len(shape) > 4 else ""
            scoped = bool(shape[6]) if len(shape) > 6 else False
            key = (path, name, kind)
            if key in seen:
                continue
            seen.add(key)
            row = by_key.get(key)
            if row is None:
                row = CodeShape(
                    project_id=project_id, repo_key=repo_key,
                    path=path, symbol=name, kind=kind,
                    first_seen_commit=seen_marker, last_seen_commit=seen_marker,
                    signature=signature, body_sha=body_sha,
                )
                if scoped:
                    await _judge(session, row, status="scoped", snippet_id=None,
                                 by="mechanical", reason=_SCOPED_REASON, at=now)
                else:
                    session.add(row)
                continue
            row.last_seen_commit = seen_marker
            if scoped and row.status == "unclassified":
                await _judge(session, row, status="scoped", snippet_id=None,
                             by="mechanical", reason=_SCOPED_REASON, at=now)
            elif not scoped and row.status == "scoped":
                await _judge(session, row, status="unclassified", snippet_id=None,
                             by=None, reason=None, at=now)
            if signature:
                row.signature = signature
            if body_sha and body_sha != row.body_sha:
                judged = row.status in ("instance", "variant")
                if judged and not row.classified_sha:
                    # Judged before fingerprints existed: the first sync
                    # that sees a body adopts it as the judged content.
                    row.classified_sha = body_sha
                elif judged and row.classified_sha != body_sha and row.recheck_at is None:
                    # The body moved under a standing judgment: the judgment
                    # stands, but asks to be confirmed again.
                    row.recheck_at = now
                    session.add(_event(row, "drifted", now, commit=seen_marker))
                row.body_sha = body_sha
            elif row.status in ("instance", "variant") and not row.classified_sha:
                row.classified_sha = row.body_sha
            # A shape that vanished and came back is live again — the vanish
            # stays visible in history via updated_at, not as a dead flag.
            if row.vanished_at is not None:
                row.vanished_at = None
                session.add(_event(row, "reappeared", now, commit=seen_marker))
            if row.status in _NEEDS_TARGET and row.snippet_id is None:
                row.status = "unclassified"
                row.classified_by = None
                row.classified_at = None
                row.reason = None
        for key, row in by_key.items():
            if key not in seen and row.vanished_at is None:
                row.vanished_at = now
                session.add(_event(row, "vanished", now, commit=row.last_seen_commit))
        await session.commit()


async def move_shapes(
    user_id: int, project_id: int, from_path: str, to_path: str,
) -> dict:
    """Carry the ledger rows under ``from_path`` (a file or a directory) to
    the same place under ``to_path``, in every repo of the project.

    Rows are keyed by path, so a directory moved in the tree would otherwise
    vanish every judged row at the next sync and start the same shapes again,
    unclassified, under the new path. A moved row is the same row — its id,
    judgment, uses edges and history ride along — and a `moved` event
    records the path it left. A row already at the target (the sync ran
    first, or a provisional judgment landed there) is replaced when nobody
    judged it; a judged one stands, and the old row is left where it is and
    reported in ``kept``. Bodies are not compared here: the next sync does
    that as always, and asks for a recheck where a judged body changed.
    """
    from sqlalchemy import or_

    from grimoire.services import access
    from grimoire.services.snippets import _path_touches

    old = (from_path or "").strip().strip("/")
    new = (to_path or "").strip().strip("/")
    if not old or not new:
        raise ValueError("from_path and to_path are both required")
    if _path_touches(old, new) or _path_touches(new, old):
        raise ValueError(f"`{old}` and `{new}` overlap; move to a path outside the source")
    if not await access.can_write_project(user_id, project_id):
        raise ValueError(f"project {project_id} not found or no write access")
    now = datetime.now(timezone.utc)
    async with async_session() as session:
        def beneath(prefix: str):
            return select(CodeShape).where(
                CodeShape.project_id == project_id,
                or_(CodeShape.path == prefix, CodeShape.path.like(prefix + "/%")),
            )

        # LIKE reads `_` as a wildcard; the prefix test is the real filter.
        sources = [
            r for r in (await session.execute(beneath(old))).scalars().all()
            if _path_touches(old, r.path)
        ]
        targets = {
            (r.repo_key, r.path, r.symbol, r.kind): r
            for r in (await session.execute(beneath(new))).scalars().all()
        }
        moved, replaced, kept = 0, 0, []
        for row in sources:
            dest = new + row.path[len(old):]
            there = targets.get((row.repo_key, dest, row.symbol, row.kind))
            if there is not None:
                if there.status not in _MECHANICAL_TODO:
                    kept.append({"path": row.path, "symbol": row.symbol, "kind": row.kind})
                    continue
                # The target is the live sighting when the sync got there
                # first; the moved row takes over its presence.
                row.vanished_at = there.vanished_at
                row.last_seen_commit = there.last_seen_commit or row.last_seen_commit
                await session.delete(there)
                await session.flush()
                replaced += 1
            session.add(CodeShapeEvent(
                shape_id=row.id, project_id=project_id,
                path=row.path, symbol=row.symbol, kind=row.kind,
                event="moved", status=row.status, snippet_id=row.snippet_id,
                classified_by=row.classified_by, reason=row.reason,
                commit=row.last_seen_commit or "", at=now,
            ))
            row.path = dest
            moved += 1
        await session.commit()
    return {"moved": moved, "replaced": replaced, "kept": kept}


def _event(row: CodeShape, event: str, at: datetime, *, commit: str = "") -> CodeShapeEvent:
    """A history row for a state change on ``row`` — status/snippet/by/reason
    are the row's CURRENT values, which for `classified` is the judgment
    just made and for presence events is the standing one."""
    return CodeShapeEvent(
        shape_id=row.id, project_id=row.project_id,
        path=row.path, symbol=row.symbol, kind=row.kind,
        event=event, status=row.status, snippet_id=row.snippet_id,
        classified_by=row.classified_by, reason=row.reason,
        commit=commit or row.last_seen_commit or "", at=at,
    )


async def _judge(
    session, row: CodeShape, *, status: str, snippet_id: int | None,
    by: str | None, reason: str | None, at: datetime, reason_code: str | None = None,
) -> None:
    """Apply a judgment to a row — the ONE place a status is set — and write
    its history. Clears what a judgment settles: the standing proposal, the
    recheck ask, the divergence flag; remembers the fingerprint judged.
    `unclassified` is the withdrawal: fields clear, the examination is
    forgotten so the proposer looks again, and history records the
    withdrawal too."""
    row.status = status
    row.snippet_id = snippet_id if status in _NEEDS_TARGET else None
    row.reason = (reason or "").strip() or None
    row.reason_code = (reason_code or "").strip() or None
    row.classified_by = by if status != "unclassified" else None
    row.classified_at = at if status != "unclassified" else None
    row.classified_sha = row.body_sha if status != "unclassified" else ""
    row.recheck_at = None
    row.diverges_from = None
    _clear_proposal(row, reexamine=(status == "unclassified"))
    if row.id is None:
        # A provisional row (hook stamp on a shape not yet synced): flush so
        # the event can point at it.
        session.add(row)
        await session.flush()
    session.add(_event(row, "classified", at))


async def record_uses(
    session, row: CodeShape, snippet_ids, *, basis: str, evidence: str | None = None,
) -> int:
    """Upsert consumption edges shape → snippet. A judgment-grade
    basis (agent/audit/import) overwrites a mechanical one (reference/hook)
    on the same edge; mechanical never overwrites a judgment. Returns the
    number of edges written or refreshed. The row must be persisted (flushed)
    so it has an id."""
    wanted = {int(x) for x in (snippet_ids or []) if x}
    if not wanted:
        return 0
    if row.id is None:
        session.add(row)
        await session.flush()
    existing = {
        e.snippet_id: e
        for e in (
            await session.execute(
                select(CodeShapeUse).where(CodeShapeUse.shape_id == row.id)
            )
        ).scalars().all()
    }
    judged = basis in _CALLER_VIAS
    n = 0
    for sid in wanted:
        edge = existing.get(sid)
        if edge is None:
            session.add(CodeShapeUse(shape_id=row.id, snippet_id=sid, basis=basis, evidence=evidence))
            n += 1
        elif judged or edge.basis not in _CALLER_VIAS:
            edge.basis, edge.evidence = basis, evidence
            n += 1
    return n


async def replace_reference_uses(
    session, row: CodeShape, snippet_ids, *, evidence: str | None = None,
) -> int:
    """Make the row's mechanical `reference` edges exactly ``snippet_ids``.

    `record_uses` only adds, so a by-name rule that tightens would otherwise
    leave every edge the looser rule wrote. Only `reference` edges are
    dropped: a judgment, or the write-path hook's evidence, outlives a
    re-read of the body. Returns the number of edges written or refreshed."""
    wanted = {int(x) for x in (snippet_ids or []) if x}
    if row.id is not None:
        stale = (
            await session.execute(
                select(CodeShapeUse).where(
                    CodeShapeUse.shape_id == row.id,
                    CodeShapeUse.basis == "reference",
                )
            )
        ).scalars().all()
        for edge in stale:
            if edge.snippet_id not in wanted:
                await session.delete(edge)
    return await record_uses(session, row, wanted, basis="reference", evidence=evidence)


async def uses_of(shape_ids) -> dict[int, list[CodeShapeUse]]:
    """{shape_id: [edges]} for a set of rows — the read side of record_uses."""
    ids = [int(x) for x in shape_ids if x]
    if not ids:
        return {}
    async with async_session() as session:
        edges = (
            await session.execute(
                select(CodeShapeUse).where(CodeShapeUse.shape_id.in_(ids))
                .order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
            )
        ).scalars().all()
    out: dict[int, list[CodeShapeUse]] = {}
    for e in edges:
        out.setdefault(e.shape_id, []).append(e)
    return out


# --- the CSS consumer map ---------------------------------------------------


def resolve_consumers(
    css_rows: Iterable[tuple[int, str, str]],
    references: dict[str, dict[str, int]],
) -> dict[tuple[int, str], int]:
    """{(shape_id, consumer_path): count} — which CSS rows each file's markup
    consumes. ``css_rows`` are (id, path, symbol) of the repo's live css rows;
    ``references`` is scan_archive's path → class token → count.

    Resolution: a class named in file F resolves to F's OWN row
    of that name when F defines it (a scoped rule is consumed by its own
    template); otherwise to every other file's row of that name — a shared
    sheet, or, when several files define it, all of them: the map says
    "ambiguous" by fanning out rather than guessing one.

    A token ending in ``PREFIX_MARK`` is a PREFIX reference — the
    static head of a name the template concatenates, `status-*` from
    `` `status-${s}` ``. It stands for every row whose symbol starts with
    that head, each resolved by the same own-file-else-fan-out rule. The
    template cannot tell us WHICH of them it built, so the map credits all
    of them rather than calling live rules unused."""
    # Lazy, like the extract_definitions import below: coverage reaches into
    # this module during a refresh, so neither may import the other at load.
    from grimoire.services.coverage import PREFIX_MARK

    by_symbol: dict[str, list[tuple[int, str]]] = {}
    for sid, path, symbol in css_rows:
        by_symbol.setdefault(symbol, []).append((sid, path))
    out: dict[tuple[int, str], int] = {}

    def credit(rows: list[tuple[int, str]], consumer: str, count: int) -> None:
        own = [sid for sid, path in rows if path == consumer]
        for sid in own or [sid for sid, _path in rows]:
            out[(sid, consumer)] = out.get((sid, consumer), 0) + int(count)

    for consumer, tokens in references.items():
        for token, count in tokens.items():
            if token.endswith(PREFIX_MARK):
                head = token[: -len(PREFIX_MARK)]
                for symbol, rows in by_symbol.items():
                    if symbol.startswith(head):
                        credit(rows, consumer, count)
                continue
            rows = by_symbol.get(token)
            if rows:
                credit(rows, consumer, count)
    return out


async def sync_repo_consumers(
    project_id: int, repo_key: str, references: dict[str, dict[str, int]]
) -> int:
    """Rebuild one repo's consumer edges from its archive's class references:
    insert the new, refresh changed counts, delete what the tree no longer
    says (a template rewritten, a class renamed, a file gone). Edges hang on
    live rows only; a vanished row's edges go with this pass. Returns how
    many edges stand afterwards."""
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape.id, CodeShape.path, CodeShape.symbol, CodeShape.vanished_at).where(
                    CodeShape.project_id == project_id,
                    CodeShape.repo_key == repo_key,
                    CodeShape.kind == "css",
                )
            )
        ).all()
        live = [(r[0], r[1], r[2]) for r in rows if r[3] is None]
        all_ids = [r[0] for r in rows]
        wanted = resolve_consumers(live, references)
        existing = (
            await session.execute(
                select(CodeShapeConsumer).where(CodeShapeConsumer.shape_id.in_(all_ids))
            )
        ).scalars().all() if all_ids else []
        have = {(e.shape_id, e.path): e for e in existing}
        for key, edge in have.items():
            if key not in wanted:
                await session.delete(edge)
            elif edge.count != wanted[key]:
                edge.count = wanted[key]
        for (sid, path), count in wanted.items():
            if (sid, path) not in have:
                session.add(CodeShapeConsumer(shape_id=sid, path=path, count=count, basis="template"))
        await session.commit()
    return len(wanted)


async def consumers_of(shape_ids) -> dict[int, list[CodeShapeConsumer]]:
    """{shape_id: [edges]} for a set of rows — the read side of the map,
    ordered by path so a readout is stable."""
    ids = [int(x) for x in shape_ids if x]
    if not ids:
        return {}
    async with async_session() as session:
        edges = (
            await session.execute(
                select(CodeShapeConsumer).where(CodeShapeConsumer.shape_id.in_(ids))
                .order_by(CodeShapeConsumer.shape_id, CodeShapeConsumer.path)
            )
        ).scalars().all()
    out: dict[int, list[CodeShapeConsumer]] = {}
    for e in edges:
        out.setdefault(e.shape_id, []).append(e)
    return out


# How many consumer files a readout names before "+N more".
_CONSUMERS_SHOWN = 4


def consumer_summary(paths: Iterable[str]) -> dict:
    """{"count", "paths"} — distinct consumer files, sorted, the first few
    named. The one shape every surface uses for "used by N template(s)"."""
    files = sorted(set(paths))
    return {"count": len(files), "paths": files[:_CONSUMERS_SHOWN]}


async def used_by_map(rows: Iterable[CodeShape]) -> dict[int, dict]:
    """{shape_id: consumer_summary} for every css row given — a row with no
    consumer gets {"count": 0, "paths": []}: "no template names it" is a
    finding, not an absence."""
    css = [r for r in rows if r.kind == "css"]
    if not css:
        return {}
    edges = await consumers_of([r.id for r in css])
    return {r.id: consumer_summary(e.path for e in edges.get(r.id, [])) for r in css}


async def mark_canonicals(
    project_id: int, recorded: list[tuple[int, str, str]]
) -> None:
    """Stamp snippet reference locations as `canonical` — the one mechanical
    rule that is always safe (the judgment happened when the snippet was
    minted; this row just makes it queryable).

    ``recorded`` is (snippet_note_id, path, symbol) for every live snippet
    location in the project. Touches only rows machinery owns: unclassified
    rows gain the stamp; mechanically-stamped canonicals no longer covered
    fall back to unclassified. Agent judgments are never overwritten.
    """
    usable = [(nid, p, s) for nid, p, s in recorded if (s or "").strip()]
    # A location naming a FILE and no symbol is a whole-file record: it makes
    # no claim about a definition inside (`location_covers`), but it is
    # exactly the claim a `file` row asks about.
    whole_files = {(p or "").strip(): nid for nid, p, s in recorded
                   if (p or "").strip() and not (s or "").strip()}
    now = datetime.now(timezone.utc)
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        for row in rows:
            covering = next(
                (
                    nid for nid, lp, ls in usable
                    if location_covers(lp, ls, row.path, row.symbol)
                ),
                None,
            )
            if covering is None and row.kind == "file":
                covering = whole_files.get(row.path)
            if covering is not None and row.status in _MECHANICAL_TODO:
                await _judge(session, row, status="canonical", snippet_id=covering,
                             by="mechanical", reason=None, at=now)
            elif (
                covering is None
                and row.status == "canonical"
                and row.classified_by == "mechanical"
            ):
                await _judge(session, row, status="unclassified", snippet_id=None,
                             by=None, reason=None, at=now)
        await session.commit()


async def live_rows_for(user_id: int, project_id: int) -> list[CodeShape]:
    """`live_rows` behind the project read gate — for callers that arrive from
    outside (an MCP tool, a route) rather than from a job that already
    established who is asking. Empty for a project the caller cannot read,
    never a partial answer."""
    # Deferred, like every other access import in this module: services/access
    # reaches back here, and a module-level import closes the cycle.
    from grimoire.services import access

    if not await access.can_read_project(user_id, project_id):
        return []
    return await live_rows(project_id)


async def live_rows(project_id: int) -> list[CodeShape]:
    """Every un-vanished ledger row for a project — the accounting readout's
    input, across ALL its repos (a repo unreachable this refresh still counts;
    accounting is project-wide)."""
    async with async_session() as session:
        return list(
            (
                await session.execute(
                    select(CodeShape).where(
                        CodeShape.project_id == project_id,
                        CodeShape.vanished_at.is_(None),
                    )
                )
            ).scalars().all()
        )


# --- classification: the judgment write path --------------------------------

# Statuses an explicit classification may set. All five: setting a row back to
# `unclassified` is how a judgment is deliberately withdrawn.
_SETTABLE = ("canonical", "instance", "variant", "exempt", "unclassified")

# Who may appear as the classifier on this path. `hook` and `mechanical` are
# server-internal feeds (steps 5-6) — a caller claiming them would launder a
# judgment as machinery.
_CALLER_VIAS = ("agent", "audit", "import")


def validate_classifications(items: list[dict]) -> str | None:
    """The structural error a classification batch would earn, or None.

    Pure and checked BEFORE anything is touched: a batch either applies or
    errors whole, so a caller never learns later that half a batch silently
    happened.
    """
    if not items:
        return "classifications is empty — nothing to apply"
    for i, item in enumerate(items):
        if not isinstance(item, dict):
            return f"classifications[{i}] is not an object"
        path = (item.get("path") or "").strip()
        symbol = (item.get("symbol") or "").strip()
        if not path or not symbol:
            return f"classifications[{i}] needs both path and symbol"
        status = item.get("status") or ""
        if status not in _SETTABLE:
            return (
                f"classifications[{i}] has unknown status {status!r} "
                f"(one of: {', '.join(_SETTABLE)})"
            )
        snippet_id = item.get("snippet_id") or 0
        if status in _NEEDS_TARGET and not snippet_id:
            return (
                f"classifications[{i}]: status {status!r} needs snippet_id (or idea_id) — "
                "the snippet this shape is (or departs from)"
            )
        if status in ("variant", "exempt") and not (item.get("reason") or "").strip():
            return (
                f"classifications[{i}]: status {status!r} needs a reason — "
                "the WHY is the record"
            )
        code = (item.get("reason_code") or "").strip()
        if code and code not in REASON_CODES:
            return (
                f"classifications[{i}]: unknown reason_code {code!r} "
                f"(one of: {', '.join(REASON_CODES)})"
            )
        uses = item.get("uses")
        if uses is not None and (
            not isinstance(uses, list) or not all(isinstance(u, int) and u > 0 for u in uses)
        ):
            return f"classifications[{i}]: uses must be a list of snippet ids"
    return None


async def classify_shapes(
    user_id: int,
    project_id: int,
    classifications: list[dict],
    *,
    via: str = "agent",
    repo: str = "",
) -> dict:
    """Apply a batch of judgments to a project's live ledger rows.

    All-or-nothing on errors: the whole batch is validated (structure, write
    access, every snippet target readable by the caller) before any row is
    touched. Rows are matched by exact (path, symbol) — plus kind when the
    item carries one — and a target no live row matches is reported in
    ``unmatched``, not an error: the tree may simply have moved since the
    caller listed. Idempotent by construction.

    ``repo`` — a repo bound to the project — lets a judgment land on a shape
    the ledger has not synced yet: the shape the agent wrote
    this turn. Such an item becomes a PROVISIONAL row under that repo, the
    same row the write-path hook has always created, with no seen commit;
    the next sync confirms it, or marks it vanished until the code reaches
    the bound ref, and either way the judgment rides along (a vanished row
    that reappears keeps its status). Without ``repo`` an unknown shape is
    `unmatched`, exactly as before. An unbound ``repo`` is refused rather
    than ignored: a verdict silently dropped reads as one recorded.
    """
    from grimoire.services import repo_bindings as repo_bindings_svc

    from grimoire.services import access
    from grimoire.services import snippets as snippets_svc

    if via not in _CALLER_VIAS:
        raise ValueError(f"via must be one of: {', '.join(_CALLER_VIAS)}")
    classifications = await _resolve_ideas(user_id, classifications)
    error = validate_classifications(classifications)
    if error:
        raise ValueError(error)
    if not await access.can_write_project(user_id, project_id):
        raise ValueError(f"project {project_id} not found or no write access")

    # Snippet targets resolve through the caller's own read access — a
    # family-canon snippet in another project counts, a snippet
    # the caller cannot read does not exist for them.
    target_ids = {
        int(item["snippet_id"])
        for item in classifications
        if item.get("status") in _NEEDS_TARGET
    }
    for item in classifications:
        target_ids.update(int(u) for u in (item.get("uses") or []))
    for sid in sorted(target_ids):
        if await snippets_svc.get_snippet(user_id, sid) is None:
            raise ValueError(f"snippet {sid} not found (or not readable)")
    repo_key = ""
    if (repo or "").strip():
        repo_key = await repo_bindings_svc.is_bound(project_id, repo)
        if not repo_key:
            raise ValueError(
                f"repo {repo!r} is not bound to project {project_id} — "
                "bind_repo it, or omit repo to judge synced shapes only"
            )

    now = datetime.now(timezone.utc)
    classified = 0
    provisional = 0
    unmatched: list[dict] = []
    touched: set[int] = set()
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        by_key: dict[tuple[str, str], list[CodeShape]] = {}
        for row in rows:
            by_key.setdefault((row.path, row.symbol), []).append(row)
        for item in classifications:
            matches = by_key.get(
                ((item.get("path") or "").strip(), (item.get("symbol") or "").strip())
            ) or []
            kind = (item.get("kind") or "").strip()
            if kind:
                matches = [r for r in matches if r.kind == kind]
            if not matches and repo_key and item.get("status") != "unclassified":
                path = (item.get("path") or "").strip()
                symbol = (item.get("symbol") or "").strip()
                row = CodeShape(
                    project_id=project_id, repo_key=repo_key, path=path,
                    symbol=symbol, kind=kind or snippet_kind(symbol, ""),
                )
                session.add(row)
                by_key.setdefault((path, symbol), []).append(row)
                matches = [row]
                provisional += 1
            if not matches:
                unmatched.append({
                    "path": item.get("path"), "symbol": item.get("symbol"),
                })
                continue
            status = item["status"]
            for row in matches:
                touched.update(x for x in (row.snippet_id, item.get("snippet_id")) if x)
                await _judge(
                    session, row, status=status,
                    snippet_id=int(item["snippet_id"]) if status in _NEEDS_TARGET else None,
                    by=via, reason=item.get("reason"), at=now,
                    reason_code=item.get("reason_code"),
                )
                if item.get("uses"):
                    await record_uses(session, row, item["uses"], basis=via,
                                      evidence=item.get("reason"))
                classified += 1
        await session.commit()
    out = {"classified": classified, "unmatched": unmatched}
    if provisional:
        out["provisional"] = provisional
    moved = await _sync_family(user_id, project_id, touched)
    if moved:
        out["family"] = moved
    return out


async def _resolve_ideas(user_id: int, items: list[dict]) -> list[dict]:
    """An item may name a family idea (`idea_id`) in place of a snippet: it
    is judged against the idea's reference in the shape's language, or its
    first when there is none in that language — an idea is shared across
    languages. An explicit snippet_id wins."""
    from grimoire.services import family_adoption

    out = []
    for i, item in enumerate(items or []):
        if (isinstance(item, dict) and item.get("idea_id") and not item.get("snippet_id")
                and item.get("status") in _NEEDS_TARGET):
            try:
                sid = await family_adoption.reference_for(
                    user_id, int(item["idea_id"]), (item.get("path") or "").strip())
            except (TypeError, ValueError) as exc:
                raise ValueError(f"classifications[{i}]: {exc}") from None
            item = {**item, "snippet_id": sid}
        out.append(item)
    return out


async def _sync_family(user_id: int, project_id: int, snippet_ids: set[int]) -> list[dict]:
    """The adoption ledger follows the shapes: after a
    judgment commits, the canon ideas whose references it touched — as the
    new target or the one it replaced — are answered from the shapes.

    A failure is logged and the judgment stands; the next judgment touching
    those references, or the next coverage refresh, brings the two back in
    line."""
    from grimoire.services import family_adoption

    if not snippet_ids:
        return []
    try:
        return await family_adoption.sync_from_shapes(user_id, project_id, snippet_ids)
    except Exception:
        logger.warning("family adoption sync failed for project %s", project_id, exc_info=True)
        return []


def rule_matches(row: CodeShape, *, path: str, pattern: str, kind: str) -> bool:
    """Does a ledger row fall under a rule-form classification?
    ``path`` is a file or a directory (everything beneath it), ``pattern``
    a shell glob on the symbol (``""`` = every symbol), ``kind`` narrows to
    sym/css. Pure, so the sweep's reach can be tested without a database."""
    import fnmatch

    clean = (path or "").strip().strip("/")
    if clean and not (row.path == clean or row.path.startswith(clean + "/")):
        return False
    if kind and row.kind != kind:
        return False
    if pattern and not fnmatch.fnmatchcase(_norm_symbol(row.symbol), pattern):
        return False
    return True


async def classify_shapes_where(
    user_id: int,
    project_id: int,
    *,
    path: str,
    status: str,
    pattern: str = "",
    kind: str = "",
    snippet_id: int | None = None,
    reason: str | None = None,
    via: str = "agent",
    include_judged: bool = False,
    reason_code: str | None = None,
    uses: list[int] | None = None,
) -> dict:
    """The sweep form of classify_shapes: one judgment applied to
    every live row under ``path`` whose symbol matches ``pattern`` (and
    ``kind``). By default only unjudged rows are touched — `unclassified`
    and the sync's mechanical `scoped` stamp — a sweep must never silently
    overwrite a judgment; ``include_judged`` opts in.
    Same gates as the row form (status vocabulary, snippet target, reason
    for variant/exempt, write access); one transaction, so it applies whole
    or not at all. Returns the count and a sample of what it judged."""
    from grimoire.services import access
    from grimoire.services import snippets as snippets_svc

    if via not in _CALLER_VIAS:
        raise ValueError(f"via must be one of: {', '.join(_CALLER_VIAS)}")
    if not (path or "").strip():
        raise ValueError("path is required — a sweep names the directory it judges")
    if status == "canonical":
        raise ValueError("canonical is the sync's stamp on a snippet's own location — a sweep cannot set it")
    probe = {"path": path, "symbol": "*", "status": status,
             "snippet_id": snippet_id or 0, "reason": reason or "",
             "reason_code": reason_code or "", "uses": uses}
    error = validate_classifications([probe])
    if error:
        raise ValueError(error.replace("classifications[0]", "rule"))
    if not await access.can_write_project(user_id, project_id):
        raise ValueError(f"project {project_id} not found or no write access")
    if status in _NEEDS_TARGET and await snippets_svc.get_snippet(user_id, int(snippet_id)) is None:
        raise ValueError(f"snippet {snippet_id} not found (or not readable)")
    for sid in sorted({int(u) for u in (uses or [])}):
        if await snippets_svc.get_snippet(user_id, sid) is None:
            raise ValueError(f"snippet {sid} not found (or not readable)")

    now = datetime.now(timezone.utc)
    judged: list[str] = []
    touched: set[int] = {int(snippet_id)} if snippet_id and status in _NEEDS_TARGET else set()
    async with async_session() as session:
        conds = [CodeShape.project_id == project_id, CodeShape.vanished_at.is_(None)]
        if not include_judged:
            conds.append(CodeShape.status.in_(_MECHANICAL_TODO))
        rows = (await session.execute(select(CodeShape).where(*conds))).scalars().all()
        for row in rows:
            if not rule_matches(row, path=path, pattern=pattern, kind=kind):
                continue
            if row.snippet_id:
                touched.add(row.snippet_id)
            await _judge(
                session, row, status=status,
                snippet_id=int(snippet_id) if status in _NEEDS_TARGET else None,
                by=via, reason=reason, at=now, reason_code=reason_code,
            )
            if uses:
                await record_uses(session, row, uses, basis=via, evidence=reason)
            judged.append(f"{row.path}::{row.symbol}")
        await session.commit()
    out = {"classified": len(judged), "sample": sorted(judged)[:12]}
    moved = await _sync_family(user_id, project_id, touched)
    if moved:
        out["family"] = moved
    return out


async def list_project_shapes(
    user_id: int,
    project_id: int,
    *,
    status: str = "",
    path: str = "",
    snippet_id: int = 0,
    include_vanished: bool = False,
    limit: int = 100,
    offset: int = 0,
    proposal: str = "",
    flag: str = "",
    uses: int = 0,
) -> tuple[list[CodeShape], int]:
    """A filtered page of a project's ledger, with the unfiltered-match total.

    ([], 0) when the caller can't read the project — the same silence every
    other project list gives. ``path`` matches the exact file or anything
    beneath it, mirroring recorded-location semantics. ``proposal`` narrows
    to rows the proposer has spoken about: "any", "canon" (an instance-of-#N
    suggestion), "derive" (a repeats-with-no-canon group), or one basis
    name (symbol/reference/text/signature/semantic). ``flag`` narrows to
    the readout's asks: "divergence" (new where a canon dominates,
    `diverges_from` names it), "recheck" (a judged shape whose body moved),
    or "unused-css" (a css rule no template names).
    """
    from sqlalchemy import func, or_

    from grimoire.services import access

    if not await access.can_read_project(user_id, project_id):
        return [], 0
    conds = [CodeShape.project_id == project_id]
    if not include_vanished:
        conds.append(CodeShape.vanished_at.is_(None))
    if status:
        conds.append(CodeShape.status == status)
    if path:
        clean = path.strip().strip("/")
        conds.append(or_(
            CodeShape.path == clean, CodeShape.path.like(clean + "/%")
        ))
    if snippet_id:
        conds.append(CodeShape.snippet_id == snippet_id)
    if proposal == "any":
        conds.append(or_(
            CodeShape.proposed_snippet_id.isnot(None),
            CodeShape.proposal_group.isnot(None),
        ))
    elif proposal == "canon":
        conds.append(CodeShape.proposed_snippet_id.isnot(None))
    elif proposal == "derive":
        conds.append(CodeShape.proposal_group.isnot(None))
    elif proposal:
        conds.append(CodeShape.proposal_basis == proposal)
    if flag == "divergence":
        conds.append(CodeShape.diverges_from.isnot(None))
    elif flag == "recheck":
        conds.append(CodeShape.recheck_at.isnot(None))
    elif flag == "unused-css":
        # The consumer map's negative space: a live css rule
        # no file's markup names. A candidate for deletion, surfaced — never
        # deleted — because the map reads templates only (a class built at
        # runtime, or used from a script, is invisible to it).
        conds.append(CodeShape.kind == "css")
        conds.append(~CodeShape.id.in_(select(CodeShapeConsumer.shape_id)))
    if uses:
        # Consumers of a canon: rows with a uses edge to it, whatever
        # shape they themselves are.
        conds.append(CodeShape.id.in_(
            select(CodeShapeUse.shape_id).where(CodeShapeUse.snippet_id == uses)
        ))
    async with async_session() as session:
        total = (
            await session.execute(
                select(func.count()).select_from(CodeShape).where(*conds)
            )
        ).scalar_one()
        rows = (
            await session.execute(
                select(CodeShape).where(*conds)
                .order_by(CodeShape.path, CodeShape.symbol, CodeShape.kind)
                .limit(max(1, min(limit, 500))).offset(max(0, offset))
            )
        ).scalars().all()
    return list(rows), int(total)


def _consumer_dict(row: CodeShape) -> dict:
    """The compact shape a snippet's consumer map carries — enough to open
    the file, none of the ledger bookkeeping."""
    out = {
        "project_id": row.project_id,
        "repo": row.repo_key,
        "path": row.path,
        "symbol": row.symbol,
        "kind": row.kind,
        "classified_by": row.classified_by,
    }
    if row.reason:
        out["reason"] = row.reason
    return out


async def snippet_consumers(user_id: int, note_id: int) -> dict:
    """The structured consumer map for one snippet: its `instances`
    (rows judged to conform) and `variants` (named departures, each carrying
    its why). Rows are filtered to projects the CALLER can read — a shared
    snippet must not become a side channel into someone else's project
    layout. Empty lists mean "attach nothing"."""
    from grimoire.services import access

    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.snippet_id == note_id,
                    CodeShape.status.in_(("instance", "variant")),
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        # The consumption edges: rows that USE this canon, whatever
        # shape they are themselves — the call-site map.
        using = (
            await session.execute(
                select(CodeShape, CodeShapeUse.basis, CodeShapeUse.evidence)
                .join(CodeShapeUse, CodeShapeUse.shape_id == CodeShape.id)
                .where(CodeShapeUse.snippet_id == note_id, CodeShape.vanished_at.is_(None))
            )
        ).all()
    readable: dict[int, bool] = {}

    async def can_read(pid: int) -> bool:
        if pid not in readable:
            readable[pid] = await access.can_read_project(user_id, pid)
        return readable[pid]

    out: dict[str, list[dict]] = {"instances": [], "variants": [], "uses": []}
    for row in rows:
        if not await can_read(row.project_id):
            continue
        out["instances" if row.status == "instance" else "variants"].append(
            _consumer_dict(row)
        )
    for row, basis, evidence in using:
        if not await can_read(row.project_id):
            continue
        d = _consumer_dict(row)
        d["basis"] = basis
        if evidence:
            d["evidence"] = evidence
        d.pop("reason", None)
        out["uses"].append(d)
    return out


# --- write-path stamping: hook evidence lands as rows -----------------------
#
# The write-path hook (plugin/hooks/grimoire_prior_art.sh) fires on every
# Write/Edit and already carries the two halves of a consumer-map row: the
# file being written and the definitions in (or enclosing) the payload. What
# it could not say on its own is WHICH canon the session is instantiating.
# The pull stream answers that: a snippet the session opened in full
# (get_snippet) and is now writing code that references or resembles is being
# reused — and a reused canon's call site is an `instance`.
#
# The rule, deliberately two-sided so it cannot fire on noise:
#   PULLED   — a PULLED usage event by this user inside PULL_WINDOW. Offered-
#              but-ignored (surfaced, never opened) stamps nothing.
#   IN PLAY  — the payload references the snippet's symbol by name, or the
#              semantic arm scored it above the write-path threshold for this
#              very payload. Either is evidence; the pull alone is not.
# Both hold → every shape the hook named at that path, of the snippet's kind,
# carries N as a PROPOSAL. Older rows stamped instance-of-N with
# classified_by="hook" — a verdict nobody read — are what `stamps_to_review`
# lists.
#
# A proposal is EVIDENCE, not judgment: it lands only on rows nobody has
# judged, never touches a status, and is what the end-of-turn question shows
# the agent as "looks like #N". The agent's classify_shapes is the verdict.

# "The write path actually pulled it": a working session's reach. The
# precision comes from the in-play test above, not from this window.
PULL_WINDOW = timedelta(hours=6)

# ── WHAT KIND OF THING a shape is, past `css | sym` ──────────────────────
#
# THE PROBLEM THIS EXISTS FOR. `kind` has exactly two values, so a frozen
# dataclass, a module constant, a sync predicate, a class and an async service
# function are all siblings of one another. Nothing downstream could possibly
# discriminate on a distinction the column does not carry, which leaves a
# directory base rate standing alone as the entire divergence argument. The
# consequence: a registry module of pure helpers gets prompted to build from
# an `async_session` service canon, and one snippet can be recorded as the
# canon for a `class`, a plain `def`, an `async def` and a dozen test
# functions at once.
#
# DERIVED FROM THE SIGNATURE ON READ, not stored. `kind` is part of the row
# identity — ("project_id", "repo_key", "path", "symbol", "kind") — so
# widening that column rewrites every row's key and needs a migration plus a
# re-extract. Reading the form off the signature costs nothing, needs no
# migration, and is reversible. Widening `kind` remains the principled fix and
# this does not foreclose it: every caller below asks `shape_form`, so the day
# the column carries the answer, this function returns it instead.
#
# DELIBERATELY COARSE, and it must stay that way. It answers "could these
# plausibly be the same kind of thing", never "what language construct is
# this". A finer taxonomy would need per-language parsing and would start
# disagreeing with itself across the four languages this ledger already holds
# (Python, Go, TypeScript, Vue SFC) — and a classifier that is wrong in a new
# way is worse than the coarse one it replaced.
FORM_UNKNOWN = ""

# How much a canon's own instances must agree before it may assert a form.
# A canon whose rows disagree about what they are cannot tell anyone else what
# to be — and that disagreement is the SYMPTOM of the bad stamping this same
# change fixes, so reading it as "no opinion" makes the two halves cooperate:
# a canon poisoned by loose stamping falls silent instead of flagging.
_FORM_SHARE = 0.6

# How much of a canon's membership may be shapes it cannot account for
# before the canon stops meaning anything. Not a majority test — see
# `canon_coherence` for why a majority test goes blind exactly when the
# ledger is worst.
_STRANGER_SHARE = 0.2

# Leading words that say nothing about the form of the thing being declared.
_FORM_NOISE = ("export ", "default ", "public ", "private ", "static ", "final ")

# A declared TYPE, across the languages in play. Coarse on purpose: a Go
# struct, a TS interface and a Python class are one bucket because the
# question is only ever "is the other thing also a type".
_FORM_TYPE_WORDS = ("class ", "interface ", "type ", "struct ", "enum ")
_FORM_FN_WORDS = ("def ", "function ", "func ", "fn ", "sub ")

# THE DECLARING KEYWORD IS OPTIONAL, because Python has none. A module
# constant is written `MAX_BUDGET = 10` — no `const`, no `let` — and a
# pattern that required one classified every Python constant as unreadable,
# which then read downstream as "do not assert" and quietly excluded a whole
# form from both checks. Caught by the payload test, not by review.
_FORM_BINDING = re.compile(r"^(?:(?:const|let|var)\s+)?[\w$]+\s*(?::[^=]+)?=")
_FORM_ARROW = re.compile(
    r"^(?:(?:const|let|var)\s+)?[\w$]+\s*(?::[^=]+)?=\s*(async\s*)?\("
)


def shape_form(signature: str, kind: str = "sym") -> str:
    """The structural form of a shape: css / type / async-fn / fn / binding.

    Returns FORM_UNKNOWN when the signature does not say, and every caller
    treats that as "do not assert", never as "no match". An unreadable
    signature must make this quieter, not more confident.
    """
    if kind == "css":
        return "css"
    if kind == "file":
        # A whole file: its own family, comparable only with
        # other files — a component is never told to build from a helper.
        return "file"
    sig = (signature or "").strip()
    if not sig:
        return FORM_UNKNOWN
    changed = True
    while changed:
        changed = False
        for lead in _FORM_NOISE:
            if sig.startswith(lead):
                sig, changed = sig[len(lead):].lstrip(), True
    if sig.startswith(_FORM_TYPE_WORDS):
        return "type"
    if sig.startswith("async "):
        return "async-fn"
    if sig.startswith(_FORM_FN_WORDS):
        return "fn"
    m = _FORM_ARROW.match(sig)
    if m:
        return "async-fn" if m.group(1) else "fn"
    if _FORM_BINDING.match(sig):
        return "binding"
    return FORM_UNKNOWN


def forms_agree(a: str, b: str) -> bool:
    """Do two forms match well enough to assert a relationship?

    BOTH must be known. "Unknown equals unknown" would make two shapes nobody
    can read into a confident pair, which is the failure this whole change is
    about — a guess dressed as a finding.
    """
    return bool(a) and bool(b) and a == b


def forms_conflict(a: str, b: str) -> bool:
    """Do two KNOWN forms rule each other out?

    The weaker sibling of `forms_agree`, and the pair exists because the two
    kinds of evidence this ledger acts on deserve different burdens.

    An explicit by-name reference — the payload literally names the canon's
    symbol — is strong, so it needs only the absence of a contradiction: stamp
    unless the forms are both readable and different. A payload-level
    RESEMBLANCE SCORE is weak, computed against the whole file, so one score
    speaks for every symbol in it; that needs positive agreement before
    asserting anything, which is `forms_agree`.

    Demanding agreement everywhere would silence the by-name path whenever a
    shape's definition is not in the payload — an Edit rather than a Write —
    turning strong evidence into no evidence for a reason that has nothing to
    do with the code.
    """
    return bool(a) and bool(b) and a != b


def shape_family(form: str) -> str:
    """The coarse category a form belongs to, or "" when the form is unknown.

    WHY THIS IS SEPARATE FROM `shape_form`. The two checks in this module ask
    opposite questions, so they cannot share a burden:

      STAMPING asserts "this IS that canon". Agreement in form is evidence
      FOR the claim, so `forms_agree` / `forms_conflict` — the precise level,
      where `fn` and `async-fn` are different — is right.

      DIVERGENCE asserts "this is NOT the canon that dominates here — did you
      mean to?" A form MISMATCH is the premise of that prompt, not an
      objection to it. Gating it on `forms_agree` would invert the check: it
      would go silent on exactly the mismatches it exists to catch, such as a
      hand-rolled sync `confirmDanger` in a directory where an async confirm
      helper is canon.

    So divergence gates at FAMILY level and only on contradiction. A sync
    function beside an async one is still a fair question. A frozen dataclass
    told to build from an async service function is not a question at all.

    This does NOT remove every false prompt. A type against a callable canon
    is silenced; an unrelated callable beside a callable canon still asks,
    because at the signature level it is indistinguishable from the
    `confirmDanger` case above. Only widening `kind` past `css | sym` or
    comparing meaning rather than form can separate them. The gap is known,
    not an oversight.
    """
    if form in ("fn", "async-fn"):
        return "callable"
    if form == "type":
        return "type"
    if form == "binding":
        return "value"
    if form == "css":
        return "css"
    if form == "file":
        return "file"
    return ""


def families_conflict(a: str, b: str) -> bool:
    """Do two forms belong to categorically different KNOWN families?

    Like `forms_conflict`, both sides must be readable: an unknown form makes
    this quieter, never more confident.
    """
    fa, fb = shape_family(a), shape_family(b)
    return bool(fa) and bool(fb) and fa != fb


def canon_form(rows: Iterable, snippet_id: int) -> str:
    """The form a canon's own judged rows agree on, or FORM_UNKNOWN."""
    forms: dict[str, int] = {}
    for r in rows:
        if r.snippet_id != snippet_id or r.status not in ("canonical", "instance"):
            continue
        if is_weak_stamp(r):
            continue
        f = shape_form(getattr(r, "signature", "") or "", r.kind)
        if f:
            forms[f] = forms.get(f, 0) + 1
    if not forms:
        return FORM_UNKNOWN
    top, n = max(forms.items(), key=lambda kv: kv[1])
    return top if n / sum(forms.values()) >= _FORM_SHARE else FORM_UNKNOWN


# A judgment nobody read before it was written. "hook" is the write-path
# stamp — a similarity score and no reader. Every other value ("agent",
# "audit", "import", "mechanical") came from something that examined the
# shape and said so, which is a different kind of evidence, not a stronger
# score.
_UNATTENDED_BY = ("hook",)


def canon_coherence(members: Iterable) -> dict:
    """Does a canon's membership still agree on what the canon IS?

    The verdict is NOT "do all the rows share a form". A form-agreement
    check reports sound canons and misses polluted ones, for two reasons, and
    the fix for the first can break the second:

    1. FAMILY, NOT FORM. A sync loop-starter and the async tick it schedules
       are one shape written two ways; `shape_family` collapses `fn` and
       `async-fn` into `callable` for exactly this reason on the divergence
       side. A canon of four starters and three ticks is coherent, and a
       form count would score it 4/7.

    2. A METHOD OF A MEMBER IS NOT A STRANGER. A model convention can cover
       the class AND the `to_dict` the class must carry. Thirty-seven classes
       and twenty-five serialisers score 37/62 on form, under the bar, for
       containing exactly what the canon says it contains.

    THE TRAP, and it is why this is not simply a family histogram: grouping
    by family and keeping a majority test makes the check BLIND. The same
    model canon polluted with 31 unrelated callables — Vue functions, route
    handlers, tests — holds 37 classes and 56 callables; as families that is
    56/93, a clean pass, and the 31 rows that have no business being there
    are never reported. A looser bar in the same shape is not a fix.

    So the verdict is inverted. Instead of asking whether most rows agree, it
    asks how many rows the canon CANNOT ACCOUNT FOR:

      * a row in the majority family is accounted for;
      * a callable defined in a FILE that also holds a majority-family `type`
        row is accounted for — it is a method of a member;
      * everything else is a STRANGER, and strangers above `_STRANGER_SHARE`
        of the readable rows make the canon incoherent.

    The majority vote abstains those methods, so a class's own serialisers
    cannot outvote the classes and turn the members into the strangers. The
    three cases above separate completely: the clean model canon has 0
    strangers in 62, the loop canon 0 in 7, and the polluted one 31 in 93.

    Scoped to the REVIEW surface on purpose. `canon_form` still answers at
    the precise form level for stamping and divergence, where a sync helper
    beside an async canon is a fair question; nothing here changes what the
    ledger writes.

    Returns a dict, deliberately. Widening a tuple return breaks every
    consumer that still unpacks the old width, and Python says nothing until
    that line runs.
    """
    forms: dict[str, int] = {}
    families: dict[str, int] = {}
    per_row: list[tuple[object, str]] = []
    has_type: set[str] = set()
    for r in members:
        f = shape_form(getattr(r, "signature", "") or "", r.kind)
        if not f:
            continue              # unreadable: never counts either way
        fam = shape_family(f)
        forms[f] = forms.get(f, 0) + 1
        families[fam] = families.get(fam, 0) + 1
        per_row.append((r, fam))
        if fam == "type":
            has_type.add(r.path)

    out = {
        "readable": len(per_row),
        "forms": dict(sorted(forms.items(), key=lambda kv: -kv[1])),
        "families": dict(sorted(families.items(), key=lambda kv: -kv[1])),
        "family": FORM_UNKNOWN,
        "attached": 0,
        "strangers": [],
        "coheres": True,
    }
    if not per_row:
        return out                # nothing legible: say nothing, not "broken"

    # The vote, with methods abstaining. A callable sitting in a file that
    # defines a class is presumed to belong to it and does not get to argue
    # that the canon is really about callables.
    def _attachable(row, fam: str) -> bool:
        return fam == "callable" and row.path in has_type

    votes: dict[str, int] = {}
    for r, fam in per_row:
        if _attachable(r, fam):
            continue
        votes[fam] = votes.get(fam, 0) + 1
    if not votes:                 # every readable row is a method
        votes = dict(families)
    # Ties go to `type`: a canon that defines a class is about the class.
    family = max(votes, key=lambda k: (votes[k], k == "type"))

    attached, strangers = 0, []
    for r, fam in per_row:
        if fam == family:
            continue
        if family == "type" and _attachable(r, fam):
            attached += 1
        else:
            strangers.append(r)
    out["family"] = family
    out["attached"] = attached
    out["strangers"] = strangers
    out["coheres"] = len(strangers) / len(per_row) <= _STRANGER_SHARE
    return out


def signature_in(code: str, symbol: str, kind: str) -> str:
    """The line in ``code`` that DEFINES ``symbol``, or "" if none does.

    The write-time checks are asked about a shape that may have no ledger row
    yet — it is being written right now — so the payload is the only place its
    signature exists. Finding nothing returns "", which reads downstream as
    FORM_UNKNOWN and therefore as silence.
    """
    if not code or not symbol:
        return ""
    name = re.escape(symbol.lstrip("."))
    if kind == "css":
        pat = re.compile(r"^\s*\." + name + r"\b")
    else:
        pat = re.compile(
            r"^\s*(?:(?:export|default|public|private|static|final)\s+)*"
            r"(?:"
            r"(?:(?:async\s+)?(?:def|function|func|fn|sub)|class|interface|type"
            r"|struct|enum|const|let|var)\s+" + name + r"\b"
            # A bare binding — `MAX = 10`, `Handler = ...` — which is how
            # Python (and plain JS assignment) declares one.
            r"|" + name + r"\s*(?::[^=\n]+)?=(?!=)"
            r")"
        )
    for line in code.splitlines():
        if pat.match(line):
            return line.strip()
    return ""


def snippet_kind(symbol: str, language: str) -> str:
    """The ledger kind a snippet's reference belongs to — "css" when its
    symbol is a class selector (or it is a stylesheet with no symbol),
    else "sym"."""
    sym = (symbol or "").strip()
    if sym.startswith("."):
        return "css"
    if not sym and (language or "").strip().lower() in ("css", "scss", "sass", "less"):
        return "css"
    return "sym"


def references_symbol(code: str, symbol: str, kind: str) -> bool:
    """Does this payload name the snippet's symbol? Word-bounded so `confirm`
    never claims `confirmed`; a CSS class matches as `.btn` or inside a class
    attribute (`btn btn-primary`), dashes counting as part of the name."""
    sym = _norm_symbol(symbol or "")
    if not sym or not code:
        return False
    if kind == "css":
        pattern = rf"(?<![\w-]){re.escape(sym)}(?![\w-])"
    else:
        pattern = rf"(?<![\w$]){re.escape(sym)}(?![\w$])"
    return re.search(pattern, code) is not None


async def recent_pulls(user_id: int, *, window: timedelta = PULL_WINDOW) -> dict[int, datetime]:
    """{note_id: last pulled at} for every note this user opened in full
    inside ``window`` — the "actually pulled it" half of the stamping rule.
    Reads the usage telemetry table; an unreadable table means no evidence."""
    from sqlalchemy import func

    from grimoire.models.note_usage import PULLED, NoteUsageEvent

    since = datetime.now(timezone.utc) - window
    try:
        async with async_session() as session:
            rows = await session.execute(
                select(NoteUsageEvent.note_id, func.max(NoteUsageEvent.created_at))
                .where(
                    NoteUsageEvent.user_id == user_id,
                    NoteUsageEvent.event == PULLED,
                    NoteUsageEvent.created_at >= since,
                )
                .group_by(NoteUsageEvent.note_id)
            )
            return {int(nid): ts for nid, ts in rows.all()}
    except Exception:
        logger.warning("recent_pulls read failed — no hook stamping this write", exc_info=True)
        return {}


# THE HOOK'S EVIDENCE, WRITTEN ONCE AND READ BACK BY ONE PARSER.
#
# The stamp's score is stored in a fixed format so something can ask "how
# strong was the evidence for this row?" without re-deriving it. A score that
# is only printed is never queried, and weak rows sit unexamined. Format and
# parser live beside each other so a change to one that breaks the other is a
# visible edit rather than a silent drift.
_REFERENCE_REASON = "hook: pulled #{sid}; payload references `{symbol}`"
_RESEMBLE_REASON = "hook: pulled #{sid}; payload resembles it ({score:.2f})"
_RESEMBLE_READ = re.compile(
    r"^hook: pulled #(\d+); payload resembles it \(([0-9]*\.?[0-9]+)\)$"
)


def stamp_score(reason: str | None) -> float | None:
    """The resemblance a hook stamp recorded, or None if it was not one.

    None means "this row was not written on a similarity score" — a by-name
    reference, an agent judgment, an audit — NOT "the score was zero". Every
    caller treats the two differently, because a row a person judged is not
    weak evidence, it is a different kind of evidence entirely.
    """
    m = _RESEMBLE_READ.match((reason or "").strip())
    return float(m.group(2)) if m else None


def _stamp_allowed(rank: int, mine: str, canon: str) -> bool:
    """May this evidence assert that a shape of form ``mine`` IS ``canon``?

    Named rather than inlined because the asymmetry is the decision, not an
    implementation detail: rank 2 (the payload names the canon's symbol) has
    only to avoid contradicting, rank 1 (a whole-file similarity score) has to
    positively agree. See `forms_conflict` for why both exist.
    """
    if rank >= 2:
        return not forms_conflict(mine, canon)
    return forms_agree(mine, canon)


async def suggest_write_path_instances(
    user_id: int,
    project_id: int,
    *,
    path: str,
    shapes: list[tuple[str, str]],
    code: str,
    pulled: dict[int, datetime],
    resembles: dict[int, float] | None = None,
    repo_key: str = "",
) -> list[dict]:
    """Offer hook evidence as a PROPOSAL on the shapes being written.

    Evidence never lands as an `instance` row: that would be a permanent
    classification written with nobody reading it. The agent that wrote the
    code says what it is, at the end of the turn, and this evidence is what
    it is shown when it does. So the row keeps its status; the candidate
    canon goes into the
    proposal fields (`proposed_snippet_id`, `proposal_basis` "reference" for a
    by-name hit or "semantic" for a resemblance, `proposal_score`), where the
    end-of-turn question reads it as "looks like #N". A judged row is left
    alone entirely.

    ``shapes`` is the hook's (kind, name) list for ``path``; ``pulled`` is
    recent_pulls(); ``resembles`` maps snippet ids the semantic arm scored
    for this payload to their score. Returns the suggestions, each
    {path, symbol, kind, snippet_id, reason} — empty in the common case.

    A shape the ledger has no live row for yet (it is being written right
    now) gets a PROVISIONAL unclassified row under ``repo_key`` — first/
    last-seen empty — so the suggestion survives to the end of the turn and
    the next sync either confirms the shape or stamps it vanished. No repo
    key → only existing rows carry a suggestion.

    Every pulled canon the payload NAMES is still recorded as a `uses` edge:
    that is a fact about the code, not a verdict about what it is.

    When more than one pulled snippet is in play for a shape, a by-name
    reference beats resemblance and the most recent pull breaks ties.
    """
    from grimoire.services import access
    from grimoire.services import snippets as snippets_svc
    from grimoire.services.snippets import snippet_fields

    resembles = resembles or {}
    path = (path or "").strip()
    wanted = [(k, n.strip()) for k, n in shapes if k in ("css", "sym") and n.strip()]
    if not project_id or not path or not wanted or not pulled:
        return []
    if not await access.can_write_project(user_id, project_id):
        return []

    # Which pulled canons are in play for this payload, by kind, ranked.
    #
    # THE FORM OF THE CANON TRAVELS WITH IT, because what gets written
    # here is an ASSERTION — "this shape IS that canon" — and it is permanent
    # until something re-judges it. The evidence below is payload-level: a
    # resemblance score is computed against the whole file, so without a
    # per-shape test every symbol in that file inherits one verdict. Measured
    # on another project: a single write stamped `class SessionAbsent`,
    # `async def attach` and `_run_control_client` as instances of one snippet.
    in_play: dict[str, list[tuple[int, datetime, int, str, str]]] = {}
    for sid, pulled_at in pulled.items():
        note = await snippets_svc.get_snippet(user_id, sid)
        if note is None:
            continue
        fields = snippet_fields(note)
        symbol = fields.get("symbol") or ""
        language = (fields.get("language") or "").strip().lower()
        kind = snippet_kind(symbol, language)
        canon = Canon(
            int(sid), kind, symbol,
            tuple(((loc.get("path") or ""), (loc.get("symbol") or ""))
                  for loc in fields.get("locations") or []),
            "", "", int(note.project_id or 0), language,
        )
        # A redefinition of the pulled name in this payload is the copy the
        # hint exists to catch, so the payload's own names are not "local".
        if (kind != "sym" or same_family(path, language)) and names_canon(
            kind, path, code, canon, project_id=project_id,
        ):
            rank, why = 2, _REFERENCE_REASON.format(
                sid=sid, symbol=_norm_symbol(symbol)
            )
        elif resembles.get(sid, 0.0) >= _RESEMBLE_MIN:
            rank, why = 1, _RESEMBLE_REASON.format(sid=sid, score=resembles[sid])
        else:
            # A SCORE BELOW THE FLOOR IS NOT WEAK EVIDENCE, IT IS NONE.
            # Taking any score at all (`elif sid in resembles`) would count
            # 0.69 exactly as much as 0.95, and a row written on that basis is
            # indistinguishable afterwards from one written on real evidence.
            continue
        in_play.setdefault(kind, []).append(
            (rank, pulled_at, sid, why, shape_form(fields.get("signature") or "", kind))
        )
    if not in_play:
        return []
    for bucket in in_play.values():
        bucket.sort(key=lambda t: (t[0], t[1]), reverse=True)

    suggested: list[dict] = []
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.path == path,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        by_key = {(r.symbol, r.kind): r for r in rows}
        for kind, name in wanted:
            bucket = in_play.get(kind)
            if not bucket:
                continue
            row = by_key.get((name, kind))
            # PER SHAPE, NOT PER FILE. The best candidate whose FORM
            # matches this shape's — not simply the best candidate. A class and
            # a function in one file are never handed the same canon because
            # the file as a whole resembled it.
            #
            # The signature comes from the payload first: a shape being written
            # right now may have no ledger row yet, and then the code is the
            # only place it exists. No readable signature on either side means
            # no stamp — `forms_agree` requires both to be known, so an
            # unreadable shape falls silent instead of matching everything.
            mine = shape_form(
                signature_in(code, name, kind) or getattr(row, "signature", "") or "", kind
            )
            # THE BURDEN SCALES WITH THE EVIDENCE. Rank 2 is an explicit
            # by-name reference to the canon in this very payload; it stands
            # unless the forms actively contradict. Rank 1 is a similarity
            # score over the whole file — one number that would otherwise
            # speak for every symbol in it — so it must positively agree.
            cand = next(
                (t for t in bucket if _stamp_allowed(t[0], mine, t[4])), None
            )
            if cand is None:
                continue
            _rank, _at, sid, why, _cform = cand
            if row is None:
                if not repo_key:
                    continue
                row = CodeShape(
                    project_id=project_id, repo_key=repo_key,
                    path=path, symbol=name, kind=kind,
                )
                session.add(row)
                by_key[(name, kind)] = row
            elif not (row.status in _MECHANICAL_TODO or row.classified_by == "hook"):
                continue  # a judgment — or the canon itself — stands
            # A proposal, never a status: the writer decides.
            row.proposed_snippet_id = sid
            row.proposal_basis = "reference" if _rank >= 2 else "semantic"
            row.proposal_score = resembles.get(sid) if _rank < 2 else 1.0
            row.proposal_group = None
            suggested.append({
                "path": path, "symbol": name, "kind": kind,
                "snippet_id": sid, "reason": why,
            })
            # Every pulled canon the payload NAMES is a uses edge — the
            # call-site fact, independent of which one the row is judged to be.
            await record_uses(
                session, row,
                # Indexed, not destructured: the bucket tuples carry the
                # canon's form as a fifth field, and a positional unpack breaks
                # silently whenever they widen. The unit lane never reaches
                # `record_uses`; only the integration lane would catch it.
                [t[2] for t in bucket if t[0] == 2],
                basis="hook", evidence="write path: pulled the snippet, payload names its symbol",
            )
        if suggested:
            await session.commit()
    return suggested


# --- the mechanical proposer: the machine proposes, judgment classifies
#
# Runs inside the coverage refresh — the one moment the shape BODIES exist
# (the archive is in memory; the ledger stores fingerprints, never code).
# Over every live unclassified row whose content changed since it was last
# examined, it tries, strongest first:
#
#   symbol     the shape bears a recorded canon's symbol at another location
#              (a second definition of the canon's name — an instance or a
#              duplicate to consolidate; either way, it answers to #N);
#   text       whitespace-insensitive containment either way between the
#              body and the canon's recorded code (git-grep, in effect —
#              a verbatim copy, or the canon's own call-site example);
#   reference  the body calls/uses a canon's symbol — the call-site shape;
#   signature  a sym whose definition line, name blanked, resembles the
#              canon's (the move_*/resequence family shape);
#   semantic   the widest net: the body (concept-queried, as the write-path
#              arm does) scores above the write-path threshold against a
#              canon — capped per refresh, so the cost is bounded and a big
#              ledger is worked through across refreshes.
#
# A hit is a PROPOSAL on the row (proposed_snippet_id/basis/score), never a
# classification: an agent confirms in batches (confirm_proposals) or judges
# otherwise (classify_shapes clears it). Rows with no canon hit are grouped
# by the derive-first rule: the same body fingerprint in ≥2
# places, or the same name defined in ≥3 files, is "a repeating shape with
# no canon — derive one first", carried as proposal_basis="derive" + a group
# key so sessions see the consolidation candidates as one thing.

# Derive-first floors. Identical bodies twice is already a copy; a bare name
# needs more repetition before it reads as a family (setup/handler/register
# recur by convention, not by duplication).
_DERIVE_MIN_DUP = 2
_DERIVE_MIN_NAME = 3
# CSS is never grouped by body: classes for different purposes
# share declarations because the style system makes them alike — `.text-muted`
# and `.pin-badge-auto` carrying the same `color: var(--fs-text-tertiary)` are
# two meanings, not two copies. A CSS family is a NAME defined in more than
# one file: that is a recipe living in several places, and two is already
# the signal (a class name is deliberate in a way `setup`/`load` are not).
_DERIVE_MIN_NAME_CSS = 2
# Semantic checks per repo per refresh — an embedding each (local fastembed),
# bounded so a 4,000-row ledger is worked through over refreshes, not in one.
_SEMANTIC_CAP = 150
# The semantic arm has NO floor of its own: it uses the write-path surface's,
# because it asks the write-path hint's question — "which canon does this body
# resemble?" — of the same snippet documents.
#
# A higher floor (0.8) would be sized for a proposal nobody reads, and that is
# not how proposals are consumed: a judge reviews each semantic one
# (confirm_proposals refuses an unnamed batch, and the tool says "semantic
# deserves a look"). Judged instances of a canon typically score 0.68–0.71
# against it, so 0.8 proposes nothing. A wrong proposal costs the judge a
# look; a floor that yields nothing costs the signal. Noise in this band is
# expected — the score rides on the proposal so the judge can weigh it.
#
# How many hits the arm scans for an ALLOWED canon. Wider than the hint's
# budget because disallowed snippets (other language families) share the
# ranking: a true instance can sit 4th, behind three it could not use.
_SEMANTIC_LIMIT = 8
# A MISS FROM THIS ARM IS NOT EVIDENCE, and nothing may read it as one.
# Letting "compared, nothing cleared the floor" withdraw a divergence prompt
# silences real divergences exactly as it silences helpers: many judged
# instances of a canon score below any floor that keeps noise out (0.66 and
# under), and helpers score 0.66–0.75 against snippets they have nothing to do
# with. A code body against a prose-forward snippet document measures the
# wrong field; no floor separates the two bands. The arm PROPOSES on a hit;
# its silence says nothing.
#
# THE FAMILY ARM. The arm above is held to the shape's own project and
# language family, because across projects it is noise. A family idea is the
# exception: an idea shared across projects ON PURPOSE, whose reference may be
# in another language (a Python throttle is an instance of a Go one). So the
# arm also proposes the reference of a canon idea that reaches the shape's
# project through a shared platform — any language, any project — under a
# stricter rule of its own, calibrated on cross-project pairs:
#
#   true pairs   A py→go 0.728 (next 0.703)   B py→go 0.715 (next 0.675)
#                D kt→kt 0.816 (next 0.734)   E go→go 0.610 (next 0.604)
#   no reference N1 0.666  N2 0.623  N3 0.626  N4 0.715  N5 0.614 (best hit,
#                none of them a family reference); unrelated hits to 0.774
#
# No floor separates those bands — the false band reaches 0.715–0.774, a
# true pair sits at 0.610. RANK does: every true pair was the single best
# snippet the user can read. So a family reference is proposed only when it
# is the TOP hit over everything readable (not merely the first allowed one,
# as above) and clears _FAMILY_FLOOR: A, B, D proposed; E missed; no
# negative proposed. A miss is still not evidence — the judge can classify
# against the idea by idea_id.
_FAMILY_FLOOR = 0.70


def _semantic_priority(row) -> tuple:
    """Order for the capped semantic pass: new shapes first.

    The pass is capped (`_SEMANTIC_CAP`), and `flag_divergence` acts only on
    shapes NEW since the previous refresh — and skips one the proposer matched
    to the dominant canon. In plain row order the new shapes are the highest
    ids, the back of the queue, so whenever the backlog exceeds the cap (every
    `_PROPOSER_VERSION` bump queues the whole todo at once) they would be
    flagged before the arm could propose for them.

    So: the human todo (`unclassified`) before `scoped`, newest first within
    each. A row with no creation time sorts last in its group.
    """
    created = row.created_at.timestamp() if row.created_at is not None else float("-inf")
    return (row.status != "unclassified", -created)
# Bump when a basis's rule changes: rows remember the (body, ruleset) they
# were examined under, so a tightened rule re-examines everything once. A
# change that does not alter what is proposed needs no bump. A stale "miss"
# basis on an old row is inert: a miss is not evidence (see _SEMANTIC_LIMIT).
_PROPOSER_VERSION = 9
# Signature resemblance floor, name blanked (difflib ratio) — and a length
# floor, because `def NAME():` resembles `def NAME(x):` at 0.95 while saying
# nothing; a family shape has parameters to resemble.
#
# A proposal is read before it is confirmed, so this admits the band where
# true instances live rather than the band where nothing is wrong. Over
# judged instance rows against every same-family canon, true instances score
# 1.0 or ~0.795 (a model class whose mixin list differs from the canon's), and
# 0.8 cuts the whole second band. At 0.75 all
# of them clear; the price is `async def NAME(user_id: int, …) -> dict`
# look-alikes (35 wrong-canon pairings vs 14 at 0.8), which read as wrong at
# a glance. Below 0.75 the next true row is at 0.6 behind 150+ wrong ones.
_SIGNATURE_FLOOR = 0.75
_SIGNATURE_MIN_LEN = 30
# Textual containment needs enough substance to mean anything.
_TEXT_FLOOR = 40
_BASIS_ORDER = ("symbol", "text", "reference", "signature")


class Canon(NamedTuple):
    snippet_id: int
    kind: str
    symbol: str
    locations: tuple[tuple[str, str], ...]
    signature: str
    code_norm: str
    project_id: int = 0
    language: str = ""   # the snippet's recorded language; "" = unknown, no gate


# Language families: the sym bases only propose within one. Cross-language
# hits are wrong — a Python tool-module canon named `register` would be
# proposed for Vue `handleSubmit`s that call `authStore.register()`, a TS
# store's `register` would match it by symbol, and TS canon would be proposed
# for Python bodies by resemblance. CSS is its own kind and is not gated here.
_FAMILY_BY_LANG = {
    "python": "py", "py": "py",
    "typescript": "js", "ts": "js", "tsx": "js", "javascript": "js", "js": "js",
    "jsx": "js", "vue": "js", "mjs": "js", "cjs": "js",
    "css": "css", "scss": "css", "sass": "css", "less": "css",
    "bash": "sh", "sh": "sh", "shell": "sh", "zsh": "sh",
    "sql": "sql",
    "go": "go", "golang": "go",
    "rust": "rs", "rs": "rs",
    "java": "jvm", "kotlin": "jvm", "kt": "jvm",
    "swift": "swift", "dart": "dart",
    "c": "c", "cpp": "c", "c++": "c", "h": "c",
    "csharp": "cs", "c#": "cs", "cs": "cs",
    "ruby": "rb", "rb": "rb", "php": "php",
}
_FAMILY_BY_EXT = {
    ".py": "py", ".pyi": "py",
    ".ts": "js", ".tsx": "js", ".js": "js", ".jsx": "js", ".vue": "js", ".mjs": "js", ".cjs": "js",
    ".css": "css", ".scss": "css", ".sass": "css", ".less": "css",
    ".sh": "sh", ".bash": "sh", ".zsh": "sh",
    ".sql": "sql",
    ".go": "go", ".rs": "rs",
    ".java": "jvm", ".kt": "jvm", ".kts": "jvm",
    ".swift": "swift", ".dart": "dart",
    ".c": "c", ".h": "c", ".cc": "c", ".cpp": "c", ".hpp": "c",
    ".cs": "cs", ".rb": "rb", ".php": "php",
}


def language_family(language: str) -> str:
    """The family a recorded snippet language belongs to ("" when unknown)."""
    return _FAMILY_BY_LANG.get((language or "").strip().lower(), "")


def path_family(path: str) -> str:
    """The family a file path belongs to, by extension ("" when unknown)."""
    p = (path or "").lower()
    for ext, fam in _FAMILY_BY_EXT.items():
        if p.endswith(ext):
            return fam
    return ""


def same_family(path: str, canon_language: str) -> bool:
    """A sym basis may propose this canon for this path: both families known
    and equal, or either unknown (no evidence either way → no gate)."""
    a = path_family(path)
    b = language_family(canon_language)
    return not a or not b or a == b


# Reference basis: generic verbs name too many unrelated things to count a
# bare mention as a call site of THIS canon (`register`, `load`, `save` …).
# The symbol basis still catches a second definition of such a name.
_REFERENCE_STOPLIST = frozenset({
    "get", "set", "put", "post", "load", "save", "run", "main", "init", "setup",
    "register", "restore", "reset", "toggle", "close", "open", "submit", "handler",
    "update", "create", "delete", "remove", "add", "start", "stop", "send",
    "receive", "render", "mount", "dispatch", "call", "apply", "execute",
})


def _norm_text(text: str) -> str:
    return " ".join((text or "").split())


def signature_similarity(sig_a: str, name_a: str, sig_b: str, name_b: str) -> float:
    """How alike two definition lines are once their own names are blanked —
    `def move_event(project_id, event_id, after_id)` against
    `def move_beat(project_id, beat_id, after_id)` reads high."""
    a = _norm_text(sig_a)
    b = _norm_text(sig_b)
    if name_a:
        a = a.replace(name_a, "NAME")
    if name_b:
        b = b.replace(name_b, "NAME")
    if len(a) < _SIGNATURE_MIN_LEN or len(b) < _SIGNATURE_MIN_LEN:
        return 0.0
    return difflib.SequenceMatcher(None, a, b).ratio()


def text_contains(body: str, code: str) -> bool:
    """Whitespace-insensitive containment, either way, above a substance
    floor — the proposer's git-grep."""
    a = _norm_text(body)
    b = _norm_text(code)
    if len(a) < _TEXT_FLOOR or len(b) < _TEXT_FLOOR:
        return False
    return a in b or b in a


def _in_canon_home(canon: Canon, path: str) -> bool:
    """Is ``path`` in the canon's own package? A location that names a file
    makes its directory home, exactly (a Go package is one directory); a
    location that names a directory is home with everything under it."""
    from grimoire.services.snippets import _path_touches

    here = posixpath.dirname((path or "").strip("/"))
    for lp, _ls in canon.locations:
        lp = (lp or "").strip("/")
        if not lp:
            continue
        if "." in lp.rsplit("/", 1)[-1]:
            if posixpath.dirname(lp) == here:
                return True
        elif _path_touches(lp, path):
            return True
    return False


def names_canon(
    kind: str, path: str, body: str, canon: Canon, *,
    project_id: int = 0, defined_at: Mapping[str, frozenset[str]] | None = None,
) -> bool:
    """Does this body name THIS canon, rather than merely contain its word?

    The by-name test behind the reference basis and every uses edge the
    proposer writes. ``defined_at`` maps each name to the repo paths that
    define it. A mention counts only when it can resolve to the canon:
    - never across projects: one repo does not call a function defined in
      another's, however alike the names are;
    - never a name the file defines itself (a module's own `_run` helper),
      unless the canon is recorded in this very file;
    - when the repo also defines the name somewhere the canon is not
      recorded (`sftpRun.file` beside the canon `importRun.file`), only from
      the canon's own package: elsewhere the name says nothing about which
      definition a call reaches;
    - never a generic verb, with or without a leading underscore;
    - a bare lowercase word (`file`, `seen`, `seal`) only where it is called
      as `word(`, because as a plain word it is everyone's variable. No space
      is allowed before the parenthesis: formatted code never has one, and
      prose does ("a file (not a directory)").
    Same kind and language family are checked by the callers' loops."""
    sym = _norm_symbol(canon.symbol or "")
    if not sym:
        return False
    if project_id and canon.project_id and canon.project_id != project_id:
        return False
    if sym.lstrip("_").lower() in _REFERENCE_STOPLIST:
        return False
    where = (defined_at or {}).get(sym, frozenset())
    if where:
        from grimoire.services.snippets import _path_touches

        def at_canon(p: str) -> bool:
            return any(lp and _path_touches(lp, p) for lp, _ls in canon.locations)

        if path in where and not at_canon(path):
            return False
        if canon.locations and not all(at_canon(p) for p in where) \
                and not _in_canon_home(canon, path):
            return False
    if kind == "sym" and re.fullmatch(r"[a-z]+", sym):
        return re.search(rf"(?<![\w$]){re.escape(sym)}\(", body or "") is not None
    return references_symbol(body, sym, kind)


def reference_canons(
    kind: str, path: str, symbol: str, body: str, canons: Iterable[Canon], *,
    project_id: int = 0, defined_at: Mapping[str, frozenset[str]] | None = None,
) -> list[int]:
    """Every canon this body NAMES — the uses edges the proposer can
    write mechanically: same kind, same language family, not the shape's own
    name, and named in the sense `names_canon` gives."""
    norm_sym = _norm_symbol(symbol)
    out: list[int] = []
    for c in canons:
        if c.kind != kind or not c.symbol:
            continue
        if kind == "sym" and not same_family(path, c.language):
            continue
        if _norm_symbol(c.symbol) == norm_sym:
            continue
        if names_canon(kind, path, body, c, project_id=project_id, defined_at=defined_at):
            out.append(c.snippet_id)
    return out


def match_canon(
    kind: str, path: str, symbol: str, signature: str, body: str,
    canons: Iterable[Canon], *, project_id: int = 0,
    defined_at: Mapping[str, frozenset[str]] | None = None,
) -> tuple[int, str, float] | None:
    """The strongest (snippet_id, basis, score) a shape earns against the
    canon catalog, by the non-semantic bases — or None. Strongest by
    basis order, then by score within the basis; on a tie, a canon recorded
    in the shape's own project beats family canon from another (the same
    helper recorded in two projects is the shape's own first)."""
    best: dict[str, tuple[float, bool, int]] = {}

    def offer(basis: str, score: float, canon: Canon) -> None:
        cur = best.get(basis)
        same = bool(project_id) and canon.project_id == project_id
        if cur is None or (score, same) > (cur[0], cur[1]):
            best[basis] = (score, same, canon.snippet_id)

    norm_sym = _norm_symbol(symbol)
    for c in canons:
        if c.kind != kind:
            continue
        if kind == "sym" and not same_family(path, c.language):
            continue  # a Python canon says nothing about a Vue body, and vice versa
        if c.symbol and _norm_symbol(c.symbol) == norm_sym:
            if not any(location_covers(lp, ls, path, symbol) for lp, ls in c.locations):
                offer("symbol", 1.0, c)
            continue  # its own location is canonical territory, not a proposal
        if names_canon(kind, path, body, c, project_id=project_id, defined_at=defined_at):
            offer("reference", 0.9, c)
        if c.code_norm and text_contains(body, c.code_norm):
            offer("text", 0.95, c)
        if kind == "sym" and c.signature:
            ratio = signature_similarity(signature, symbol, c.signature, c.symbol)
            if ratio >= _SIGNATURE_FLOOR:
                offer("signature", round(ratio, 3), c)
    for basis in _BASIS_ORDER:
        if basis in best:
            score, _same, sid = best[basis]
            return (sid, basis, score)
    return None


async def canon_catalog(user_id: int) -> list[Canon]:
    """Every snippet this user can browse, as matchable canon — own projects
    and shared ones alike, because family canon counts."""
    from grimoire.models.note import Note
    from grimoire.services.access import browsable_notes_clause
    from grimoire.services.coverage import extract_definitions
    from grimoire.services.snippets import SNIPPET_NOTE_TYPE, snippet_fields

    async with async_session() as session:
        notes = (
            await session.execute(
                select(Note).where(
                    browsable_notes_clause(user_id),
                    Note.note_type == SNIPPET_NOTE_TYPE,
                    Note.deleted_at.is_(None),
                )
            )
        ).scalars().all()
    out: list[Canon] = []
    for note in notes:
        fields = snippet_fields(note)
        symbol = (fields.get("symbol") or "").strip()
        kind = snippet_kind(symbol, fields.get("language") or "")
        code = fields.get("code") or ""
        # The signature basis matches against the canon's OWN definition
        # line — never against an example in its code. A snippet whose code
        # opens with a call-site example (confirmed()'s does) would otherwise
        # make every `async function x(): Promise<void>` resemble it.
        own = next(
            (d for d in extract_definitions(code)
             if symbol and _norm_symbol(d.name) == _norm_symbol(symbol)),
            None,
        )
        signature = own.signature if own else ""
        out.append(Canon(
            int(note.id), kind, symbol,
            tuple(
                ((loc.get("path") or ""), (loc.get("symbol") or ""))
                for loc in fields.get("locations") or []
            ),
            signature, _norm_text(code), int(note.project_id or 0),
            (fields.get("language") or "").strip().lower(),
        ))
    return out


def _clear_proposal(row: CodeShape, *, reexamine: bool = False) -> None:
    row.proposed_snippet_id = None
    row.proposal_basis = None
    row.proposal_score = None
    row.proposal_group = None
    if reexamine:
        row.proposed_at = None
        row.proposed_sha = ""


def _substance(text: str) -> int:
    return len("".join((text or "").split()))


def pick_semantic(
    hits: list[tuple[float, int]], allowed: set[int], family: set[int], *, floor: float,
) -> tuple[int, float, str] | None:
    """Which ranked hit the semantic arm proposes, as (snippet_id, score,
    basis). Pure; ``hits`` are (score, snippet_id) over everything the user
    can read, best first. The TOP hit may be a family reference (basis
    "family", at _FAMILY_FLOOR); any hit at ``floor`` may be an own-project
    canon (basis "semantic")."""
    for rank, (score, sid) in enumerate(hits):
        if rank == 0 and sid in family and score >= _FAMILY_FLOOR:
            return sid, round(float(score), 3), "family"
        if sid in allowed and score >= floor:
            return sid, round(float(score), 3), "semantic"
    return None


async def _semantic_canon(
    user_id: int, body: str, allowed: set[int], family: set[int] | None = None,
) -> tuple[int, float, str] | None:
    """The canon this body MEANS, as (snippet_id, score, basis), or None.
    None is "no proposal", never "not the canon" — see the note above
    `_SEMANTIC_LIMIT`; the family arm's rule is the note above
    `_FAMILY_FLOOR`."""
    from grimoire.services.embeddings import semantic_search_notes
    from grimoire.services.plugin_context import (
        WRITEPATH_DEFAULT_THRESHOLD, WRITEPATH_MIN_CODE_CHARS, concept_query,
    )

    family = family or set()
    if _substance(body) < WRITEPATH_MIN_CODE_CHARS or not (allowed or family):
        return None
    query = concept_query(body) or body
    hits = await semantic_search_notes(
        user_id, query, limit=_SEMANTIC_LIMIT,
        threshold=min(WRITEPATH_DEFAULT_THRESHOLD, _FAMILY_FLOOR),
        note_type="snippet", scope="browse",
    )
    return pick_semantic(
        [(float(score), int(note.id)) for score, note in hits], allowed, family,
        floor=WRITEPATH_DEFAULT_THRESHOLD,
    )


async def propose_for_repo(
    user_id: int,
    project_id: int,
    repo_key: str,
    definitions: list,
    *,
    canons: list[Canon] | None = None,
    semantic_cap: int = _SEMANTIC_CAP,
) -> dict:
    """Examine one repo's live unclassified rows against canon and record
    proposals. ``definitions`` are the ArchiveShape records the sync just
    upserted (their bodies are the matching material). Rows whose fingerprint
    is unchanged since their last examination are skipped; rows that only
    the capped semantic pass could not reach stay unexamined, so the next
    refresh reaches the next slice. Returns counts."""
    if canons is None:
        canons = await canon_catalog(user_id)
    by_key = {(d[0], d[1], d[2]): d for d in definitions}
    # Where the repo defines each name: a mention resolves to the definition
    # in its own file, and a name defined in several packages only from the
    # canon's own (see names_canon).
    paths_by_name: dict[str, set[str]] = {}
    for d in definitions:
        if d[1] == "sym":
            paths_by_name.setdefault(_norm_symbol(d[2]), set()).add(d[0])
    defined_at = {n: frozenset(p) for n, p in paths_by_name.items()}
    # The semantic arm is the widest net and, across projects, is noise: it
    # is held to the shape's own project and language family. The precise
    # bases (symbol/text) still reach family canon in other projects.
    sym_canons = [c for c in canons if c.kind == "sym" and c.project_id == project_id]

    def semantic_allowed(path: str) -> set[int]:
        return {c.snippet_id for c in sym_canons if same_family(path, c.language)}
    # The family arm: references of canon ideas on a platform this project
    # shares, in any language (see _FAMILY_FLOOR). A project on no shared
    # platform gets none, so nothing is proposed across to it.
    try:
        from grimoire.services import family_adoption

        family_refs = await family_adoption.family_reference_ids(project_id)
    except Exception:
        logger.warning("family references unreadable for project %s", project_id, exc_info=True)
        family_refs = set()
    now = datetime.now(timezone.utc)
    examined = proposed = checked = 0
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.repo_key == repo_key,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        semantic_todo: list[tuple[CodeShape, object]] = []
        for row in rows:
            d = by_key.get((row.path, row.kind, row.symbol))
            if d is None:
                continue
            signature, body_sha, body = d[3], d[4], d[5]
            examined_as = f"{body_sha}@{_PROPOSER_VERSION}"
            if row.proposed_at is not None and row.proposed_sha == examined_as:
                continue
            if row.status not in _MECHANICAL_TODO:
                # A judged row gets no proposal — but its uses edges
                # are a fact about the body, judged or not: the consumer map
                # of a canon must include the call sites someone already
                # classified. Mark it examined so the scan runs once per body.
                used = reference_canons(row.kind, row.path, row.symbol, body, canons,
                                        project_id=project_id, defined_at=defined_at)
                await replace_reference_uses(session, row, used,
                                             evidence="proposer: body names the canon's symbol")
                row.proposed_at = now
                row.proposed_sha = examined_as
                continue
            examined += 1
            group = row.proposal_group  # derive grouping is reassigned below
            hit = match_canon(
                row.kind, row.path, row.symbol, signature, body, canons,
                project_id=project_id, defined_at=defined_at,
            )
            _clear_proposal(row)
            row.proposal_group = group
            row.proposed_at = now
            row.proposed_sha = examined_as
            # Consumption is recorded for every canon the body names,
            # whatever the row is then judged to be.
            used = reference_canons(row.kind, row.path, row.symbol, body, canons,
                                    project_id=project_id, defined_at=defined_at)
            await replace_reference_uses(session, row, used,
                                         evidence="proposer: body names the canon's symbol")
            if hit:
                row.proposed_snippet_id, row.proposal_basis, row.proposal_score = hit
                row.proposal_group = None
                proposed += 1
            elif row.kind == "sym":
                semantic_todo.append((row, d))
        semantic_todo.sort(key=lambda item: _semantic_priority(item[0]))
        for i, (row, d) in enumerate(semantic_todo):
            if i >= semantic_cap:
                # Not reached this refresh: leave it unexamined so the next
                # refresh picks it up, rather than stamping a false "nothing".
                row.proposed_at = None
                row.proposed_sha = ""
                continue
            checked += 1
            try:
                found = await _semantic_canon(
                    user_id, d[5], semantic_allowed(row.path), family_refs)
            except Exception:
                logger.warning("semantic proposal failed", exc_info=True)
                found = None
            if found:
                row.proposed_snippet_id, row.proposal_score, row.proposal_basis = found
                row.proposal_group = None
                proposed += 1
        await session.commit()
    return {"examined": examined, "proposed": proposed, "semantic_checked": checked}


def derive_groups(
    rows: Iterable[tuple[str, str, str, str]]
) -> dict[tuple[str, str, str], str]:
    """The derive-first grouping over (path, kind, symbol, body_sha) rows
    that matched no canon: {(path, kind, symbol): group_key}. For code
    (kind `sym`) identical bodies in ≥2 places group as `dup:<sha>` and the
    same name defined in ≥3 files groups as `name:sym:<symbol>`, the copy
    before the name. CSS groups by name only — the same class defined in
    ≥2 files is `name:css:<symbol>`; its body never groups it."""
    by_sha: dict[str, list[tuple[str, str, str]]] = {}
    by_name: dict[tuple[str, str], list[tuple[str, str, str]]] = {}
    for path, kind, symbol, sha in rows:
        key = (path, kind, symbol)
        if sha and kind != "css":
            by_sha.setdefault(sha, []).append(key)
        by_name.setdefault((kind, _norm_symbol(symbol)), []).append(key)
    out: dict[tuple[str, str, str], str] = {}
    for sha, keys in by_sha.items():
        if len(set(keys)) >= _DERIVE_MIN_DUP:
            for key in keys:
                out.setdefault(key, f"dup:{sha}")
    for (kind, symbol), keys in by_name.items():
        floor = _DERIVE_MIN_NAME_CSS if kind == "css" else _DERIVE_MIN_NAME
        if len({k[0] for k in keys}) >= floor:
            for key in keys:
                out.setdefault(key, f"name:{kind}:{symbol}")
    return out


async def apply_derive_groups(project_id: int) -> int:
    """Recompute derive-first groups over the project's live unclassified
    rows that carry no canon proposal; returns how many rows are grouped."""
    now = datetime.now(timezone.utc)
    grouped = 0
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.status.in_(_MECHANICAL_TODO),
                    CodeShape.vanished_at.is_(None),
                    CodeShape.proposed_snippet_id.is_(None),
                    # A file row is named by its stem, and `index`, `+page`
                    # and `__init__` repeat by convention, not by copying —
                    # a derive family of them would be pure noise.
                    CodeShape.kind != "file",
                )
            )
        ).scalars().all()
        groups = derive_groups(
            (r.path, r.kind, r.symbol, r.body_sha or "") for r in rows
        )
        sizes: dict[str, int] = {}
        for g in groups.values():
            sizes[g] = sizes.get(g, 0) + 1
        for row in rows:
            key = groups.get((row.path, row.kind, row.symbol))
            if key:
                row.proposal_basis = "derive"
                row.proposal_group = key
                row.proposal_score = float(sizes[key])
                if row.proposed_at is None:
                    row.proposed_at = now
                grouped += 1
            elif row.proposal_group:
                row.proposal_basis = None
                row.proposal_group = None
                row.proposal_score = None
        await session.commit()
    return grouped


def proposal_summary(
    rows: Iterable[CodeShape], *, top: int = 8,
    consumer_paths: dict[int, list[str]] | None = None,
) -> dict:
    """The readout's view of the proposer's standing: how many canon
    proposals await confirmation, and the largest derive-first groups.
    ``consumer_paths`` (shape_id → files whose markup names it, milestone
    302) puts `consumers` on each group — the family's distinct consumer
    files across its members, the datum that separates a shared recipe
    from a scoped convention."""
    proposed = 0
    by_canon: dict[int, int] = {}
    groups: dict[str, dict] = {}
    files: dict[str, set[str]] = {}
    consumers: dict[str, set[str]] = {}
    for row in rows:
        if row.status not in _MECHANICAL_TODO:
            continue
        if row.proposed_snippet_id is not None:
            proposed += 1
            by_canon[row.proposed_snippet_id] = by_canon.get(row.proposed_snippet_id, 0) + 1
        elif row.proposal_group:
            dup = not row.proposal_group.startswith("name:")
            g = groups.setdefault(row.proposal_group, {
                "group": row.proposal_group, "kind": row.kind,
                "label": (
                    f"{row.symbol} (identical body)" if dup
                    else ("." if row.kind == "css" else "") + row.symbol
                ),
                "size": 0, "files": 0, "paths": [],
            })
            g["size"] += 1
            files.setdefault(row.proposal_group, set()).add(row.path)
            if len(g["paths"]) < 3:
                g["paths"].append(row.path)
            if consumer_paths is not None and row.kind == "css":
                consumers.setdefault(row.proposal_group, set()).update(
                    consumer_paths.get(row.id) or ()
                )
    for key, g in groups.items():
        g["files"] = len(files[key])
        if key in consumers:
            g["consumers"] = consumer_summary(consumers[key])
    # Body-identical groups first: the things an audit actually
    # consolidated were identical bodies under different names/files; a
    # name repeated across modules is usually convention. Within a tier,
    # the group spread over more files is the bigger copy.
    ranked = sorted(
        groups.values(),
        key=lambda g: (g["group"].startswith("name:"), -g["files"], -g["size"], g["group"]),
    )
    top_canon = None
    if by_canon:
        sid, n = max(by_canon.items(), key=lambda kv: (kv[1], -kv[0]))
        top_canon = {"snippet_id": sid, "count": n}
    return {"proposed": proposed, "derive_groups": ranked[:top], "top_canon": top_canon}


def derive_new_summary(
    rows: Iterable[CodeShape], *, since: datetime | None, top: int = 3
) -> dict:
    """The arrival-moment drift signal: derive-grouped rows FIRST
    SEEN after ``since`` — the previous refresh's stamp, the same one
    flag_divergence uses. "Since the last refresh, N more copies joined a
    duplicate family" is the sentence that makes the derive queue a thing
    you notice on entering, not a thing an audit finds. ``since`` None (a
    first seed) means nothing is new. Judged rows never count."""
    if since is None:
        return {"count": 0, "examples": []}
    fresh = [
        r for r in rows
        if r.proposal_basis == "derive" and r.proposal_group
        and r.status in _MECHANICAL_TODO and r.vanished_at is None
        and r.created_at is not None and r.created_at > since
    ]
    fresh.sort(key=lambda r: r.created_at, reverse=True)
    return {
        "count": len(fresh),
        "examples": [
            {"label": ("." if r.kind == "css" else "") + r.symbol,
             "path": r.path, "group": r.proposal_group}
            for r in fresh[:top]
        ],
    }


async def confirm_proposals(
    user_id: int,
    project_id: int,
    *,
    snippet_id: int = 0,
    path: str = "",
    basis: str = "",
    min_score: float = 0.0,
) -> dict:
    """Turn reviewed canon proposals into `instance` rows, in one batch.

    At least one of snippet_id / path / basis must narrow the batch — "confirm
    everything proposed" without having looked is not a judgment. Each row
    becomes instance-of-its-proposed-snippet, classified_by="agent", reason
    naming the basis and score; the proposal is retired. Returns
    {"confirmed": N}."""
    from grimoire.services import access

    if not (snippet_id or path.strip() or basis.strip()):
        raise ValueError(
            "name what you reviewed: confirm by snippet_id, path, and/or basis"
        )
    if not await access.can_write_project(user_id, project_id):
        raise ValueError(f"project {project_id} not found or no write access")
    from sqlalchemy import or_

    conds = [
        CodeShape.project_id == project_id,
        CodeShape.status.in_(_MECHANICAL_TODO),
        CodeShape.vanished_at.is_(None),
        CodeShape.proposed_snippet_id.isnot(None),
    ]
    if snippet_id:
        conds.append(CodeShape.proposed_snippet_id == snippet_id)
    if path.strip():
        clean = path.strip().strip("/")
        conds.append(or_(CodeShape.path == clean, CodeShape.path.like(clean + "/%")))
    if basis.strip():
        conds.append(CodeShape.proposal_basis == basis.strip())
    now = datetime.now(timezone.utc)
    confirmed = 0
    touched: set[int] = set()
    async with async_session() as session:
        rows = (await session.execute(select(CodeShape).where(*conds))).scalars().all()
        for row in rows:
            if (row.proposal_score or 0.0) < min_score:
                continue
            touched.add(row.proposed_snippet_id)
            await _judge(
                session, row, status="instance", snippet_id=row.proposed_snippet_id,
                by="agent", at=now,
                reason=(
                    f"confirmed {row.proposal_basis} proposal"
                    f" ({(row.proposal_score or 0.0):.2f})"
                ),
            )
            confirmed += 1
        await session.commit()
    out = {"confirmed": confirmed}
    moved = await _sync_family(user_id, project_id, touched)
    if moved:
        out["family"] = moved
    return out


# --- the divergence readout: button B where button A is canon --------------
#
# Three answers the ledger can now give mechanically:
#   DIVERGENCE  a shape NEW since the previous refresh, in a directory+kind
#               where one canon dominates the judged siblings, that the
#               proposer did not match to that canon → `diverges_from=#N`.
#               Read: "button B appeared where button A is canon — divergence
#               or variant? classify it." Surfaced in the coverage readout and
#               in-band at write time (the prior-art hook names the shapes).
#   HISTORY     every judgment / vanish / reappearance / drift is an event;
#               shape_history answers "what was used here, when, and why".
#   RECHECK     an instance/variant whose body moved since it was judged is
#               flagged recheck_at (sync) — the judgment stands, re-confirm it.

# How much a payload must resemble a canon before the hook SUGGESTS it as the
# shape's canon — and the bar below which a stored hook stamp (an `instance`
# row written unattended, classified_by="hook") is weak (`is_weak_stamp`).
# The hook itself never asserts a classification.
#
# WHY A FLOOR AT ALL. Without one, any score the semantic arm produced would
# count, 0.69 as confidently as 0.95. A stamp feeds `dominant_canon`, which
# tells the next writer in that directory what to build from, so a thin stamp
# does not stay thin, it compounds.
#
# 0.80 rather than the retrieval floors near 0.70, deliberately. Those bars
# decide whether to SHOW someone a record, where being wrong costs a glance.
# This one weighs a claim about the codebase, where being wrong costs a wrong
# instruction to everyone who writes in that directory afterwards. A claim
# should need more evidence than a suggestion to read, not the same.
_RESEMBLE_MIN = 0.80


def is_weak_stamp(row) -> bool:
    """Was this row written by the hook, unattended, on a resemblance below
    the floor it would have to clear today?

    ONE PREDICATE FOR "WEAK", read by the review surface that lists these rows
    and by every reader that would otherwise take them as evidence. Rows
    stamped below the floor stay stored, and until an agent re-judges them
    they must not speak. They are not withdrawn here — `stamps_to_review`
    lists them for judgment — they are simply not counted, which is the
    module's rule for evidence it cannot trust: quieter, never more confident.
    """
    if getattr(row, "classified_by", None) not in _UNATTENDED_BY:
        return False
    score = stamp_score(getattr(row, "reason", None))
    return score is not None and score < _RESEMBLE_MIN


# A canon dominates a directory+kind when at least this many siblings are
# judged (canonical/instance) and this share of them answer to one snippet.
_DENSITY_MIN_JUDGED = 3
_DENSITY_SHARE = 0.6


def comparable_siblings(rows: Iterable, form: str) -> list:
    """The siblings a candidate of ``form`` can honestly be counted against.

    Excludes only a KNOWN family contradiction, via `families_conflict` — the
    same predicate the divergence gate uses, so the denominator and the gate
    cannot drift into disagreeing about what "comparable" means.

    `form` is a FORM (`fn`, `async-fn`, `type`, …), never a family:
    `families_conflict` coarsens both sides itself, and handing it a family
    makes `shape_family("callable")` return "" so nothing is excluded — a
    narrowing that silently becomes a no-op while still reading as applied.

    AN UNREADABLE SIBLING STAYS COUNTED, and that direction is the point.
    Dropping it would shrink `judged`, raise the dominant canon's share, and
    make the check fire MORE on the directories it can read least. Every
    unknown-form decision in this module goes the same way: quieter, never
    more confident.
    """
    if not form:
        return list(rows)
    return [
        r for r in rows
        if not families_conflict(
            shape_form(getattr(r, "signature", "") or "", getattr(r, "kind", "sym")),
            form,
        )
    ]


def dominant_canon(rows: Iterable[CodeShape]) -> tuple[int, int, int] | None:
    """(snippet_id, its_count, judged_count) when one canon dominates these
    sibling rows (same directory + kind), else None."""
    counts: dict[int, int] = {}
    judged = 0
    for r in rows:
        # A weak stamp is not a judgment: counted, it could let a YAML CI
        # snippet "dominate" a Svelte directory, and every write there would
        # be told it diverged from it.
        if is_weak_stamp(r):
            continue
        if r.status in ("canonical", "instance") and r.snippet_id is not None:
            judged += 1
            counts[r.snippet_id] = counts.get(r.snippet_id, 0) + 1
    if judged < _DENSITY_MIN_JUDGED or not counts:
        return None
    sid, n = max(counts.items(), key=lambda kv: (kv[1], -kv[0]))
    if n / judged < _DENSITY_SHARE:
        return None
    return sid, n, judged


def _dir_of(path: str) -> str:
    return path.rsplit("/", 1)[0] if "/" in path else ""


async def canon_density(
    project_id: int, path: str, kind: str, form: str = ""
) -> tuple[int, int, int, str] | None:
    """The dominant canon for the directory ``path`` sits in, for ``kind`` —
    the write-time question "is this a canon-dense place?".

    ``form`` is the candidate's own FORM — `fn`, `async-fn`, `type`, … as
    `shape_form` returns it, NOT a family. `families_conflict` coarsens both
    sides itself, and handing it a family makes `shape_family("callable")`
    return "" so nothing is ever excluded: the narrowing silently becomes a
    no-op that still reads as applied. It narrows the DENOMINATOR. Without
    it the base rate is computed over every code symbol in the directory as
    one bucket, counting a dataclass, a constant and an async service unit as
    three comparable things, and the share that comes out is a statement about
    a population nobody asked a question about.

    EXCLUDES ONLY A KNOWN CONFLICT, using `families_conflict` — the same
    predicate the divergence gate uses, so the two cannot drift apart. A
    sibling whose form is unreadable STAYS COUNTED. That direction is
    deliberate and it is the one that matters: dropping unknown rows would
    shrink `judged`, raise the share, and make the check fire MORE on exactly
    the directories it can read least. Every other unknown-form decision in
    this module goes the same way — quieter, never more confident.

    NOT narrowed to the candidate's exact form, for the reason `shape_family`
    gives at length: `fn` beside `async-fn` is a fair question, not noise.
    Bucketing the denominator by form would take the async canon out of a
    sync candidate's count and silence that flag — the inversion a form gate
    makes, one layer down.
    """
    directory = _dir_of(path)
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.kind == kind,
                    CodeShape.vanished_at.is_(None),
                    CodeShape.path.like(directory + "/%") if directory
                    else CodeShape.path.notlike("%/%"),
                )
            )
        ).scalars().all()
    siblings = [r for r in rows if _dir_of(r.path) == directory]
    siblings = comparable_siblings(siblings, form)
    dom = dominant_canon(siblings)
    if dom is None:
        return None
    # The canon's form travels with the count, read from its OWN judged rows.
    # `dominant_canon` keeps its three-value shape: it answers "what dominates
    # here", which is still a true and separately useful question, and its
    # test pins that arithmetic. Callers act on the count together with the
    # canon's form, never on the count alone.
    return dom[0], dom[1], dom[2], canon_form(siblings, dom[0])


async def write_time_divergence(
    project_id: int, path: str, shapes: list[tuple[str, str]], suggested: list[dict],
    code: str = "",
) -> list[dict]:
    """The in-band check for the shapes the hook named at ``path``: for each
    kind whose directory has a dominant canon, the named shapes that are
    not that canon's instance/canonical (judged, or just suggested as it) — new or
    unclassified rows only; a judged shape is not re-litigated at every
    edit. Returns [{symbol, kind, canon_snippet_id, instances, judged}]."""
    just_stamped = {(s["symbol"], s["kind"]): s["snippet_id"] for s in suggested}
    out: list[dict] = []
    if not shapes:
        return out
    # DENSITY IS PER CANDIDATE, not per kind: the denominator
    # excludes siblings whose family contradicts what is being written, so it
    # cannot be computed until the candidate's own form is known. Cached on
    # (kind, family) — a write names a handful of shapes and they collapse to
    # one or two buckets, so this stays one or two queries.
    #
    # The cost is that the row load below does not sit behind an early exit
    # on "nothing is dense here". That is one indexed lookup on
    # (project_id, path), and it has to happen first regardless: the
    # candidate's signature comes from its stored row when the payload does
    # not carry one.
    _density: dict[tuple[str, str], tuple[int, int, int, str] | None] = {}

    async def density_for(kind: str, form: str):
        # Keyed and passed as a FORM, not a family — see `canon_density`.
        key = (kind, form)
        if key not in _density:
            _density[key] = await canon_density(project_id, path, kind, form)
        return _density[key]

    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.path == path,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
    by_key = {(r.symbol, r.kind): r for r in rows}
    for kind, name in shapes:
        row = by_key.get((name, kind))
        # The candidate's own form, read before anything is counted. Signature
        # from the payload first — a shape being written now may have no row
        # yet — falling back to the stored row's.
        mine = shape_form(
            signature_in(code, name, kind) or getattr(row, "signature", "") or "", kind
        )
        dom = await density_for(kind, mine)
        if not dom:
            continue
        sid, n, judged, cform = dom
        if just_stamped.get((name, kind)) == sid:
            continue
        if row is not None and (
            row.status != "unclassified" or row.proposed_snippet_id == sid
        ):
            continue
        # IS THIS EVEN THE SAME CATEGORY OF THING AS THE CANON?
        # Without this the line was a pure base rate: "most things here are X,
        # so be X", with the candidate never examined at all. It told a frozen
        # dataclass to build from the `async_session` service canon.
        #
        # FAMILY, not form, and only on contradiction — see `shape_family`. A
        # divergence prompt is ABOUT a mismatch, so requiring the candidate to
        # match would silence the check precisely where it belongs.
        #
        # `mine` was read above, before the denominator was counted — the
        # same value serves both, and they must not be able to disagree.
        # An unreadable signature produces a fair question rather than a
        # guess, because `families_conflict` needs both sides.
        if families_conflict(mine, cform):
            continue
        out.append({"symbol": name, "kind": kind, "canon_snippet_id": sid,
                    "instances": n, "judged": judged})
    return out


# How many other files a family line names before "…" — enough to go look,
# not a wall.
_DERIVE_FILES_SHOWN = 4


async def write_time_derive(
    project_id: int, path: str, shapes: list[tuple[str, str]]
) -> list[dict]:
    """The in-band DERIVE check: for each (kind, name) the hook
    named at ``path``, what the ledger already knows about that name
    elsewhere in the project —

      family  the name sits in a derive-first group (code: identical body
              in N files or the same name in ≥3; CSS: the same class in
              ≥2 files): "this is a known family with no canon
              — derive it now, don't add a copy";
      canon   a `canonical` row of that name at another path: "this is
              canon #N at <path> — reuse, don't redefine".

    Only for shapes not yet judged at ``path`` (a judged shape is not
    re-litigated at every edit), never for the canon's own file. Returns
    [{symbol, kind, key, family?|canon?}] — `key` is the dedup token the
    hook keeps per session (the group id, or canon:<snippet_id>)."""
    wanted = {(k, _norm_symbol(n)): n for k, n in shapes if n}
    if not wanted:
        return []
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.vanished_at.is_(None),
                    CodeShape.symbol.in_({norm for (_k, norm) in wanted}),
                )
            )
        ).scalars().all()
    out: list[dict] = []
    for (kind, norm), name in wanted.items():
        same = [r for r in rows if r.kind == kind and _norm_symbol(r.symbol) == norm]
        here = next((r for r in same if r.path == path), None)
        if here is not None and here.status not in _MECHANICAL_TODO:
            continue  # judged here (or this IS the canon): nothing to say
        others = [r for r in same if r.path != path]
        label = ("." if kind == "css" else "") + name
        canon = next((r for r in others if r.status == "canonical" and r.snippet_id), None)
        if canon is not None:
            out.append({"symbol": name, "kind": kind, "key": f"canon:{canon.snippet_id}",
                        "canon": {"snippet_id": canon.snippet_id, "path": canon.path,
                                  "label": label}})
            continue
        grouped = [r for r in others if r.proposal_group and r.status in _MECHANICAL_TODO]
        if here is not None and here.proposal_group:
            grouped = [r for r in grouped if r.proposal_group == here.proposal_group] or grouped
        if not grouped:
            continue
        group = grouped[0].proposal_group
        members = [r for r in grouped if r.proposal_group == group]
        files = sorted({r.path for r in members})
        family = {
            "group": group, "label": label,
            "identical": not group.startswith("name:"),
            "files": files[:_DERIVE_FILES_SHOWN], "file_count": len(files),
            "size": len(members) + (1 if here is not None else 0),
        }
        if kind == "css":
            # What renders the family: the members' consumer
            # files, the row at `path` included when it already exists.
            ids = [r.id for r in members] + ([here.id] if here is not None else [])
            edges = await consumers_of(ids)
            family["consumers"] = consumer_summary(
                e.path for es in edges.values() for e in es
            )
        out.append({"symbol": name, "kind": kind, "key": group, "family": family})
    return out


async def new_shapes_at(
    project_id: int, path: str, shapes: list[tuple[str, str]]
) -> list[tuple[str, str]]:
    """The (kind, name) shapes the hook named at ``path`` that the ledger has
    no live row for there — definitions this write introduces, as opposed to
    ones it edits inside. Order follows ``shapes``."""
    wanted = [(k, n) for k, n in shapes if n]
    if not wanted:
        return []
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape.kind, CodeShape.symbol).where(
                    CodeShape.project_id == project_id,
                    CodeShape.path == path,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).all()
    here = {(k, _norm_symbol(s)) for k, s in rows}
    return [(k, n) for k, n in wanted if (k, _norm_symbol(n)) not in here]


async def flag_divergence(project_id: int, *, since: datetime | None) -> int:
    """Flag shapes created after ``since`` (the previous refresh) that sit
    where a canon dominates and were not proposed as that canon. With no
    previous refresh (first seed) nothing is new, nothing is flagged.
    Standing flags persist until judged. Returns how many are flagged."""
    if since is None:
        return 0
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        by_dir: dict[tuple[str, str], list[CodeShape]] = {}
        for r in rows:
            by_dir.setdefault((_dir_of(r.path), r.kind), []).append(r)
        flagged = 0
        for siblings in by_dir.values():
            dom = dominant_canon(siblings)
            # Computed once per directory rather than per row: the canon's
            # form is a property of the canon, not of who is being judged.
            cform = canon_form(siblings, dom[0]) if dom else FORM_UNKNOWN
            for r in siblings:
                if r.status not in _MECHANICAL_TODO:
                    continue
                if r.diverges_from is not None:
                    # A flag is a mechanical prompt resting on one premise —
                    # "this canon dominates here". When the premise no longer
                    # holds (the rows that elected it were weak stamps, or
                    # were re-judged away) the prompt is false, and a false
                    # prompt on every write is a flag nobody reads.
                    # Withdrawing it judges nothing: the row stays
                    # unclassified, and a judgment is never touched here.
                    if dom is None or dom[0] != r.diverges_from:
                        r.diverges_from = None
                        continue
                    flagged += 1
                    continue
                if dom is None or r.created_at is None or r.created_at <= since:
                    continue
                if r.proposed_snippet_id == dom[0]:
                    continue  # the proposer already says "instance of the canon"
                # NOT the converse. A semantic MISS does not mean "not the
                # canon": true instances of a canon often miss the arm's floor
                # too (see the note above _SEMANTIC_LIMIT). A helper beside a
                # canon and a genuine divergence — a sync `confirmDanger`
                # beside an async confirm canon — are the same case to every
                # signal this ledger holds, so the prompt asks, and the judge
                # answers it (`exempt` with a reason retires a false one for
                # good).
                # The same structural test the write-time check applies. The
                # sweep and the hook must agree about what counts
                # as divergence, or an audit contradicts the line the writer
                # was shown at the keyboard.
                if families_conflict(shape_form(r.signature or "", r.kind), cform):
                    continue
                r.diverges_from = dom[0]
                flagged += 1
        await session.commit()
    return flagged


# --- the end-of-turn question: what did this turn write that
# nobody has judged? ---------------------------------------------------------
#
# The write hooks keep a session ledger of the definitions each turn wrote; at
# the end of the turn the client asks this, and the agent that wrote them
# answers. "Judged" means a verdict SOMEONE READ: an agent's, an audit's, an
# import's. The sync's `scoped` stamp, `unclassified`, a hook stamp and no row
# at all are all unjudged — each is a machine's guess or nothing, and the
# point of the question is that the writer, who knows, says.

def _is_judged(row) -> bool:
    return (
        row.status not in _MECHANICAL_TODO
        and row.classified_by not in _UNATTENDED_BY
    )


async def unjudged_shapes(
    project_id: int, written: list[tuple[str, str, str]], *, repo_key: str = "",
) -> list[dict]:
    """The (path, kind, name) shapes in ``written`` that carry no judgment.

    Each comes back as {path, kind, symbol, status, evidence} — status is the
    row's, or "new" when the ledger has none yet; evidence is what the
    machinery holds that the writer may want to weigh: the proposer's
    candidate canon, a hook's stamp, a divergence flag. Evidence, never a
    verdict — the writer is asked because the machine's guesses were the
    thing that poisoned two ledgers. Order follows ``written``; duplicates
    collapse. The caller has already resolved the project and checked access.
    """
    wanted: list[tuple[str, str, str]] = []
    for path, kind, name in written:
        key = ((path or "").strip(), (kind or "").strip(), (name or "").strip())
        if all(key) and key not in wanted:
            wanted.append(key)
    if not project_id or not wanted:
        return []
    paths = sorted({p for p, _k, _n in wanted})
    query = select(CodeShape).where(
        CodeShape.project_id == project_id,
        CodeShape.vanished_at.is_(None),
        CodeShape.path.in_(paths),
    )
    if repo_key:
        query = query.where(CodeShape.repo_key == repo_key)
    async with async_session() as session:
        rows = (await session.execute(query)).scalars().all()
    by_key = {(r.path, r.kind, r.symbol): r for r in rows}
    out: list[dict] = []
    for path, kind, name in wanted:
        row = by_key.get((path, kind, name))
        if row is not None and _is_judged(row):
            continue
        evidence: dict = {}
        if row is not None:
            if row.proposed_snippet_id:
                evidence["looks_like"] = {
                    "snippet_id": row.proposed_snippet_id,
                    "basis": row.proposal_basis,
                    "score": row.proposal_score,
                }
            if row.classified_by in _UNATTENDED_BY and row.snippet_id:
                evidence["stamped"] = {
                    "snippet_id": row.snippet_id, "score": stamp_score(row.reason),
                }
            if row.proposal_group:
                evidence["copies"] = row.proposal_group
            if row.diverges_from:
                evidence["diverges_from"] = row.diverges_from
        out.append({
            "path": path, "kind": kind, "symbol": name,
            "status": row.status if row is not None else "new",
            "evidence": evidence,
        })
    return out


# How many of a canon's rows a review listing shows before "…" — enough to
# judge from, not the whole table.
_REVIEW_ROWS_SHOWN = 12


def stamps_to_review(rows: Iterable[CodeShape], *, top: int = 10) -> dict:
    """Judged rows whose EVIDENCE no longer meets the bar the ledger now
    holds, and canons whose own rows no longer agree what they are.

    THIS CHANGES NOTHING, AND THAT IS THE POINT. Every `weak` row was written
    unattended by the write-path hook on a similarity score, and the harm in
    that is not a score being wrong once — it is a machine asserting a
    permanent classification with nobody reading it. A sweep that silently
    un-asserted them would be the identical mistake with a larger blast
    radius and a cleaner conscience. So this reports; an agent reads the
    evidence, decides, and records the decision through `classify_shapes`
    under its own name. The ledger should be able to say "these look wrong"
    without being able to act on it alone.

    Two lists, because they are different questions:

    `weak` — rows stamped on a resemblance below `_RESEMBLE_MIN`. A guard at
    the point of classification does not undo classifications already stored,
    so rows written before the floor need reading. Rows a
    person or an audit judged are never listed, however old: a human judgment
    is not weak evidence, it is a different kind of evidence.

    `incoherent` — canons whose membership no longer agrees what the canon
    IS. A canon is a claim that some set of shapes are the same sort of
    thing; when its own members are a class, three getters and a dozen tests,
    that claim has stopped being true, and every base-rate reading built on
    it is reading noise. `canon_form` already makes such a canon fall silent
    — this is what makes it VISIBLE.

    What counts as disagreement is `canon_coherence`, and it is deliberately
    looser than "one form": it compares FAMILIES, and it does not hold a
    method against the class it is defined beside. A review surface whose
    output is noise stops being read, which costs more than the check is
    worth.

    `strangers` names the rows that do not fit, not the first dozen members:
    the reader's question is which ones are wrong. `unattended` counts rows
    written by the hook with nobody reading — an incoherence made entirely of
    judged rows is far more likely to be this check being too strict than a
    ledger full of junk, and the reader should be able to see that without
    opening the canon.
    """
    live = [r for r in rows if r.vanished_at is None]
    weak = []
    for r in live:
        if r.status not in _NEEDS_TARGET or not is_weak_stamp(r):
            continue
        score = stamp_score(r.reason)
        weak.append({
            "path": r.path, "symbol": r.symbol, "kind": r.kind,
            "status": r.status, "snippet_id": r.snippet_id,
            "score": score, "signature": r.signature or "",
            "form": shape_form(r.signature or "", r.kind),
            "classified_at": r.classified_at.isoformat() if r.classified_at else None,
        })
    weak.sort(key=lambda d: (d["score"], d["path"], d["symbol"]))

    by_canon: dict[int, list[CodeShape]] = {}
    for r in live:
        if r.status in _NEEDS_TARGET and r.snippet_id:
            by_canon.setdefault(int(r.snippet_id), []).append(r)
    incoherent = []
    for sid, members in by_canon.items():
        coh = canon_coherence(members)
        if coh["readable"] < _DENSITY_MIN_JUDGED:
            continue   # too few to say anything either way
        if coh["coheres"]:
            continue
        unattended = sum(1 for r in members if r.classified_by in _UNATTENDED_BY)
        incoherent.append({
            "snippet_id": sid,
            "judged": len(members),
            "forms": coh["forms"],
            "families": coh["families"],
            "family": coh["family"],
            "attached": coh["attached"],
            "unattended": unattended,
            # The listing below is capped; without this you cannot tell a
            # canon with twelve strangers from one with three hundred.
            "stranger_count": len(coh["strangers"]),
            "weak_rows": sum(1 for r in members if is_weak_stamp(r)),
            # The rows that do NOT fit — not the first dozen members. The
            # reader's question is "which ones are wrong", and a sample of
            # the majority cannot answer it.
            "strangers": [
                {"path": r.path, "symbol": r.symbol,
                 "signature": r.signature or "", "by": r.classified_by,
                 "form": shape_form(r.signature or "", r.kind)}
                for r in coh["strangers"][:_REVIEW_ROWS_SHOWN]
            ],
        })
    incoherent.sort(key=lambda d: (-d["weak_rows"], -d["unattended"], -d["judged"]))
    return {
        "weak_count": len(weak),
        "weak": weak[:top],
        "incoherent_count": len(incoherent),
        "incoherent": incoherent[:top],
        "floor": _RESEMBLE_MIN,
    }


def divergence_summary(rows: Iterable[CodeShape], *, top: int = 10) -> dict:
    """Readout view: flagged shapes (newest first) and the recheck count."""
    flagged = [r for r in rows if r.diverges_from is not None and r.status in _MECHANICAL_TODO]
    flagged.sort(key=lambda r: (r.created_at or datetime.min.replace(tzinfo=timezone.utc)), reverse=True)
    recheck = sum(1 for r in rows if r.recheck_at is not None and r.vanished_at is None)
    return {
        "divergent": len(flagged),
        "divergence": [
            {"path": r.path, "symbol": r.symbol, "kind": r.kind,
             "canon_snippet_id": r.diverges_from}
            for r in flagged[:top]
        ],
        "recheck": recheck,
    }


async def shape_history(
    user_id: int, project_id: int, path: str, *, symbol: str = "", limit: int = 200
) -> dict:
    """What was used at ``path`` (a file or directory), when, and why: the
    current rows plus their events, oldest first. Read-gated like every
    other ledger read; {} when the caller cannot read the project."""
    from sqlalchemy import or_

    from grimoire.services import access

    if not await access.can_read_project(user_id, project_id):
        return {}
    clean = (path or "").strip().strip("/")
    conds = [CodeShape.project_id == project_id]
    if clean:
        conds.append(or_(CodeShape.path == clean, CodeShape.path.like(clean + "/%")))
    if symbol.strip():
        conds.append(CodeShape.symbol == symbol.strip())
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(*conds)
                .order_by(CodeShape.path, CodeShape.symbol, CodeShape.kind)
                .limit(500)
            )
        ).scalars().all()
        ids = [r.id for r in rows]
        events = (
            await session.execute(
                select(CodeShapeEvent).where(CodeShapeEvent.shape_id.in_(ids))
                .order_by(CodeShapeEvent.at.asc(), CodeShapeEvent.id.asc())
                .limit(max(1, min(limit, 1000)))
            )
        ).scalars().all() if ids else []
    return {
        "shapes": [
            {
                "path": r.path, "symbol": r.symbol, "kind": r.kind,
                "status": r.status, "snippet_id": r.snippet_id,
                "classified_by": r.classified_by, "reason": r.reason,
                "first_seen_commit": r.first_seen_commit,
                "last_seen_commit": r.last_seen_commit,
                "first_seen_at": iso(r.created_at),
                "vanished_at": iso(r.vanished_at),
                "recheck_at": iso(r.recheck_at),
                "diverges_from": r.diverges_from,
            }
            for r in rows
        ],
        "events": [e.to_dict() for e in events],
    }