"""The shape ledger's write side — sync and mechanical marking (#2788).

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 re
from datetime import datetime, timedelta, timezone
from typing import Iterable, NamedTuple

from sqlalchemy import select

from scribe.models import async_session
from scribe.models.code_shape import REASON_CODES, CodeShape, CodeShapeEvent, CodeShapeUse
from scribe.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 scribe.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` (#2869) 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 (#2792) whose 4th/5th fields — signature, body_sha — refresh the
    row's content fingerprint, and whose 7th (#2869) 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 (#2793).
                    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()


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 (#2870). 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 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


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()]
    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 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(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 (#2789): 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 — the StrictArgs lesson (#2709), a caller must never learn
    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 — "
                "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 (note 2786)"
            )
        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",
) -> 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.
    """
    from scribe.services import access
    from scribe.services import snippets as snippets_svc

    if via not in _CALLER_VIAS:
        raise ValueError(f"via must be one of: {', '.join(_CALLER_VIAS)}")
    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 (note 2786), 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)")

    now = datetime.now(timezone.utc)
    classified = 0
    unmatched: list[dict] = []
    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:
                unmatched.append({
                    "path": item.get("path"), "symbol": item.get("symbol"),
                })
                continue
            status = item["status"]
            for row in matches:
                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()
    return {"classified": classified, "unmatched": unmatched}


def rule_matches(row: CodeShape, *, path: str, pattern: str, kind: str) -> bool:
    """Does a ledger row fall under a rule-form classification (#2868)?
    ``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 (#2868): 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 scribe.services import access
    from scribe.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] = []
    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
            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()
    return {"classified": len(judged), "sample": sorted(judged)[:12]}


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 (#2793): "divergence" (new where a canon dominates,
    `diverges_from` names it) or "recheck" (a judged shape whose body moved).
    """
    from sqlalchemy import func, or_

    from scribe.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))
    if uses:
        # Consumers of a canon (#2870): 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 (#2789): 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" (#2483)."""
    from scribe.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 (#2870): 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 (#2791): hook evidence lands as rows ----------------
#
# The write-path hook (plugin/hooks/scribe_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` (note 2786).
#
# 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,
# becomes instance-of-N with classified_by="hook" and the evidence as reason.
#
# A hook row is EVIDENCE, not judgment: it only ever lands on rows nobody has
# judged (unclassified) or rows an earlier hook stamped, never on a canonical
# row or an agent/audit/import judgment. Re-judge with classify_shapes.

# "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)

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


async def stamp_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]:
    """Land hook evidence as `instance` rows for the shapes being written.

    ``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 rows stamped, 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 row under ``repo_key`` — first/last-seen empty —
    so the stamp is not lost to the next sync, which either confirms the
    shape (sets its seen marker) or stamps it vanished. No repo key → only
    existing rows are stamped.

    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: a row
    holds one canon (the known model limit logged on #2790).
    """
    from scribe.services import access
    from scribe.services import snippets as snippets_svc
    from scribe.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.
    in_play: dict[str, list[tuple[int, datetime, int, 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 ""
        kind = snippet_kind(symbol, fields.get("language") or "")
        if references_symbol(code, symbol, kind):
            rank, why = 2, f"hook: pulled #{sid}; payload references `{_norm_symbol(symbol)}`"
        elif sid in resembles:
            rank, why = 1, f"hook: pulled #{sid}; payload resembles it ({resembles[sid]:.2f})"
        else:
            continue
        in_play.setdefault(kind, []).append((rank, pulled_at, sid, why))
    if not in_play:
        return []
    for bucket in in_play.values():
        bucket.sort(key=lambda t: (t[0], t[1]), reverse=True)

    now = datetime.now(timezone.utc)
    stamped: 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
            _rank, _at, sid, why = bucket[0]
            row = by_key.get((name, kind))
            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
            await _judge(session, row, status="instance", snippet_id=sid, by="hook",
                         reason=why, at=now)
            stamped.append({
                "path": path, "symbol": name, "kind": kind,
                "snippet_id": sid, "reason": why,
            })
            # Every pulled canon the payload NAMES is a uses edge (#2870) — the
            # call-site fact, independent of which one the row is judged to be.
            await record_uses(
                session, row,
                [s_id for rank, _at, s_id, _why in bucket if rank == 2],
                basis="hook", evidence="write path: pulled the snippet, payload names its symbol",
            )
        if stamped:
            await session.commit()
    return stamped


# --- the mechanical proposer (#2792): 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
#              the P7 backfill classified by hand (#2790);
#   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 (note 2786): 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 (note 2917): 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 basis's own floor — stricter than the write-path arm's, because
# a code BODY against a prose-forward snippet document scores 0.68–0.75 for
# "both are about migrations"; first live run paired every alembic
# upgrade()/downgrade() with an unrelated canon at exactly that band.
_SEMANTIC_FLOOR = 0.8
# Bump when a basis's rule changes: rows remember the (body, ruleset) they
# were examined under, so a tightened rule re-examines everything once.
# v3: language-family gate on the sym bases, reference stoplist, semantic
# restricted to the shape's own project (#2871).
_PROPOSER_VERSION = 3
# 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.
_SIGNATURE_FLOOR = 0.8
_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. The 2026-08
# audit (#2871) found every cross-language hit wrong — a Python tool-module
# canon named `register` proposed for Vue `handleSubmit`s that call
# `authStore.register()`, and a TS store's `register` matched it by symbol;
# Minstrel/Forge TS canon 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",
}
_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",
}


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; the
# call-site relation for these becomes a `uses` edge once #2870 lands.
_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 reference_canons(kind: str, path: str, symbol: str, body: str, canons: Iterable[Canon]) -> list[int]:
    """Every canon this body NAMES (#2870) — the uses edges the proposer can
    write mechanically: same kind, same language family, symbol not in the
    generic-verb stoplist, and not the shape's own name."""
    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 _norm_symbol(c.symbol).lower() in _REFERENCE_STOPLIST:
            continue
        if references_symbol(body, c.symbol, kind):
            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,
) -> 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 (
            c.symbol
            and _norm_symbol(c.symbol).lower() not in _REFERENCE_STOPLIST
            and references_symbol(body, c.symbol, kind)
        ):
            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 (note 2786)."""
    from scribe.models.note import Note
    from scribe.services.access import browsable_notes_clause
    from scribe.services.coverage import extract_definitions
    from scribe.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()))


async def _semantic_canon(
    user_id: int, body: str, allowed: set[int]
) -> tuple[int, float] | None:
    from scribe.services.embeddings import semantic_search_notes
    from scribe.services.plugin_context import (
        WRITEPATH_DEFAULT_THRESHOLD, WRITEPATH_MIN_CODE_CHARS, concept_query,
    )

    if _substance(body) < WRITEPATH_MIN_CODE_CHARS or not allowed:
        return None
    query = concept_query(body) or body
    hits = await semantic_search_notes(
        user_id, query, limit=3,
        threshold=max(WRITEPATH_DEFAULT_THRESHOLD, _SEMANTIC_FLOOR),
        note_type="snippet", scope="browse",
    )
    for score, note in hits:
        if int(note.id) in allowed:
            return int(note.id), round(float(score), 3)
    return None


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}
    # The semantic arm is the widest net and, across projects, was pure noise
    # in the 2026-08 audit (#2871): it is held to the shape's own project and
    # language family. The precise bases (symbol/text) still reach family
    # canon in other projects (note 2786).
    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)}
    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 (#2870)
                # 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)
                if used:
                    await record_uses(session, row, used, basis="reference",
                                      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,
            )
            _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 (#2870),
            # whatever the row is then judged to be.
            used = reference_canons(row.kind, row.path, row.symbol, body, canons)
            if used:
                await record_uses(session, row, used, basis="reference",
                                  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))
        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))
            except Exception:
                logger.warning("semantic proposal failed", exc_info=True)
                found = None
            if found:
                row.proposed_snippet_id, row.proposal_score = found
                row.proposal_basis = "semantic"
                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 (note 2917)."""
    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),
                )
            )
        ).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) -> dict:
    """The readout's view of the proposer's standing: how many canon
    proposals await confirmation, and the largest derive-first groups."""
    proposed = 0
    by_canon: dict[int, int] = {}
    groups: dict[str, dict] = {}
    files: 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)
    for key, g in groups.items():
        g["files"] = len(files[key])
    # Body-identical groups first (#2872): 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 (#2899): 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 scribe.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
    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
            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()
    return {"confirmed": confirmed}


