feat(ledger): coverage refresh feeds the shape ledger; the readout inverts to accounting (#2788, milestone 294 step 2)

compute_coverage is now the ledger's sync point: every walk upserts the
extracted shapes (new → unclassified, the todo state; surviving → last-seen
bump; vanished → stamped, kept as history), re-files judgments whose snippet
target went away, and mechanically stamps snippet reference locations as
canonical — the one always-safe rule, self-healing only for its own stamps
(an agent's judgment is never unwound by machinery).

The covering predicate moves to shape_ledger.location_covers as the single
home (match_shapes retired with its consumer); coverage's payload and line
invert from 'N/M shapes recorded' to shape ACCOUNTING per note 2786:
accounted/total with a canonical·instance·variant·exempt breakdown, and
unclassified — THE todo — with its largest directories. Cache key bumps to
v2 so pre-ledger blobs honestly read 'not measured yet' instead of rendering
in a shape no longer spoken.

Readout is deliberately project-wide (all repos' live rows), while the walk
serves whichever repos the owner's keyring reaches this refresh.

Integration tests pin the new contract: rows for every extracted shape,
mechanical canonical stamps carrying snippet ids, idempotent recompute,
agent judgments surviving recompute AND vanish/return, vanished rows leaving
the readout but keeping their history.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/scribe/services/shape_ledger.py

@@ -0,0 +1,176 @@
+"""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
+
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.code_shape import CodeShape
+
+# 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[tuple[str, str, str]],
+    *,
+    seen_marker: str,
+) -> None:
+    """Upsert one repo's extracted (path, kind, name) shapes into the ledger.
+
+    ``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 path, kind, name in shapes:
+            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,
+                ))
+                continue
+            row.last_seen_commit = seen_marker
+            # A shape that vanished and came back is live again — the vanish
+            # stays visible in history via updated_at, not as a dead flag.
+            row.vanished_at = None
+            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
+        await session.commit()
+
+
+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":
+                row.status = "canonical"
+                row.snippet_id = covering
+                row.classified_by = "mechanical"
+                row.classified_at = now
+            elif (
+                covering is None
+                and row.status == "canonical"
+                and row.classified_by == "mechanical"
+            ):
+                row.status = "unclassified"
+                row.snippet_id = None
+                row.classified_by = None
+                row.classified_at = None
+        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()
+        )