feat(ledger): divergence readout — button B where button A is canon, shape history, and judged-shape recheck (#2793, milestone 294 step 7)

Every judgment now goes through one helper that remembers the fingerprint
judged (classified_sha) and writes a code_shape_events row; the sync writes
vanished / reappeared / drifted events and flags recheck_at when a body
moves under an instance/variant. The refresh flags diverges_from on shapes
new since the previous computation that sit where one canon dominates the
judged siblings of their directory+kind and were not proposed as that canon
(a first seed flags nothing); the write-path hint asks the same question
in-band for the shapes the hook names. list_shapes(flag=divergence|recheck),
shape_history(project_id, path, symbol) (read-only), coverage line/payload/
card carry divergent + recheck. Backup v8 carries the history. Plugin 0.1.36.

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

src/scribe/services/backup.py

@@ -11,7 +11,7 @@ from scribe.models.note_supersession import NoteSupersession
 from scribe.models.note_version import NoteVersion
 from scribe.models.design_system import DesignSystem, DesignToken
 from scribe.models.note_usage import NoteUsageEvent
-from scribe.models.code_shape import CodeShape
+from scribe.models.code_shape import CodeShape, CodeShapeEvent
 from scribe.models.project import Project
 from scribe.models.repo_binding import RepoBinding
 from scribe.models.rulebook import (
@@ -40,8 +40,10 @@ logger = logging.getLogger(__name__)
 # v7 (2026-08) added code_shapes — the shape ledger (#2787). Classifications
 # are judgment data worth carrying; a restore keeps a judgment only when its
 # snippet target survives the id re-mapping, else the row rejoins the todo.
+# v8 (2026-08) added code_shape_events — the ledger's history (#2793): what
+# was used where, when, and why is not recomputable, so it travels.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 7
+BACKUP_VERSION = 8
 
 # Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
 # below, these two lists must together account for the entire schema — which is
@@ -59,8 +61,8 @@ _BACKED_UP = [
     # v5 (2026-08): the five-year gap this list was written to stop.
     "systems", "record_systems", "design_systems", "design_tokens",
     "note_usage_events", "repo_bindings", "note_supersessions",
-    # v7 (2026-08): the shape ledger (#2787).
-    "code_shapes",
+    # v7 (2026-08): the shape ledger (#2787); v8: its history (#2793).
+    "code_shapes", "code_shape_events",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -176,6 +178,10 @@ def _code_shape_rows(rows) -> list[dict]:
     return [r.to_dict() for r in rows]
 
 
+def _code_shape_event_rows(rows) -> list[dict]:
+    return [r.to_dict() for r in rows]
+
+
 def _repo_binding_rows(rows) -> list[dict]:
     return [
         {"user_id": r.user_id, "project_id": r.project_id, "repo_key": r.repo_key}
@@ -216,6 +222,9 @@ async def export_full_backup() -> dict:
         usage_events = (await session.execute(select(NoteUsageEvent))).scalars().all()
         repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
         code_shapes = (await session.execute(select(CodeShape))).scalars().all()
+        code_shape_events = (await session.execute(
+            select(CodeShapeEvent).order_by(CodeShapeEvent.at, CodeShapeEvent.id)
+        )).scalars().all()
         rulebooks = (await session.execute(select(Rulebook))).scalars().all()
         topics = (await session.execute(select(RulebookTopic))).scalars().all()
         rules = (await session.execute(select(Rule))).scalars().all()
@@ -391,6 +400,7 @@ async def export_full_backup() -> dict:
         "repo_bindings": _repo_binding_rows(repo_bindings),
         "note_supersessions": _note_supersession_rows(supersessions),
         "code_shapes": _code_shape_rows(code_shapes),
+        "code_shape_events": _code_shape_event_rows(code_shape_events),
     }
 
 
@@ -463,6 +473,10 @@ async def export_user_backup(user_id: int) -> dict:
         code_shapes = (await session.execute(
             select(CodeShape).where(CodeShape.project_id.in_(project_ids))
         )).scalars().all() if project_ids else []
+        code_shape_events = (await session.execute(
+            select(CodeShapeEvent).where(CodeShapeEvent.project_id.in_(project_ids))
+            .order_by(CodeShapeEvent.at, CodeShapeEvent.id)
+        )).scalars().all() if project_ids else []
         rulebooks = (await session.execute(
             select(Rulebook).where(Rulebook.owner_user_id == user_id)
         )).scalars().all()
@@ -652,6 +666,7 @@ async def export_user_backup(user_id: int) -> dict:
         "repo_bindings": _repo_binding_rows(repo_bindings),
         "note_supersessions": _note_supersession_rows(supersessions),
         "code_shapes": _code_shape_rows(code_shapes),
+        "code_shape_events": _code_shape_event_rows(code_shape_events),
     }
 
 
@@ -755,7 +770,7 @@ async def _restore_v2(data: dict) -> dict:
         "topic_suppressions": 0,
         "systems": 0, "record_systems": 0, "design_systems": 0,
         "design_tokens": 0, "note_usage_events": 0, "repo_bindings": 0,
-        "note_supersessions": 0, "code_shapes": 0,
+        "note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
     }
 
     async with async_session() as session:
@@ -1137,6 +1152,7 @@ async def _restore_v2(data: dict) -> dict:
         # survive the re-mapping (canonical/instance/variant with a gone
         # snippet) is downgraded to unclassified so it rejoins the todo
         # honestly instead of dangling; exempt needs no target and keeps.
+        shape_id_map: dict[int, int] = {}
         for cs_data in data.get("code_shapes", []):
             mapped_pid = project_id_map.get(cs_data.get("project_id", 0))
             if mapped_pid is None:
@@ -1149,7 +1165,7 @@ async def _restore_v2(data: dict) -> dict:
                 status = "unclassified"
                 classified_by = None
                 classified_at = None
-            session.add(CodeShape(
+            shape = CodeShape(
                 project_id=mapped_pid,
                 repo_key=cs_data.get("repo_key", ""),
                 path=cs_data.get("path", ""),
@@ -1168,11 +1184,41 @@ async def _restore_v2(data: dict) -> dict:
                 # against the restored snippet ids.
                 signature=cs_data.get("signature", ""),
                 body_sha=cs_data.get("body_sha", ""),
+                classified_sha=cs_data.get("classified_sha", ""),
                 created_at=_dt(cs_data.get("created_at")),
                 updated_at=_dt(cs_data.get("updated_at")),
-            ))
+            )
+            session.add(shape)
+            await session.flush()
+            if cs_data.get("id"):
+                shape_id_map[int(cs_data["id"])] = shape.id
             stats["code_shapes"] += 1
 
+        # v8: the ledger's history rides its shapes. snippet_id is kept as
+        # the history's own claim (FK-free by design) but re-mapped when the
+        # snippet survived, so a restored timeline points at restored records.
+        for ev in data.get("code_shape_events", []):
+            new_shape_id = shape_id_map.get(ev.get("shape_id") or 0)
+            mapped_pid = project_id_map.get(ev.get("project_id", 0))
+            if new_shape_id is None or mapped_pid is None:
+                continue
+            old_sid = ev.get("snippet_id")
+            session.add(CodeShapeEvent(
+                shape_id=new_shape_id,
+                project_id=mapped_pid,
+                path=ev.get("path", ""),
+                symbol=ev.get("symbol", ""),
+                kind=ev.get("kind", "sym"),
+                event=ev.get("event", "classified"),
+                status=ev.get("status"),
+                snippet_id=note_id_map.get(old_sid, old_sid) if old_sid else None,
+                classified_by=ev.get("classified_by"),
+                reason=ev.get("reason"),
+                commit=ev.get("commit", ""),
+                at=_dt(ev.get("at")),
+            ))
+            stats["code_shape_events"] += 1
+
         await session.commit()
 
     logger.info("Restored v2/v3 backup: %s", stats)

src/scribe/services/coverage.py

@@ -397,6 +397,18 @@ async def compute_coverage(
         await shape_ledger.apply_derive_groups(project_id)
     except Exception:
         logger.warning("derive-first grouping failed", exc_info=True)
+    # The button-B pass (#2793): shapes new since the PREVIOUS computation,
+    # where a canon dominates. The previous computation's stamp is the cache;
+    # a first seed has none, so it flags nothing (everything is new then).
+    try:
+        previous = await get_setting(user_id, f"{_CACHE_KEY_PREFIX}{project_id}")
+        since = None
+        if previous:
+            stamp = (json.loads(previous) or {}).get("computed_at")
+            since = datetime.fromisoformat(stamp) if stamp else None
+        await shape_ledger.flag_divergence(project_id, since=since)
+    except Exception:
+        logger.warning("divergence pass failed", exc_info=True)
 
     # Project-wide readout, deliberately wider than this walk: a second bound
     # repo that was unreachable today still has live rows, and they count.
@@ -413,6 +425,7 @@ async def compute_coverage(
 
     unclassified = counts.pop("unclassified")
     proposals = shape_ledger.proposal_summary(rows)
+    divergence = shape_ledger.divergence_summary(rows)
     return {
         "total": len(rows),
         "accounted": len(rows) - unclassified,
@@ -423,6 +436,11 @@ async def compute_coverage(
         "proposed": proposals["proposed"],
         "derive_groups": proposals["derive_groups"],
         "proposer": proposer_stats,
+        # The divergence readout (#2793): button B where button A is canon,
+        # and judged shapes whose bodies moved since they were judged.
+        "divergent": divergence["divergent"],
+        "divergence": divergence["divergence"],
+        "recheck": divergence["recheck"],
         # Honesty flag, not decoration: every surface that shows the number
         # is expected to carry it through.
         "estimate": True,
@@ -572,9 +590,13 @@ def coverage_line(coverage: dict) -> str:
         n_groups = len(coverage.get("derive_groups") or [])
         if n_groups:
             standing.append(f"{n_groups} derive group{'s' if n_groups != 1 else ''}")
+        if coverage.get("divergent"):
+            standing.append(f"{coverage['divergent']} DIVERGENT")
         if standing:
             line += f" ({', '.join(standing)})"
         gaps = [g["dir"] for g in coverage.get("largest_gaps") or []]
         if gaps:
             line += ", largest: " + ", ".join(gaps)
+    if coverage.get("recheck"):
+        line += f"; {coverage['recheck']} judged shape{'s' if coverage['recheck'] != 1 else ''} changed since judged — recheck"
     return line

src/scribe/services/plugin_context.py

@@ -923,7 +923,19 @@ async def build_write_path_hint(
             )
         except Exception:
             logger.warning("Write-path ledger stamping failed", exc_info=True)
-    if not synced and not menu and not stamped:
+    # The in-band button-B check (#2793): the hook named the shapes being
+    # written; if this directory+kind is canon-dense and a named shape isn't
+    # (about to be) an instance of that canon, say so NOW — at the write,
+    # not at the next audit.
+    divergence: list[dict] = []
+    if stamp_shapes and project_id:
+        try:
+            divergence = await shape_ledger_svc.write_time_divergence(
+                project_id, path, stamp_shapes, stamped
+            )
+        except Exception:
+            logger.warning("write-time divergence check failed", exc_info=True)
+    if not synced and not menu and not stamped and not divergence:
         return empty
 
     owners = await owner_names_for({
@@ -989,6 +1001,8 @@ async def build_write_path_hint(
 
     if stamped:
         lines.append(_stamp_line(path, stamped))
+    if divergence:
+        lines.append(_divergence_line(path, divergence))
 
     # Split by arm, which is the whole reason this table exists. The place arm
     # carries no score and so has no home in retrieval_logs; before #2085 a
@@ -1012,9 +1026,26 @@ async def build_write_path_hint(
         "sync_note_ids": sync_note_ids,
         "config": cfg,
         "stamped": stamped,
+        "divergence": divergence,
     }
 
 
+def _divergence_line(path: str, divergence: list[dict]) -> str:
+    """Button B where button A is canon — named at the write (#2793)."""
+    parts = [
+        f"`{('.' if d['kind'] == 'css' else '') + d['symbol']}` → #{d['canon_snippet_id']} "
+        f"({d['instances']} of {d['judged']} judged siblings are its instances)"
+        for d in divergence
+    ]
+    return (
+        f"> Divergence check at `{path}`: a canon dominates this directory — "
+        f"{'; '.join(parts)}. If this is a new instance, pull that snippet "
+        "and build from it; if it is a deliberate departure, "
+        "`classify_shapes(..., status=\"variant\", reason=…)` records the why; "
+        "otherwise it reads as unintended divergence."
+    )
+
+
 def _stamp_line(path: str, stamped: list[dict]) -> str:
     """One line saying what the ledger just recorded, so the session can
     correct a wrong stamp in the moment rather than an audit finding it."""

src/scribe/services/shape_ledger.py

@@ -30,7 +30,7 @@ from typing import Iterable, NamedTuple
 from sqlalchemy import select
 
 from scribe.models import async_session
-from scribe.models.code_shape import CodeShape
+from scribe.models.code_shape import CodeShape, CodeShapeEvent
 
 logger = logging.getLogger(__name__)
 
@@ -111,11 +111,25 @@ async def sync_repo_shapes(
             row.last_seen_commit = seen_marker
             if signature:
                 row.signature = signature
-            if body_sha:
+            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.
-            row.vanished_at = None
+            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
@@ -124,9 +138,50 @@ async def sync_repo_shapes(
         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:
@@ -159,19 +214,15 @@ async def mark_canonicals(
                 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
+                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"
             ):
-                row.status = "unclassified"
-                row.snippet_id = None
-                row.classified_by = None
-                row.classified_at = None
+                await _judge(session, row, status="unclassified", snippet_id=None,
+                             by=None, reason=None, at=now)
         await session.commit()
 
 
@@ -308,23 +359,11 @@ async def classify_shapes(
                 continue
             status = item["status"]
             for row in matches:
-                row.status = status
-                # The machine proposes, judgment classifies: any judgment
-                # retires the standing proposal; a withdrawal also forgets
-                # the examination so the next refresh proposes afresh.
-                _clear_proposal(row, reexamine=(status == "unclassified"))
-                if status == "unclassified":
-                    row.snippet_id = None
-                    row.reason = None
-                    row.classified_by = None
-                    row.classified_at = None
-                else:
-                    row.snippet_id = (
-                        int(item["snippet_id"]) if status in _NEEDS_TARGET else None
-                    )
-                    row.reason = (item.get("reason") or "").strip() or None
-                    row.classified_by = via
-                    row.classified_at = now
+                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}
@@ -341,6 +380,7 @@ async def list_project_shapes(
     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.
 
@@ -349,7 +389,9 @@ async def list_project_shapes(
     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).
+    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_
 
@@ -380,6 +422,10 @@ async def list_project_shapes(
         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(
@@ -615,12 +661,8 @@ async def stamp_write_path_instances(
                 by_key[(name, kind)] = row
             elif not (row.status == "unclassified" or row.classified_by == "hook"):
                 continue  # a judgment — or the canon itself — stands
-            row.status = "instance"
-            row.snippet_id = sid
-            row.reason = why
-            row.classified_by = "hook"
-            row.classified_at = now
-            _clear_proposal(row)
+            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,
@@ -1051,15 +1093,224 @@ async def confirm_proposals(
         for row in rows:
             if (row.proposal_score or 0.0) < min_score:
                 continue
-            row.status = "instance"
-            row.snippet_id = row.proposed_snippet_id
-            row.reason = (
-                f"confirmed {row.proposal_basis} proposal"
-                f" ({(row.proposal_score or 0.0):.2f})"
+            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})"
+                ),
             )
-            row.classified_by = "agent"
-            row.classified_at = now
-            _clear_proposal(row)
             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": r.created_at.isoformat() if r.created_at else None,
+                "vanished_at": r.vanished_at.isoformat() if r.vanished_at else None,
+                "recheck_at": r.recheck_at.isoformat() if r.recheck_at else None,
+                "diverges_from": r.diverges_from,
+            }
+            for r in rows
+        ],
+        "events": [e.to_dict() for e in events],
+    }