# --- the divergence readout (#2793): 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.

# 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 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:
        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) -> tuple[int, int, int] | None:
    """The dominant canon for the directory ``path`` sits in, for ``kind`` —
    the write-time question "is this a canon-dense place?"."""
    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]
    return dominant_canon(siblings)


async def write_time_divergence(
    project_id: int, path: str, shapes: list[tuple[str, str]], stamped: list[dict],
) -> 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 (already or just now) that canon's instance/canonical — 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 stamped}
    out: list[dict] = []
    kinds = {k for k, _n in shapes}
    density = {k: await canon_density(project_id, path, k) for k in kinds}
    if not any(density.values()):
        return out
    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:
        dom = density.get(kind)
        if not dom:
            continue
        sid, n, judged = dom
        if just_stamped.get((name, kind)) == sid:
            continue
        row = by_key.get((name, kind))
        if row is not None and (
            row.status != "unclassified" or row.proposed_snippet_id == sid
        ):
            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 (#2900): 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, note 2917): "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})
        out.append({
            "symbol": name, "kind": kind, "key": group,
            "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),
            },
        })
    return out


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)
            for r in siblings:
                if r.status not in _MECHANICAL_TODO:
                    continue
                if r.diverges_from is not None:
                    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"
                r.diverges_from = dom[0]
                flagged += 1
        await session.commit()
    return flagged


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 scribe.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],
    }