"""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 CodeShape, CodeShapeEvent
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)


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. ``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 ""
            key = (path, name, kind)
            if key in seen:
                continue
            seen.add(key)
            row = by_key.get(key)
            if row is None:
                session.add(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,
                ))
                continue
            row.last_seen_commit = seen_marker
            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,
) -> 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.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 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 == "unclassified":
                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)"
            )
    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 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,
                )
                classified += 1
        await session.commit()
    return {"classified": classified, "unmatched": unmatched}


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 = "",
) -> 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))
    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()
    readable: dict[int, bool] = {}
    out: dict[str, list[dict]] = {"instances": [], "variants": []}
    for row in rows:
        if row.project_id not in readable:
            readable[row.project_id] = await access.can_read_project(
                user_id, row.project_id
            )
        if not readable[row.project_id]:
            continue
        out["instances" if row.status == "instance" else "variants"].append(
            _consumer_dict(row)
        )
    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 == "unclassified" 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,
            })
        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
# 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 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.status == "unclassified",
                    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
            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
            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}. Identical
    bodies in ≥2 places group as `dup:<sha>`; the same name defined in ≥3
    files groups as `name:<kind>:<symbol>`; a row joins at most one group,
    the copy before the name."""
    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:
            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():
        if len({k[0] for k in keys}) >= _DERIVE_MIN_NAME:
            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 == "unclassified",
                    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
    groups: dict[str, dict] = {}
    for row in rows:
        if row.status != "unclassified":
            continue
        if row.proposed_snippet_id is not None:
            proposed += 1
        elif row.proposal_group:
            g = groups.setdefault(row.proposal_group, {
                "group": row.proposal_group, "kind": row.kind,
                "label": (
                    ("." if row.kind == "css" else "") + row.symbol
                    if row.proposal_group.startswith("name:")
                    else f"{row.symbol} (identical body)"
                ),
                "size": 0, "paths": [],
            })
            g["size"] += 1
            if len(g["paths"]) < 3:
                g["paths"].append(row.path)
    ranked = sorted(groups.values(), key=lambda g: (-g["size"], g["group"]))
    return {"proposed": proposed, "derive_groups": ranked[: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 == "unclassified",
        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


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 != "unclassified":
                    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 == "unclassified"]
    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],
    }