Merge pull request 'Shape ledger steps 5–7: write-path stamping, mechanical proposer, divergence readout (#2791, #2792, #2793)' (#117) from dev into main

alembic/versions/0080_code_shape_proposals.py

@@ -0,0 +1,54 @@
+"""Shape fingerprints + the mechanical proposer's columns (#2792, milestone 294)
+
+Revision ID: 0080
+Revises: 0079
+Create Date: 2026-08-21
+
+Two additions to the ledger. `signature` / `body_sha` fingerprint each shape
+(definition line + a whitespace/comment-insensitive hash of its block) so the
+proposer can match on content and a later drift recheck can notice change,
+without the ledger ever storing code. The proposal columns carry the
+proposer's standing suggestion for an unclassified row — instance-of-#N with
+a basis and score, or a derive-first group key — and `proposed_sha`
+remembers the content it was judged at so a refresh re-examines only what
+changed. Mechanical and recomputable: a restore that lacks them loses
+nothing the next refresh does not rebuild.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0080"
+down_revision = "0079"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("code_shapes", sa.Column("signature", sa.Text(), nullable=False, server_default=""))
+    op.add_column("code_shapes", sa.Column("body_sha", sa.Text(), nullable=False, server_default=""))
+    op.add_column(
+        "code_shapes",
+        sa.Column(
+            "proposed_snippet_id",
+            sa.BigInteger(),
+            sa.ForeignKey("notes.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+    )
+    op.add_column("code_shapes", sa.Column("proposal_basis", sa.Text(), nullable=True))
+    op.add_column("code_shapes", sa.Column("proposal_score", sa.Float(), nullable=True))
+    op.add_column("code_shapes", sa.Column("proposal_group", sa.Text(), nullable=True))
+    op.add_column("code_shapes", sa.Column("proposed_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column("code_shapes", sa.Column("proposed_sha", sa.Text(), nullable=False, server_default=""))
+    op.create_index(
+        "ix_code_shapes_proposed", "code_shapes", ["project_id", "proposed_snippet_id"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_code_shapes_proposed", table_name="code_shapes")
+    for col in (
+        "proposed_sha", "proposed_at", "proposal_group", "proposal_score",
+        "proposal_basis", "proposed_snippet_id", "body_sha", "signature",
+    ):
+        op.drop_column("code_shapes", col)

alembic/versions/0081_shape_history_and_divergence.py

@@ -0,0 +1,70 @@
+"""Shape history, recheck, and the divergence flag (#2793, milestone 294)
+
+Revision ID: 0081
+Revises: 0080
+Create Date: 2026-08-21
+
+The payoff surface of the ledger. `classified_sha` remembers the fingerprint
+a judgment was made at so a later body change under an instance/variant can
+flag `recheck_at`; `diverges_from` is the button-B flag (a shape new since
+the previous refresh, where one canon dominates its directory+kind, and not
+proposed as that canon). `code_shape_events` is the what-was-used-when
+record: every classification, vanish, reappearance, and drift as it
+happened — history the row alone cannot keep once it moves on.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0081"
+down_revision = "0080"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("code_shapes", sa.Column("classified_sha", sa.Text(), nullable=False, server_default=""))
+    op.add_column("code_shapes", sa.Column("recheck_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column(
+        "code_shapes",
+        sa.Column(
+            "diverges_from",
+            sa.BigInteger(),
+            sa.ForeignKey("notes.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+    )
+    op.create_index("ix_code_shapes_diverges", "code_shapes", ["project_id", "diverges_from"])
+    op.create_table(
+        "code_shape_events",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "shape_id",
+            sa.Integer(),
+            sa.ForeignKey("code_shapes.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("project_id", sa.Integer(), nullable=False),
+        sa.Column("path", sa.Text(), nullable=False),
+        sa.Column("symbol", sa.Text(), nullable=False),
+        sa.Column("kind", sa.Text(), nullable=False),
+        sa.Column("event", sa.Text(), nullable=False),
+        sa.Column("status", sa.Text(), nullable=True),
+        sa.Column("snippet_id", sa.BigInteger(), nullable=True),
+        sa.Column("classified_by", sa.Text(), nullable=True),
+        sa.Column("reason", sa.Text(), nullable=True),
+        sa.Column("commit", sa.Text(), nullable=False, server_default=""),
+        sa.Column("at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_code_shape_events_shape", "code_shape_events", ["shape_id", "at"])
+    op.create_index(
+        "ix_code_shape_events_project_path", "code_shape_events", ["project_id", "path"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_code_shape_events_project_path", table_name="code_shape_events")
+    op.drop_index("ix_code_shape_events_shape", table_name="code_shape_events")
+    op.drop_table("code_shape_events")
+    op.drop_index("ix_code_shapes_diverges", table_name="code_shapes")
+    for col in ("diverges_from", "recheck_at", "classified_sha"):
+        op.drop_column("code_shapes", col)

frontend/src/views/ProjectView.vue

@@ -427,6 +427,13 @@ interface CoverageGap {
   unclassified: number;
   total: number;
 }
+interface DeriveGroup {
+  group: string;
+  kind: string;
+  label: string;
+  size: number;
+  paths: string[];
+}
 interface Coverage {
   total: number;
   accounted: number;
@@ -436,6 +443,21 @@ interface Coverage {
   computed_at: string;
   repos: { repo: string; ref: string; total: number; accounted: number }[];
   largest_gaps: CoverageGap[];
+  // The mechanical proposer's standing (#2792): canon proposals awaiting an
+  // agent's confirm, and the biggest repeats-with-no-canon families.
+  proposed?: number;
+  derive_groups?: DeriveGroup[];
+  // The divergence readout (#2793): button B where button A is canon, and
+  // judged shapes whose bodies moved since they were judged.
+  divergent?: number;
+  divergence?: Divergence[];
+  recheck?: number;
+}
+interface Divergence {
+  path: string;
+  symbol: string;
+  kind: string;
+  canon_snippet_id: number;
 }
 
 const coverage = ref<Coverage | null>(null);
@@ -746,6 +768,46 @@ async function confirmDelete() {
               {{ gap.dir }} <span class="coverage-gap-count">{{ gap.unclassified }}</span>
             </span>
           </div>
+          <div
+            v-if="coverage.proposed || coverage.derive_groups?.length"
+            class="coverage-gaps"
+            title="The proposer matched these against canon; an agent confirms them in batches (confirm_shape_proposals)."
+          >
+            <span class="coverage-gaps-label">Proposed:</span>
+            <span v-if="coverage.proposed" class="coverage-gap-chip">
+              {{ coverage.proposed }} awaiting confirm
+            </span>
+            <span
+              v-for="g in coverage.derive_groups || []"
+              :key="g.group"
+              class="coverage-gap-chip"
+              :title="'Repeats with no canon — derive one first. ' + g.paths.join(', ')"
+            >
+              {{ g.label }} <span class="coverage-gap-count">×{{ g.size }}</span>
+            </span>
+          </div>
+          <div
+            v-if="coverage.divergent || coverage.recheck"
+            class="coverage-gaps"
+            title="Divergent: a new shape where one canon dominates its directory and that isn't proposed as that canon — button B where button A is canon. Recheck: a judged shape whose body changed since it was judged."
+          >
+            <span class="coverage-gaps-label">Divergence:</span>
+            <span
+              v-for="d in coverage.divergence || []"
+              :key="d.path + '::' + d.symbol"
+              class="coverage-gap-chip coverage-divergent"
+              :title="d.path + ' — canon here is snippet #' + d.canon_snippet_id"
+            >
+              {{ d.kind === 'css' ? '.' : '' }}{{ d.symbol }}
+              <span class="coverage-gap-count">→ #{{ d.canon_snippet_id }}</span>
+            </span>
+            <span v-if="(coverage.divergent || 0) > (coverage.divergence?.length || 0)" class="coverage-gap-chip">
+              +{{ (coverage.divergent || 0) - (coverage.divergence?.length || 0) }} more
+            </span>
+            <span v-if="coverage.recheck" class="coverage-gap-chip">
+              {{ coverage.recheck }} to recheck
+            </span>
+          </div>
         </template>
         <p v-else class="coverage-empty">
           Not measured yet — Refresh reads the bound repo's definitions into
@@ -1300,6 +1362,7 @@ async function confirmDelete() {
   font-size: 0.74rem;
 }
 .coverage-gap-count { opacity: 0.65; }
+.coverage-divergent { border-color: var(--fs-warning); }
 .coverage-empty {
   margin: 0;
   font-size: 0.8rem;

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.33",
+  "version": "0.1.36",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/hooks/scribe_prior_art.sh

@@ -14,6 +14,12 @@
 # on an instance with no forge connection (decision #2707). Everything else is
 # the REUSE menu. The two dedup separately (see the state files below).
 #
+# It is also the shape ledger's write-path feed (#2791): it names the
+# definitions being written (`shapes=`), and the server — only when the
+# session has PULLED a snippet this code references or resembles — records
+# them as instance rows, classified_by=hook. Evidence, not judgment; the
+# context line says what landed so a wrong stamp is corrected in the moment.
+#
 # NEVER BLOCKS. It returns `additionalContext` with no `permissionDecision`, so
 # the write proceeds untouched and Claude sees the note beside the tool result.
 # Any failure — unconfigured, unreachable, malformed — exits 0 in silence. A
@@ -96,10 +102,14 @@ fi
 # `ReturnType name(...)`) needs a real parser, and `impl` blocks are excluded
 # because several per type is normal Rust, not duplication.
 # ---------------------------------------------------------------------------
-local_lines=""
-if [ -n "$repo_root" ] && [ -n "$code" ]; then
-  # kind<TAB>name for each thing this payload DEFINES.
-  names=$(printf '%s' "$code" | awk '
+# kind<TAB>name for each thing a piece of code DEFINES, in source order. One
+# program, two consumers: the local duplicate arm (every definition in the
+# payload) and the ledger feed (#2791, below: the definitions being written,
+# or the one enclosing an Edit). Rule-for-rule mirrored by the server's
+# services/coverage.py extract_shapes — ledger rows are keyed by what THAT
+# sees, so the two must agree on what counts as a definition.
+scribe_defs() {
+  awk '
     {
       # CSS class definition: .name { or .name,
       if (match($0, /^[[:space:]]*\.[A-Za-z][A-Za-z0-9_-]*[[:space:]]*[,{]/)) {
@@ -132,8 +142,16 @@ if [ -n "$repo_root" ] && [ -n "$code" ]; then
         if (t != "") print "sym\t" t; next
       }
     }
-  ' 2>/dev/null | sort -u | head -12) || names=""
+  ' 2>/dev/null
+}
 
+names=""
+if [ -n "$code" ]; then
+  names=$(printf '%s' "$code" | scribe_defs | sort -u | head -12) || names=""
+fi
+
+local_lines=""
+if [ -n "$repo_root" ] && [ -n "$names" ]; then
   while IFS=$'\t' read -r kind name; do
     [ -n "${name:-}" ] || continue
     case "$kind" in
@@ -156,6 +174,38 @@ if [ -n "$local_lines" ]; then
   local_context="> Already defined elsewhere in this repo — check before adding another copy (\`git grep\` shown; this is a nudge, not a gate):"$'\n'"${local_lines}"
 fi
 
+# ---------------------------------------------------------------------------
+# THE LEDGER FEED (#2791). The server keeps a shape ledger — every definition
+# in the bound repo, classified against recorded canon — and this hook is the
+# one place that sees a shape AT THE MOMENT IT IS WRITTEN. So it names the
+# shapes in play: every definition in the payload, or — for an Edit that
+# changes the inside of a function rather than its signature — the definition
+# enclosing the edit, found by walking the target file upward from the edited
+# lines. The server decides whether evidence exists (the session pulled a
+# snippet this code references or resembles) and stamps instance rows; with
+# no pulled canon in play, nothing is recorded. Titles only still — this sends
+# names, not bodies.
+# ---------------------------------------------------------------------------
+shapes="$names"
+if [ -z "$shapes" ] && [ -f "$file_path" ] && command -v tac >/dev/null 2>&1; then
+  old_first=$(printf '%s' "$event" \
+    | jq -r '.tool_input.old_string // .tool_input.old_str // empty' 2>/dev/null \
+    | grep -m1 -v '^[[:space:]]*$') || old_first=""
+  if [ -n "$old_first" ]; then
+    ln=$(grep -nF -m1 -- "$old_first" "$file_path" 2>/dev/null | cut -d: -f1) || ln=""
+    if [ -n "$ln" ]; then
+      shapes=$(head -n "$ln" "$file_path" | tac | scribe_defs | head -1) || shapes=""
+    fi
+  fi
+fi
+shapes_q=""
+if [ -n "$shapes" ]; then
+  enc=$(printf '%s\n' "$shapes" \
+    | awk -F'\t' 'NF>=2 {printf "%s%s:%s", (n++?",":""), $1, $2}' \
+    | jq -sRr '@uri' 2>/dev/null) || enc=""
+  [ -n "$enc" ] && shapes_q="&shapes=${enc}"
+fi
+
 url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_API_ENDPOINT:-}}
 token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_API_TOKEN:-}}
 # Guard against an unexpanded ${...} placeholder arriving as a literal.
@@ -230,7 +280,7 @@ fi
 # finding that needed no instance to produce.
 body=$(curl -fsS --max-time 5 \
   -H "Authorization: Bearer ${token}" \
-  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}" 2>/dev/null) || body=""
+  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${shapes_q}" 2>/dev/null) || body=""
 
 context=""
 if [ -n "$body" ]; then

plugin/skills/reusing-code/SKILL.md

@@ -34,6 +34,9 @@ through recall/auto-inject; this skill is the active reflex around that.
   to ask both at once.
 - If a snippet fits, pull it in full with `get_snippet(id)` and reuse it — its
   `location` points at the reference implementation. Adapt, don't re-derive.
+  The pull also does the accounting: the code you then write that references
+  or resembles it is stamped an `instance` of that canon in the shape ledger
+  (classified_by=hook) — reuse from memory leaves no row.
 - If auto-inject already surfaced a snippet title that looks relevant, that's
   your cue to `get_snippet` it rather than start from scratch.
 - **Prior art offered beside a write is not noise — read it.** When Scribe notes

plugin/skills/shape-accounting/SKILL.md

@@ -34,6 +34,44 @@ row carries a status:
    nothing. Rows, never prose — a consumer list in a note or verification
    detail cannot be sorted, queried, or diffed.
 
+## Rows that arrive on their own
+
+Two feeds keep the ledger current between your batches, so most shapes never
+need a hand judgment:
+
+- **The sync** stamps a snippet's own reference location `canonical`
+  (`classified_by: mechanical`).
+- **The write path** stamps instances as you work: when you `get_snippet` a
+  canon and then Write/Edit code that references or resembles it, the
+  definitions being written land as `instance` rows (`classified_by: hook`,
+  the evidence in `reason`), and the prior-art hook tells you what landed
+  ("Shape accounting: recorded at … → instance of #N"). Offered-but-unopened
+  snippets stamp nothing — so *pull the canon you are instantiating*; that
+  pull is what turns your reuse into accounting. A hook row is evidence, not
+  judgment: it never overrides a classification you made, and a
+  `classify_shapes` call overrides it.
+
+## The machine proposes, judgment classifies
+
+Every coverage refresh runs the **mechanical proposer** over the unclassified
+rows: same symbol as a canon elsewhere → textual containment → body
+references a canon → signature resemblance → semantic (capped per refresh). A hit
+is a *proposal* on the row, never a classification. Work the queue in bulk:
+
+1. `list_shapes(project_id, proposal="canon", snippet_id=N)` or
+   `path="dir"` — read the page; `proposal` carries snippet_id, basis, score.
+2. `confirm_shape_proposals(project_id, snippet_id=N)` (or `path=`,
+   `basis=`) for the ones that hold — hundreds at a time; `symbol` and
+   `reference` proposals are near-certain, `semantic` deserves a look.
+3. `classify_shapes` the rest — variant, exempt, or instance of a different
+   snippet. Any judgment retires the proposal.
+
+`list_shapes(project_id, proposal="derive")` lists the **derive-first
+candidates** — the same body in ≥2 places or the same name defined in ≥3
+files, with no canon at all (`proposal.group` names the family; the coverage
+payload's `derive_groups` ranks the biggest). That is the consolidation
+queue, not a classification queue: see below.
+
 ## The derive-first rule
 
 N same-shaped occurrences matching **no** recorded canon is never N loose
@@ -42,6 +80,25 @@ the dominant form, `create_snippet` it, migrate the outliers, then classify
 the rest as instances. Canon is determined from the code; consistency comes
 from the derivation, not from asking permission.
 
+## The divergence readout — button B where button A is canon
+
+Three questions the ledger answers mechanically (#2793):
+
+- **Divergence** — `list_shapes(project_id, flag="divergence")` (and the
+  coverage line's "N DIVERGENT"): a shape new since the previous refresh, in
+  a directory where one canon dominates the judged siblings, that the
+  proposer did not match to that canon. `diverges_from` names the canon.
+  Judge it: `instance` if it should be built from the canon (and rebuild
+  it), `variant` with the why if the departure is deliberate. The write-path
+  hook asks the same question in-band the moment such a shape is written.
+- **History** — `shape_history(project_id, path, symbol?)`: the current rows
+  plus every `classified` / `vanished` / `reappeared` / `drifted` event with
+  its commit — "instance of #N from <date>, re-judged variant of #M because
+  R, vanished at C". Rows, not recollection.
+- **Recheck** — `list_shapes(project_id, flag="recheck")`: judged
+  instances/variants whose body changed since judged. The judgment stands;
+  re-confirm it (classify again with the same status) or re-judge.
+
 ## What this buys
 
 Divergence becomes mechanical: when button B appears where button A is canon,

src/scribe/mcp/server.py

@@ -112,7 +112,7 @@ _READ_ONLY_TOOLS = frozenset({
     "list_repo_bindings",
     # The shape ledger's todo query (#2789). Reads only — classify_shapes is
     # the write, and it is deliberately NOT here.
-    "list_shapes",
+    "list_shapes", "shape_history",
 })
 
 # Read-SHAPED tools that must NOT be reachable with a read key — a getter that

src/scribe/mcp/tools/shapes.py

@@ -62,6 +62,8 @@ async def list_shapes(
     include_vanished: bool = False,
     limit: int = 100,
     offset: int = 0,
+    proposal: str = "",
+    flag: str = "",
 ) -> dict:
     """Read a project's shape ledger — `status="unclassified"` IS the todo.
 
@@ -75,22 +77,108 @@ async def list_shapes(
         snippet_id: rows classified against this snippet — a consumer map.
         include_vanished: include shapes no longer in the tree (history).
         limit/offset: page through big ledgers (limit caps at 500).
+        proposal: the proposer's queue (#2792) — "any", "canon" (rows the
+            machine thinks are an instance of a snippet: `proposal` carries
+            snippet_id, basis, score), "derive" (rows that repeat with NO
+            canon: `proposal.group` names the family), or one basis
+            (symbol/text/reference/signature/semantic).
+        flag: the divergence readout (#2793) — "divergence": shapes new
+            since the previous refresh in a directory where one canon
+            dominates the judged siblings and NOT proposed as that canon
+            (`diverges_from` names it: button B where button A is canon —
+            classify it: instance if it should use the canon, variant with
+            the why if deliberate); "recheck": judged instances/variants
+            whose body changed since judged (the judgment stands; confirm
+            it again with classify_shapes, or re-judge).
 
     Returns {"shapes": [...], "total": N} — total counts every match, not
-    just this page. Classify what you can judge with classify_shapes; a
-    repeating shape with NO recorded canon is a derive-one-first moment
-    (consolidate onto a reference, create_snippet it, then classify the
-    rest against it), never N loose classifications.
+    just this page. Each row's `classified_by` says who judged: agent /
+    audit / import are judgments; `mechanical` is the canonical stamp the
+    sync applies; `hook` is write-path EVIDENCE (#2791) — the session pulled
+    a snippet and then wrote code referencing/resembling it, so the shape
+    was stamped an instance with the evidence in `reason`. A hook row is
+    overridable by any classify_shapes call; it never overrides yours.
+
+    THE FAST PATH through a big todo is the proposer's queue: every coverage
+    refresh matches unclassified shapes against canon (strongest basis
+    first: same symbol elsewhere → textual containment → body references
+    the canon → signature resemblance → semantic) and attaches a
+    `proposal` to each row it can speak for. Review `proposal="canon"` by
+    snippet or directory, then confirm_shape_proposals the ones that hold —
+    hundreds at a time — and classify_shapes the rest (variant/exempt, or
+    instance of a different snippet). `proposal="derive"` lists the
+    derive-first candidates: a repeating shape with NO recorded canon is
+    never N loose classifications — consolidate onto a reference,
+    create_snippet it, then classify the group against it.
     """
     uid = current_user_id()
     rows, total = await shape_ledger_svc.list_project_shapes(
         uid, project_id,
         status=status, path=path, snippet_id=snippet_id,
         include_vanished=include_vanished, limit=limit, offset=offset,
+        proposal=proposal, flag=flag,
     )
     return {"shapes": [r.to_dict() for r in rows], "total": total}
 
 
+async def shape_history(
+    project_id: int, path: str, symbol: str = "", limit: int = 200
+) -> dict:
+    """What was used here, when, and why — a shape's (or a directory's)
+    history from the ledger (#2793).
+
+    `shapes` are the current rows at `path` (a file, or a directory and
+    everything beneath it; `symbol` narrows to one definition) with
+    first/last-seen commits, vanished_at, and the standing judgment;
+    `events` are the state changes, oldest first: `classified` (status,
+    snippet_id, who, why — one per judgment, so a shape that was an instance
+    of #N and later a variant of #M shows both), `vanished`, `reappeared`,
+    `drifted` (the body moved under a judgment; see list_shapes flag=
+    "recheck"). Each event carries the commit the tree was read at.
+
+    Read it as a timeline: "instance of #N from <first classified at>,
+    re-judged variant of #M at <at> because <reason>, vanished at <commit>".
+    Read-only; requires read access to the project.
+    """
+    uid = current_user_id()
+    return await shape_ledger_svc.shape_history(
+        uid, project_id, path, symbol=symbol, limit=limit
+    )
+
+
+async def confirm_shape_proposals(
+    project_id: int,
+    snippet_id: int = 0,
+    path: str = "",
+    basis: str = "",
+    min_score: float = 0.0,
+) -> dict:
+    """Confirm the proposer's canon proposals you have reviewed, in batch.
+
+    The machine proposes, judgment classifies (#2792): each matching row —
+    live, unclassified, carrying a `proposal` with a snippet_id — becomes
+    `instance` of that snippet, classified_by="agent", reason naming the
+    basis and score. Narrow to what you actually looked at: at least one of
+    snippet_id (confirm one canon's whole queue after reading its
+    `list_shapes(proposal="canon", ...)` page), path (a directory you
+    audited), or basis (e.g. "symbol" and "reference" are near-certain;
+    "semantic" deserves a look first) is required — a bare confirm-all is
+    not a judgment. min_score trims a basis's tail.
+
+    Proposals you do NOT confirm are judged with classify_shapes (variant,
+    exempt, or instance of a different snippet) — any judgment retires the
+    proposal. Requires write access. Returns {"confirmed": N}.
+    """
+    uid = current_user_id()
+    try:
+        return await shape_ledger_svc.confirm_proposals(
+            uid, project_id, snippet_id=snippet_id, path=path, basis=basis,
+            min_score=min_score,
+        )
+    except ValueError as exc:
+        return {"error": str(exc)}
+
+
 async def refresh_pattern_coverage(project_id: int) -> dict:
     """Seed or refresh the project's shape ledger NOW, and return the readout.
 
@@ -107,9 +195,17 @@ async def refresh_pattern_coverage(project_id: int) -> dict:
     owner adds one (Settings → Integrations → Git Forges); no served repo →
     bind_repo on a host a connection serves.
 
+    The refresh is also when the mechanical proposer runs (#2792): with the
+    repo bodies in hand it matches every changed unclassified shape against
+    canon and records proposals (see list_shapes proposal=), then regroups
+    the derive-first candidates. Semantic matching is capped per refresh, so
+    a large ledger's queue grows across refreshes rather than in one.
+
     Returns the accounting payload — total, accounted, counts by status,
-    unclassified, repos, largest_gaps — plus `pattern_coverage`, the same
-    one-line summary enter_project carries.
+    unclassified, repos, largest_gaps, `proposed` (canon proposals awaiting
+    confirmation), `derive_groups` (the biggest repeats-with-no-canon
+    families), `proposer` (what this refresh examined) — plus
+    `pattern_coverage`, the same one-line summary enter_project carries.
     """
     uid = current_user_id()
     coverage = await coverage_svc.refresh_for_caller(uid, project_id)
@@ -120,5 +216,8 @@ async def refresh_pattern_coverage(project_id: int) -> dict:
 
 
 def register(mcp) -> None:
-    for fn in (classify_shapes, list_shapes, refresh_pattern_coverage):
+    for fn in (
+        classify_shapes, list_shapes, refresh_pattern_coverage,
+        confirm_shape_proposals, shape_history,
+    ):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/models/__init__.py

@@ -44,6 +44,6 @@ from scribe.models.rulebook import (  # noqa: E402, F401
 )
 from scribe.models.repo_binding import RepoBinding  # noqa: E402, F401
 from scribe.models.forge_connection import ForgeConnection  # noqa: E402, F401
-from scribe.models.code_shape import CodeShape  # noqa: E402, F401
+from scribe.models.code_shape import CodeShape, CodeShapeEvent  # noqa: E402, F401
 from scribe.models.system import System, RecordSystem  # noqa: E402, F401
 from scribe.models.design_system import DesignSystem, DesignToken  # noqa: E402, F401

src/scribe/models/code_shape.py

@@ -3,6 +3,7 @@ from datetime import datetime
 from sqlalchemy import (
     BigInteger,
     DateTime,
+    Float,
     ForeignKey,
     Index,
     Integer,
@@ -19,6 +20,12 @@ from scribe.models.base import TimestampMixin
 SHAPE_STATUSES = ("canonical", "instance", "variant", "exempt", "unclassified")
 SHAPE_CLASSIFIERS = ("agent", "audit", "hook", "mechanical", "import")
 
+# How the mechanical proposer (#2792) arrived at a proposal, strongest first.
+# `derive` is the odd one out: not "this is an instance of #N" but "this
+# shape repeats with NO canon — derive one first" (note 2786's derive-first
+# rule), so it carries a group key instead of a snippet.
+PROPOSAL_BASES = ("symbol", "text", "reference", "signature", "semantic", "derive")
+
 
 class CodeShape(Base, TimestampMixin):
     """One extracted code shape and its classification against canon (#2787).
@@ -42,6 +49,29 @@ class CodeShape(Base, TimestampMixin):
     snippet_id is SET NULL on snippet deletion: the classification's target
     is gone but the judgment happened; the sync pass (step 2) re-files such
     rows as unclassified so they rejoin the todo instead of dangling.
+
+    `signature` / `body_sha` (#2792) are the shape's content fingerprint —
+    its definition line and a whitespace/comment-insensitive hash of its
+    block — refreshed by every sync. They are what the mechanical proposer
+    matches on and what a later drift recheck compares against; the ledger
+    still never stores code bodies.
+
+    `classified_sha` remembers the fingerprint a judgment was made at;
+    when a later sync sees the body change under an instance/variant, the
+    row is flagged `recheck_at` (the judgment stands, it just asks to be
+    confirmed again) and a `drifted` event is written. `diverges_from`
+    (#2793) is the button-B flag: 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 — "button B appeared where button
+    A is canon: divergence or variant? classify it." Both clear on judgment.
+
+    The proposal columns hold the proposer's standing suggestion for an
+    UNCLASSIFIED row: `proposed_snippet_id` + `proposal_basis` + score for
+    "looks like an instance of #N", or `proposal_basis="derive"` +
+    `proposal_group` for "repeats with no canon". `proposed_sha` is the
+    body_sha the row was last examined at, so a refresh re-examines only
+    what changed. A judgment clears the proposal — the machine proposes,
+    judgment classifies.
     """
 
     __tablename__ = "code_shapes"
@@ -52,6 +82,8 @@ class CodeShape(Base, TimestampMixin):
         ),
         Index("ix_code_shapes_project_status", "project_id", "status"),
         Index("ix_code_shapes_snippet", "snippet_id"),
+        Index("ix_code_shapes_proposed", "project_id", "proposed_snippet_id"),
+        Index("ix_code_shapes_diverges", "project_id", "diverges_from"),
     )
 
     id: Mapped[int] = mapped_column(primary_key=True)
@@ -76,6 +108,38 @@ class CodeShape(Base, TimestampMixin):
     vanished_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True
     )
+    signature: Mapped[str] = mapped_column(Text, default="")
+    body_sha: Mapped[str] = mapped_column(Text, default="")
+    proposed_snippet_id: Mapped[int | None] = mapped_column(
+        BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
+    )
+    proposal_basis: Mapped[str | None] = mapped_column(Text, nullable=True)
+    proposal_score: Mapped[float | None] = mapped_column(Float, nullable=True)
+    proposal_group: Mapped[str | None] = mapped_column(Text, nullable=True)
+    proposed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    proposed_sha: Mapped[str] = mapped_column(Text, default="")
+    classified_sha: Mapped[str] = mapped_column(Text, default="")
+    recheck_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    diverges_from: Mapped[int | None] = mapped_column(
+        BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
+    )
+
+    @property
+    def proposal(self) -> dict | None:
+        """The standing proposal as one object, or None when the proposer
+        has nothing to say about this row."""
+        if self.proposed_snippet_id is None and not self.proposal_group:
+            return None
+        out: dict = {"basis": self.proposal_basis, "score": self.proposal_score}
+        if self.proposed_snippet_id is not None:
+            out["snippet_id"] = self.proposed_snippet_id
+        if self.proposal_group:
+            out["group"] = self.proposal_group
+        return out
 
     def to_dict(self) -> dict:
         return {
@@ -93,6 +157,69 @@ class CodeShape(Base, TimestampMixin):
             "first_seen_commit": self.first_seen_commit,
             "last_seen_commit": self.last_seen_commit,
             "vanished_at": self.vanished_at.isoformat() if self.vanished_at else None,
+            "signature": self.signature,
+            "body_sha": self.body_sha,
+            "proposal": self.proposal,
+            "classified_sha": self.classified_sha,
+            "recheck_at": self.recheck_at.isoformat() if self.recheck_at else None,
+            "diverges_from": self.diverges_from,
             "created_at": self.created_at.isoformat(),
             "updated_at": self.updated_at.isoformat(),
         }
+
+
+# What a shape's history records (#2793). Not "appeared" — first_seen and
+# created_at already say that on the row; history is for what CHANGED:
+SHAPE_EVENTS = ("classified", "vanished", "reappeared", "drifted")
+
+
+class CodeShapeEvent(Base):
+    """One state change in a shape's life — the what-was-used-when record.
+
+    "We used #N here from <date>, #M replaced it at commit C, reason R" is a
+    question the ledger row alone cannot answer once it has moved on; this
+    table keeps each judgment (status, snippet, who, why, at which commit)
+    and each presence change (vanished / reappeared / drifted) as it
+    happened. Denormalised path/symbol/kind so a directory's history reads
+    without joining; `snippet_id` is deliberately FK-free — history outlives
+    the snippet it names, which is the point.
+    """
+
+    __tablename__ = "code_shape_events"
+    __table_args__ = (
+        Index("ix_code_shape_events_shape", "shape_id", "at"),
+        Index("ix_code_shape_events_project_path", "project_id", "path"),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    shape_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("code_shapes.id", ondelete="CASCADE"), nullable=False
+    )
+    project_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    path: Mapped[str] = mapped_column(Text, nullable=False)
+    symbol: Mapped[str] = mapped_column(Text, nullable=False)
+    kind: Mapped[str] = mapped_column(Text, nullable=False)
+    event: Mapped[str] = mapped_column(Text, nullable=False)
+    status: Mapped[str | None] = mapped_column(Text, nullable=True)
+    snippet_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    classified_by: Mapped[str | None] = mapped_column(Text, nullable=True)
+    reason: Mapped[str | None] = mapped_column(Text, nullable=True)
+    commit: Mapped[str] = mapped_column(Text, default="")
+    at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "shape_id": self.shape_id,
+            "project_id": self.project_id,
+            "path": self.path,
+            "symbol": self.symbol,
+            "kind": self.kind,
+            "event": self.event,
+            "status": self.status,
+            "snippet_id": self.snippet_id,
+            "classified_by": self.classified_by,
+            "reason": self.reason,
+            "commit": self.commit,
+            "at": self.at.isoformat(),
+        }

src/scribe/routes/plugin.py

@@ -127,6 +127,15 @@ async def write_path_prior_art():
                              surfaced. A separate channel on purpose: a reuse
                              hint shown early must not suppress the record-sync
                              nudge when the recorded file is edited later.
+        shapes (opt)       — comma-separated `kind:name` definitions the hook
+                             found in (or enclosing) the payload, kind being
+                             css|sym. The shape ledger's write-path feed
+                             (#2791): when the session recently PULLED a
+                             snippet this payload references or resembles,
+                             these land as instance rows (classified_by=hook).
+                             Honoured only for a caller allowed to write — a
+                             read-scoped key still gets the hint, and never
+                             changes accounting on a GET.
     """
     path = (request.args.get("path") or "").strip()
     code = request.args.get("code") or ""
@@ -149,14 +158,40 @@ async def write_path_prior_art():
         int(p) for p in (request.args.get("exclude_sync_ids") or "").split(",")
         if p.strip().isdigit()
     ]
+    shapes = _parse_shapes(request.args.get("shapes") or "")
+    api_key = getattr(g, "api_key", None)
+    may_stamp = api_key is None or getattr(api_key, "scope", "") == "write"
 
     result = await plugin_ctx_svc.build_write_path_hint(
         g.user.id, path, code=code, project_id=project_id,
         exclude_ids=exclude_ids, exclude_sync_ids=exclude_sync_ids,
+        stamp_shapes=shapes if may_stamp else None,
+        repo_key=repo_bindings_svc.normalize_repo_key(repo) if repo else "",
     )
     return jsonify(result)
 
 
+# The hook names at most a dozen definitions per write; anything past that is
+# a generated file, not a shape being instantiated.
+_SHAPES_CAP = 12
+
+
+def _parse_shapes(raw: str) -> list[tuple[str, str]]:
+    """`css:btn-primary,sym:onTrash` → [("css", "btn-primary"), ("sym", "onTrash")].
+    Unknown kinds and empty names are dropped, duplicates collapse, and the
+    list is capped — the hook's own cap, re-applied so the contract holds
+    for any caller."""
+    out: list[tuple[str, str]] = []
+    for part in raw.split(","):
+        kind, _sep, name = part.strip().partition(":")
+        kind, name = kind.strip(), name.strip()
+        if kind in ("css", "sym") and name and (kind, name) not in out:
+            out.append((kind, name))
+        if len(out) >= _SHAPES_CAP:
+            break
+    return out
+
+
 @plugin_bp.get("/processes")
 @login_required
 async def process_manifest():

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", ""),
@@ -1163,11 +1179,46 @@ async def _restore_v2(data: dict) -> dict:
                 first_seen_commit=cs_data.get("first_seen_commit", ""),
                 last_seen_commit=cs_data.get("last_seen_commit", ""),
                 vanished_at=_dt(cs_data["vanished_at"]) if cs_data.get("vanished_at") else None,
+                # Fingerprints restore; proposals (#2792) deliberately do not —
+                # they are mechanical, and the next refresh recomputes them
+                # 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

@@ -25,12 +25,14 @@ only ever reads the cache.
 """
 from __future__ import annotations
 
+import hashlib
 import io
 import json
 import logging
 import posixpath
 import re
 import tarfile
+from typing import NamedTuple
 from datetime import datetime, timedelta, timezone
 
 from scribe.services.forge import ForgeSelector, get_forges
@@ -91,6 +93,104 @@ _ARROW_RE = re.compile(
 )
 
 
+class Definition(NamedTuple):
+    """One extracted definition with its content fingerprint (#2792).
+
+    `signature` is the definition line itself; `body_sha` hashes the block
+    whitespace- and comment-insensitively; `body` is the block's text, held
+    only for the duration of a refresh (the proposer matches on it) and
+    never stored.
+    """
+
+    kind: str
+    name: str
+    signature: str
+    body_sha: str
+    body: str
+
+
+def _definition_on(raw: str) -> tuple[str, str] | None:
+    """The (kind, name) this one line defines, or None. First match wins —
+    the same order the hook's awk program tries."""
+    m = _CSS_RE.match(raw)
+    if m:
+        return ("css", m.group(1))
+    line = _MODIFIERS_RE.sub("", raw.lstrip())
+    if m := _GO_METHOD_RE.match(line):
+        return ("sym", m.group(1))
+    if m := _KEYWORD_RE.match(line):
+        name = m.group(1)
+        if name.startswith("__") and name.endswith("__"):
+            return None
+        return ("sym", name)
+    if m := _ARROW_RE.match(line):
+        return ("sym", m.group(1))
+    return None
+
+
+# A definition's block runs from its line until the next non-blank line at
+# its own indentation or shallower that is not a closer — so a Python def ends
+# at the next top-level statement, a braces block keeps its `}`, a CSS rule
+# keeps its `}`. Capped so a generated monolith can't make one shape's
+# fingerprint cover the file.
+_BLOCK_CAP = 120
+_CLOSERS = ("}", ")", "]", "end", "};", "});", ");", "})", "]);")
+# Lines that don't change what a shape IS: comments and decorators. Dropped
+# from the fingerprint so touching a comment above the next function doesn't
+# read as this one's body changing.
+_NOISE_PREFIXES = ("#", "//", "/*", "*", "*/", "@", "<!--", "-->")
+_SIGNATURE_CAP = 300
+
+
+def _indent(line: str) -> int:
+    return len(line) - len(line.lstrip())
+
+
+def _block_sha(lines: list[str]) -> str:
+    kept = [
+        " ".join(ln.split())
+        for ln in lines
+        if ln.strip() and not ln.lstrip().startswith(_NOISE_PREFIXES)
+    ]
+    return hashlib.sha1("\n".join(kept).encode("utf-8")).hexdigest()[:16]
+
+
+def extract_definitions(text: str) -> list[Definition]:
+    """Every definition this text makes, with signature + fingerprint.
+
+    Duplicate (kind, name) within one text collapse to the first — the
+    ledger's identity is per file, so a second definition of the same name
+    (an overload, a re-declaration) is the same shape to it.
+    """
+    lines = text.splitlines()
+    starts: list[tuple[int, str, str]] = []
+    for i, raw in enumerate(lines):
+        hit = _definition_on(raw)
+        if hit:
+            starts.append((i, hit[0], hit[1]))
+    seen: set[tuple[str, str]] = set()
+    out: list[Definition] = []
+    for i, kind, name in starts:
+        if (kind, name) in seen:
+            continue
+        seen.add((kind, name))
+        base = _indent(lines[i])
+        end = min(len(lines), i + _BLOCK_CAP)
+        for j in range(i + 1, min(len(lines), i + _BLOCK_CAP)):
+            ln = lines[j]
+            if not ln.strip():
+                continue
+            if _indent(ln) <= base and ln.strip() not in _CLOSERS:
+                end = j
+                break
+        block = lines[i:end]
+        out.append(Definition(
+            kind, name, lines[i].strip()[:_SIGNATURE_CAP], _block_sha(block),
+            "\n".join(block),
+        ))
+    return out
+
+
 def extract_shapes(text: str) -> list[tuple[str, str]]:
     """Every (kind, name) this text DEFINES — kind is "css" or "sym".
 
@@ -98,29 +198,7 @@ def extract_shapes(text: str) -> list[tuple[str, str]]:
     line, dunders are skipped (every class defines __init__ — guaranteed
     noise), duplicates within one text count once.
     """
-    seen: set[tuple[str, str]] = set()
-    out: list[tuple[str, str]] = []
-    for raw in text.splitlines():
-        m = _CSS_RE.match(raw)
-        if m:
-            shape = ("css", m.group(1))
-        else:
-            line = _MODIFIERS_RE.sub("", raw.lstrip())
-            if m := _GO_METHOD_RE.match(line):
-                shape = ("sym", m.group(1))
-            elif m := _KEYWORD_RE.match(line):
-                name = m.group(1)
-                if name.startswith("__") and name.endswith("__"):
-                    continue
-                shape = ("sym", name)
-            elif m := _ARROW_RE.match(line):
-                shape = ("sym", m.group(1))
-            else:
-                continue
-        if shape not in seen:
-            seen.add(shape)
-            out.append(shape)
-    return out
+    return [(d.kind, d.name) for d in extract_definitions(text)]
 
 
 def scannable(path: str) -> bool:
@@ -131,14 +209,33 @@ def scannable(path: str) -> bool:
     return not path.lower().endswith(_SKIP_SUFFIXES)
 
 
+class ArchiveShape(NamedTuple):
+    """A definition located in a repo archive — what the sync upserts and
+    the proposer matches. The leading (path, kind, name) triple is the
+    ledger identity; the rest is the fingerprint and the transient body."""
+
+    path: str
+    kind: str
+    name: str
+    signature: str
+    body_sha: str
+    body: str
+
+
 def shapes_from_archive(blob: bytes) -> list[tuple[str, str, str]]:
-    """(path, kind, name) for every definition in a repo tarball.
+    """(path, kind, name) for every definition in a repo tarball — the
+    identity view of definitions_from_archive."""
+    return [(d.path, d.kind, d.name) for d in definitions_from_archive(blob)]
+
+
+def definitions_from_archive(blob: bytes) -> list[ArchiveShape]:
+    """Every definition in a repo tarball, with its fingerprint and body.
 
     Forge archives wrap content in a single top-level directory (repo-ref/);
     that component is stripped so paths match recorded snippet locations,
     which are repo-relative. Non-UTF-8 files are binaries and skipped.
     """
-    shapes: list[tuple[str, str, str]] = []
+    shapes: list[ArchiveShape] = []
     with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar:
         for member in tar:
             if not member.isfile() or "/" not in member.name:
@@ -153,7 +250,10 @@ def shapes_from_archive(blob: bytes) -> list[tuple[str, str, str]]:
                 text = handle.read().decode("utf-8")
             except UnicodeDecodeError:
                 continue
-            shapes.extend((path, kind, name) for kind, name in extract_shapes(text))
+            shapes.extend(
+                ArchiveShape(path, d.kind, d.name, d.signature, d.body_sha, d.body)
+                for d in extract_definitions(text)
+            )
     return shapes
 
 
@@ -253,13 +353,17 @@ async def compute_coverage(
 
     served: list[tuple[str, str]] = []
     recorded = await _recorded_locations(user_id, project_id)
+    # The proposer's canon catalog, read once per refresh and shared across
+    # the project's repos (#2792).
+    canons = None
+    proposer_stats = {"examined": 0, "proposed": 0, "semantic_checked": 0}
     for key in await keys_for_project(user_id, project_id):
         hit = selector.resolve(key)
         if hit is None:
             continue  # bound to a host no connection serves
         forge, api_repo = hit
         ref = await forge.default_branch(api_repo)
-        shapes = shapes_from_archive(await forge.archive(api_repo, ref))
+        definitions = definitions_from_archive(await forge.archive(api_repo, ref))
         # The head commit is provenance sugar on the ledger rows; failing to
         # learn it must not fail the sync — the ref names the point well
         # enough and the row timestamps carry the when.
@@ -268,13 +372,43 @@ async def compute_coverage(
         except ForgeError:
             marker = ref
         await shape_ledger.sync_repo_shapes(
-            project_id, key, shapes, seen_marker=marker
+            project_id, key, definitions, seen_marker=marker
         )
         served.append((key, ref))
+        # Propose while the bodies are in hand — the one moment they exist.
+        # Canonical marking below only touches rows the proposer leaves
+        # alone (a canon's own location never gets a proposal), so the order
+        # is immaterial; the proposer must not be able to fail the refresh.
+        try:
+            if canons is None:
+                canons = await shape_ledger.canon_catalog(user_id)
+            stats = await shape_ledger.propose_for_repo(
+                user_id, project_id, key, definitions, canons=canons
+            )
+            for k in proposer_stats:
+                proposer_stats[k] += stats.get(k, 0)
+        except Exception:
+            logger.warning("shape proposer failed for %s", key, exc_info=True)
     if not served:
         return None
 
     await shape_ledger.mark_canonicals(project_id, recorded)
+    try:
+        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.
@@ -290,11 +424,23 @@ async def compute_coverage(
         agg["accounted"] += row.status != "unclassified"
 
     unclassified = counts.pop("unclassified")
+    proposals = shape_ledger.proposal_summary(rows)
+    divergence = shape_ledger.divergence_summary(rows)
     return {
         "total": len(rows),
         "accounted": len(rows) - unclassified,
         "unclassified": unclassified,
         "counts": counts,
+        # The proposer's standing (#2792): canon proposals awaiting a
+        # confirm, the largest derive-first groups, and what this refresh did.
+        "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,
@@ -438,7 +584,19 @@ def coverage_line(coverage: dict) -> str:
     unclassified = coverage.get("unclassified", 0)
     if unclassified:
         line += f"; {unclassified} unclassified"
+        standing = []
+        if coverage.get("proposed"):
+            standing.append(f"{coverage['proposed']} proposed")
+        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

@@ -27,6 +27,7 @@ from scribe.services import knowledge as knowledge_svc
 from scribe.services import notes as notes_svc
 from scribe.services import projects as projects_svc
 from scribe.services import rulebooks as rulebooks_svc
+from scribe.services import shape_ledger as shape_ledger_svc
 from scribe.services import snippets as snippets_svc
 from scribe.services.access import label_shared_items, owner_names_for
 from scribe.services.embeddings import semantic_search_notes
@@ -703,6 +704,8 @@ async def build_write_path_hint(
     project_id: int = 0,
     exclude_ids: list[int] | None = None,
     exclude_sync_ids: list[int] | None = None,
+    stamp_shapes: list[tuple[str, str]] | None = None,
+    repo_key: str = "",
 ) -> dict:
     """Prior-art hint for the plugin's PreToolUse hook on Write/Edit.
 
@@ -749,9 +752,20 @@ async def build_write_path_hint(
     un-scored surfacing now has its own home: every arm emits note_usage_events,
     tagged 'write_path_sync' vs 'write_path_place' vs 'write_path_semantic', so
     each claim's pull-through rate is measurable on its own.
+
+    ``stamp_shapes`` turns the same request into the ledger's write-path feed
+    (#2791): the (kind, name) definitions the hook saw in — or enclosing —
+    the payload. When the session has PULLED a snippet recently and this
+    payload references or resembles it, those shapes land as `instance` rows
+    (classified_by=hook, see shape_ledger.stamp_write_path_instances) and
+    the result's ``stamped`` lists them. The route passes it only for a
+    caller allowed to write — a read-scoped key gets the hint, never the
+    stamp. ``repo_key`` (the hook's remote, normalised) homes a provisional
+    row for a shape the ledger has not synced yet.
     """
     cfg = await get_writepath_config(user_id)
-    empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg}
+    empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg,
+             "stamped": [], "divergence": []}
     path = (path or "").strip()
     if not cfg["enabled"] or not path:
         return empty
@@ -800,6 +814,15 @@ async def build_write_path_hint(
         seen.add(nid)
         placed.append(("nearby", item))
 
+    # The stamping feed's "actually pulled it" half (#2791). Read once, before
+    # the semantic arm, because the arm's query doubles as the resemblance
+    # test: a pulled snippet this session already saw (so it sits in `seen`)
+    # must still be SCORED for this payload — it just isn't re-listed.
+    pulled: dict = {}
+    if stamp_shapes:
+        pulled = await shape_ledger_svc.recent_pulls(user_id)
+    resembles: dict[int, float] = {}
+
     # --- arm 2: by meaning ---
     scored: list[tuple[str, dict]] = []
     remaining = top_k - len(synced) - len(placed)
@@ -821,12 +844,15 @@ async def build_write_path_hint(
         query = concept_query(query) or query
     if remaining > 0 and query:
         t0 = time.perf_counter()
+        # Pulled-and-seen ids stay in the query (as evidence) but never in
+        # the menu — the dedup contract holds, the resemblance still lands.
+        pulled_seen = seen & set(pulled)
         hits = await semantic_search_notes(
             user_id, query,
-            limit=remaining,
+            limit=remaining + len(pulled_seen),
             threshold=cfg["threshold"],
             project_id=scope_project,
-            exclude_ids=seen,
+            exclude_ids=seen - pulled_seen,
             # Snippets AND recorded experience (#2246). This arm was
             # snippets-only, which is auto-inject's mistake inverted: an issue
             # saying "we tried this and it deadlocked", or a dev-log recording
@@ -845,6 +871,11 @@ async def build_write_path_hint(
             # the browse scope and never surfaces a one-to-one direct share.
             scope="browse",
         )
+        resembles = {
+            int(note.id): float(score) for score, note in hits
+            if int(note.id) in pulled
+        }
+        hits = [(s, n) for s, n in hits if int(n.id) not in seen][:remaining]
         record_retrieval(
             user_id=user_id, source="write_path", query=query,
             threshold=cfg["threshold"], limit=remaining,
@@ -879,7 +910,32 @@ async def build_write_path_hint(
                 ))
 
     menu = (placed + scored)[:max(0, top_k - len(synced))]
-    if not synced and not menu:
+
+    # The stamp runs whether or not anything is rendered — after dedup, the
+    # common case is a silent hint and a pulled canon being instantiated.
+    stamped: list[dict] = []
+    if stamp_shapes and pulled:
+        try:
+            stamped = await shape_ledger_svc.stamp_write_path_instances(
+                user_id, project_id, path=path, shapes=stamp_shapes,
+                code=code or "", pulled=pulled, resembles=resembles,
+                repo_key=repo_key,
+            )
+        except Exception:
+            logger.warning("Write-path ledger stamping failed", exc_info=True)
+    # 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({
@@ -943,6 +999,11 @@ async def build_write_path_hint(
         note_ids.append(int(item["id"]))
         lines.append(_prior_art_line(item, marker, owner, foreign_lang))
 
+    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
     # snippet surfaced BY PLACE left no trace anywhere, making the arm that
@@ -964,9 +1025,46 @@ async def build_write_path_hint(
         "note_ids": note_ids,
         "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."""
+    by_snippet: dict[int, list[str]] = {}
+    for row in stamped:
+        label = f".{row['symbol']}" if row["kind"] == "css" else row["symbol"]
+        by_snippet.setdefault(int(row["snippet_id"]), []).append(f"`{label}`")
+    parts = [
+        f"{', '.join(names)} → instance of #{sid}"
+        for sid, names in by_snippet.items()
+    ]
+    return (
+        f"> Shape accounting: recorded at `{path}` — {'; '.join(parts)} "
+        "(classified_by=hook: you pulled that snippet this session and this "
+        "code references/resembles it). Not an instance? `classify_shapes` "
+        "overrides a hook stamp."
+    )
+
+
 async def _topic_titles(topic_ids: set[int]) -> dict[int, str]:
     """Map topic_id -> title for the given ids (live topics only)."""
     if not topic_ids:

tests/test_integration_shape_classify.py

@@ -219,3 +219,404 @@ async def test_sync_refiles_rows_whose_snippet_was_purged(seeded):
         ))).scalar_one()
     assert row.status == "unclassified"
     assert row.snippet_id is None
+
+
+# --- #2791: the write-path feed lands hook evidence as rows -------------------
+
+
+@pytest.mark.integration
+async def test_write_path_stamp_is_evidence_that_yields_to_judgment(seeded):
+    """Pulled + referenced → every named shape of the snippet's kind becomes
+    an instance row, classified_by=hook, carrying the evidence as reason. A
+    later agent judgment on one of them stands against a re-stamp; the hook
+    may only overwrite nobody's judgment or its own. The outsider stamps
+    nothing (write-gated like every other ledger write)."""
+    from datetime import datetime, timezone
+
+    from scribe.services.shape_ledger import stamp_write_path_instances
+
+    owner, other, pid, sid = (
+        seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
+    )
+    pulled = {sid: datetime.now(timezone.utc)}
+    code = "app = factory()\nreturn app\n"          # references the snippet's symbol
+
+    assert await stamp_write_path_instances(
+        other, pid, path="src/app.py", shapes=[("sym", "make_app")],
+        code=code, pulled=pulled,
+    ) == []
+
+    stamped = await stamp_write_path_instances(
+        owner, pid, path="src/app.py",
+        shapes=[("sym", "make_app"), ("sym", "Config"), ("css", "nope")],
+        code=code, pulled=pulled,
+    )
+    assert {s["symbol"] for s in stamped} == {"make_app", "Config"}   # css skipped: no css canon
+    rows, _ = await list_project_shapes(owner, pid, snippet_id=sid)
+    by_symbol = {r.symbol: r for r in rows}
+    assert by_symbol["make_app"].status == "instance"
+    assert by_symbol["make_app"].classified_by == "hook"
+    assert by_symbol["make_app"].reason == f"hook: pulled #{sid}; payload references `factory`"
+
+    # A judgment lands; the next stamp must leave it alone but may re-stamp
+    # its own earlier row.
+    await classify_shapes(owner, pid, [
+        {"path": "src/app.py", "symbol": "make_app", "status": "exempt",
+         "reason": "the app factory is its own thing"},
+    ])
+    again = await stamp_write_path_instances(
+        owner, pid, path="src/app.py",
+        shapes=[("sym", "make_app"), ("sym", "Config")], code=code, pulled=pulled,
+    )
+    assert {s["symbol"] for s in again} == {"Config"}
+    rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
+    by_symbol = {r.symbol: r for r in rows}
+    assert by_symbol["make_app"].status == "exempt"
+    assert by_symbol["Config"].status == "instance"
+
+    # Neither pulled nor in play → nothing, even with shapes named.
+    assert await stamp_write_path_instances(
+        owner, pid, path="src/util.py", shapes=[("sym", "helper")],
+        code="print('unrelated')", pulled=pulled,
+    ) == []
+
+
+@pytest.mark.integration
+async def test_a_brand_new_shape_gets_a_provisional_row_the_sync_settles(seeded):
+    """The shape being written right now has no ledger row yet. With the
+    hook's repo key it gets a provisional one — seen markers empty — so the
+    stamp survives until the next sync, which confirms it (sets the marker)
+    or stamps it vanished. Without a repo key only existing rows are touched."""
+    from datetime import datetime, timezone
+
+    from scribe.services.shape_ledger import stamp_write_path_instances
+
+    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
+    pulled = {sid: datetime.now(timezone.utc)}
+    code = "def build():\n    return factory()\n"
+
+    assert await stamp_write_path_instances(
+        owner, pid, path="src/new.py", shapes=[("sym", "build")],
+        code=code, pulled=pulled,                              # no repo_key
+    ) == []
+    stamped = await stamp_write_path_instances(
+        owner, pid, path="src/new.py", shapes=[("sym", "build")],
+        code=code, pulled=pulled, repo_key=REPO,
+    )
+    assert [s["symbol"] for s in stamped] == ["build"]
+    async with async_session() as s:
+        row = (await s.execute(select(CodeShape).where(
+            CodeShape.project_id == pid, CodeShape.path == "src/new.py",
+        ))).scalar_one()
+        assert row.status == "instance" and row.classified_by == "hook"
+        # "Unset" is the column's empty default — the markers are non-null
+        # Text, and the sync is what first fills them.
+        assert row.first_seen_commit == "" and row.last_seen_commit == ""
+
+    # The sync sees the shape in the tree → confirmed, stamp intact.
+    await sync_repo_shapes(
+        pid, REPO, SHAPES + [("src/new.py", "sym", "build")], seen_marker="abc123",
+    )
+    rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
+    assert rows[0].status == "instance" and rows[0].last_seen_commit == "abc123"
+
+    # The sync no longer sees it → vanished, out of the live accounting.
+    await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="def456")
+    rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
+    assert rows == []
+    rows, _ = await list_project_shapes(owner, pid, path="src/new.py", include_vanished=True)
+    assert rows[0].vanished_at is not None
+
+
+@pytest.mark.integration
+async def test_recent_pulls_reads_the_usage_stream(seeded):
+    """The "actually pulled it" half is the PULLED usage event, inside the
+    window; a surfacing alone is not a pull."""
+    from datetime import datetime, timedelta, timezone
+
+    from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
+    from scribe.services.shape_ledger import recent_pulls
+
+    owner, sid = seeded["owner"], seeded["snippet"]
+    now = datetime.now(timezone.utc)
+    async with async_session() as s:
+        s.add_all([
+            NoteUsageEvent(user_id=owner, note_id=sid, event=PULLED, source="mcp_get_snippet"),
+            NoteUsageEvent(user_id=owner, note_id=sid + 1000, event=SURFACED, source="auto_inject"),
+            NoteUsageEvent(user_id=owner, note_id=sid + 2000, event=PULLED,
+                           source="mcp_get_snippet", created_at=now - timedelta(days=2)),
+        ])
+        await s.commit()
+    pulls = await recent_pulls(owner)
+    assert sid in pulls
+    assert sid + 1000 not in pulls
+    assert sid + 2000 not in pulls
+
+
+# --- #2792: the mechanical proposer against real rows -------------------------
+
+
+def _quiet_semantic():
+    """The semantic basis needs the embedder; these tests prove the other
+    bases and the bookkeeping, so it answers "nothing" here."""
+    from unittest.mock import AsyncMock, patch
+
+    from scribe.services import shape_ledger
+    return patch.object(shape_ledger, "_semantic_canon", AsyncMock(return_value=None))
+
+
+def _defs(*items):
+    """ArchiveShape-like records: (path, kind, name, signature, body_sha, body)."""
+    import hashlib
+    out = []
+    for path, kind, name, signature, body in items:
+        sha = hashlib.sha1(" ".join(body.split()).encode()).hexdigest()[:16]
+        out.append((path, kind, name, signature, sha, body))
+    return out
+
+
+@pytest.mark.integration
+async def test_proposer_proposes_and_confirm_classifies(seeded):
+    """Bodies in hand, the proposer records proposals on unclassified rows —
+    symbol (a second `factory` elsewhere), reference (a call site), and
+    nothing for the unrelated — skips rows whose content it already judged,
+    and a scoped confirm turns proposals into agent instances while a
+    classify on another retires its proposal."""
+    from scribe.services.shape_ledger import (
+        confirm_proposals, propose_for_repo,
+    )
+
+    owner, other, pid, sid = (
+        seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
+    )
+    defs = _defs(
+        ("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n    app = factory()\n    return app"),
+        ("src/app.py", "sym", "Config", "class Config:", "class Config:\n    debug = False"),
+        ("src/util.py", "sym", "helper", "def helper(x):", "def helper(x):\n    return x"),
+        ("src/dup.py", "sym", "factory", "def factory():", "def factory():\n    return 1"),
+        ("web/button.css", "css", "btn", ".btn {", ".btn {\n  color: red;\n}"),
+    )
+    await sync_repo_shapes(pid, REPO, defs, seen_marker="main")
+
+    with _quiet_semantic():
+        stats = await propose_for_repo(owner, pid, REPO, defs)
+    assert stats == {"examined": 5, "proposed": 2, "semantic_checked": 2}
+    rows, total = await list_project_shapes(owner, pid, proposal="canon")
+    by_symbol = {r.symbol: r for r in rows}
+    assert total == 2
+    assert by_symbol["factory"].proposal == {"basis": "symbol", "score": 1.0, "snippet_id": sid}
+    assert by_symbol["make_app"].proposal == {"basis": "reference", "score": 0.9, "snippet_id": sid}
+    rows, _ = await list_project_shapes(owner, pid, proposal="reference")
+    assert [r.symbol for r in rows] == ["make_app"]
+
+    # Same content again → nothing re-examined (the semantic cap would
+    # otherwise be spent on the same rows every refresh). A cap that leaves
+    # rows unreached leaves them UNexamined, so the next refresh gets them.
+    with _quiet_semantic():
+        assert (await propose_for_repo(owner, pid, REPO, defs))["examined"] == 0
+        await classify_shapes(owner, pid, [
+            {"path": "src/util.py", "symbol": "helper", "status": "unclassified"},
+        ])
+        assert (await propose_for_repo(owner, pid, REPO, defs, semantic_cap=0))["semantic_checked"] == 0
+        assert (await propose_for_repo(owner, pid, REPO, defs))["examined"] == 1
+
+    # Outsider can't confirm; the owner confirms by snippet, scoped.
+    with pytest.raises(ValueError):
+        await confirm_proposals(other, pid, snippet_id=sid)
+    assert await confirm_proposals(owner, pid, basis="symbol") == {"confirmed": 1}
+    rows, _ = await list_project_shapes(owner, pid, snippet_id=sid)
+    factory = next(r for r in rows if r.symbol == "factory")
+    assert factory.status == "instance" and factory.classified_by == "agent"
+    assert factory.reason == "confirmed symbol proposal (1.00)"
+    assert factory.proposal is None
+
+    # A judgment on a proposed row retires the proposal; withdrawing a
+    # judgment forgets the examination so the next pass proposes afresh.
+    await classify_shapes(owner, pid, [
+        {"path": "src/app.py", "symbol": "make_app", "status": "exempt", "reason": "bootstrap"},
+    ])
+    rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
+    make_app = next(r for r in rows if r.symbol == "make_app")
+    assert make_app.status == "exempt" and make_app.proposal is None
+    await classify_shapes(owner, pid, [
+        {"path": "src/app.py", "symbol": "make_app", "status": "unclassified"},
+    ])
+    with _quiet_semantic():
+        assert (await propose_for_repo(owner, pid, REPO, defs))["proposed"] == 1
+    rows, _ = await list_project_shapes(owner, pid, proposal="canon")
+    assert [r.symbol for r in rows] == ["make_app"]
+
+
+@pytest.mark.integration
+async def test_derive_groups_land_on_rows_and_in_the_summary(seeded):
+    """Shapes with no canon hit that repeat — identical bodies in two files,
+    the same name in three — carry a derive proposal, and the readout ranks
+    the families. A canon proposal keeps a row out of any derive group."""
+    from scribe.services.shape_ledger import (
+        apply_derive_groups, live_rows, propose_for_repo, proposal_summary,
+    )
+
+    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
+    defs = _defs(
+        ("a/one.py", "sym", "slug", "def slug(t):", "def slug(t):\n    return t.lower()"),
+        ("a/two.py", "sym", "slug", "def slug(t):", "def slug(t):\n    return t.lower()"),
+        ("b/x.css", "css", "card", ".card {", ".card { padding: 1px }"),
+        ("b/y.css", "css", "card", ".card {", ".card { padding: 2px }"),
+        ("b/z.css", "css", "card", ".card {", ".card { padding: 3px }"),
+        ("c/only.py", "sym", "alone", "def alone():", "def alone():\n    return 0"),
+        ("c/use.py", "sym", "boot", "def boot():", "def boot():\n    return factory()"),
+    )
+    await sync_repo_shapes(pid, REPO, defs, seen_marker="main")
+    with _quiet_semantic():
+        await propose_for_repo(owner, pid, REPO, defs)
+    assert await apply_derive_groups(pid) == 5
+
+    rows, total = await list_project_shapes(owner, pid, proposal="derive")
+    assert total == 5
+    groups = {(r.path, r.symbol): r.proposal for r in rows}
+    assert groups[("a/one.py", "slug")]["group"] == groups[("a/two.py", "slug")]["group"]
+    assert groups[("a/one.py", "slug")]["group"].startswith("dup:")
+    assert groups[("b/x.css", "card")] == {"basis": "derive", "score": 3.0, "group": "name:css:card"}
+    rows, _ = await list_project_shapes(owner, pid, proposal="any")
+    assert {r.symbol for r in rows} == {"slug", "card", "boot"}   # boot: reference proposal
+
+    summary = proposal_summary(await live_rows(pid))
+    assert summary["proposed"] == 1
+    assert [g["group"] for g in summary["derive_groups"]][0] == "name:css:card"
+    assert summary["derive_groups"][0]["label"] == ".card"
+    assert summary["derive_groups"][0]["size"] == 3
+
+    # One of the css copies gets judged → the group shrinks on the next pass.
+    await classify_shapes(owner, pid, [
+        {"path": "b/z.css", "symbol": "card", "status": "exempt", "reason": "print sheet"},
+    ])
+    await apply_derive_groups(pid)
+    rows, _ = await list_project_shapes(owner, pid, proposal="derive")
+    assert {r.symbol for r in rows} == {"slug"}                   # 2 files < the name floor
+
+
+# --- #2793: the divergence readout against real rows -------------------------
+
+
+@pytest.mark.integration
+async def test_a_second_confirm_dialog_is_detected_and_named(seeded):
+    """The milestone's acceptance case. A directory where one canon dominates
+    the judged siblings (a confirm helper with four instance call sites);
+    after a previous refresh, a new shape lands there that the proposer does
+    not match to the canon — it is flagged `diverges_from` the canon, the
+    readout names it, and the in-band check names it at write time. A
+    judgment clears the flag; a shape proposed AS the canon is not flagged."""
+    from datetime import datetime, timedelta, timezone
+
+    from scribe.services.shape_ledger import (
+        divergence_summary, flag_divergence, live_rows, propose_for_repo,
+        write_time_divergence,
+    )
+
+    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
+    comp = "frontend/src/components"
+    base = _defs(
+        *[(f"{comp}/{n}.vue", "sym", f"on{n}", f"async function on{n}() {{",
+           f"async function on{n}() {{\n  const ok = await factory();\n  if (!ok) return;\n}}")
+          for n in ("Trash", "Delete", "Remove", "Restore")],
+    )
+    await sync_repo_shapes(pid, REPO, base, seen_marker="aaa111")
+    await classify_shapes(owner, pid, [
+        {"path": f"{comp}/{n}.vue", "symbol": f"on{n}", "status": "instance", "snippet_id": sid}
+        for n in ("Trash", "Delete", "Remove", "Restore")
+    ], via="audit")
+    previous = datetime.now(timezone.utc)
+
+    # Button B: a hand-rolled confirm that never touches the canon, plus a
+    # proper new instance (references the canon → the proposer claims it).
+    later = base + _defs(
+        (f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
+         "function confirmDanger() {\n  return window.confirm('Really?');\n}"),
+        (f"{comp}/Proper.vue", "sym", "onPurge", "async function onPurge() {",
+         "async function onPurge() {\n  const ok = await factory();\n  if (!ok) return;\n}"),
+    )
+    await sync_repo_shapes(pid, REPO, later, seen_marker="bbb222")
+    with _quiet_semantic():
+        await propose_for_repo(owner, pid, REPO, later)
+    assert await flag_divergence(pid, since=None) == 0          # a first seed flags nothing
+    assert await flag_divergence(pid, since=previous - timedelta(seconds=1)) == 1
+
+    rows, total = await list_project_shapes(owner, pid, flag="divergence")
+    assert total == 1
+    assert rows[0].symbol == "confirmDanger" and rows[0].diverges_from == sid
+    summary = divergence_summary(await live_rows(pid))
+    assert summary["divergent"] == 1
+    assert summary["divergence"][0]["symbol"] == "confirmDanger"
+    assert summary["divergence"][0]["canon_snippet_id"] == sid
+
+    # In-band: the hook names the shape at write time → the check names the canon.
+    named = await write_time_divergence(
+        pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], stamped=[]
+    )
+    assert named == [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": sid,
+                      "instances": 4, "judged": 4}]
+    # ...but an already-judged shape, or one just stamped as the canon's
+    # instance, is not re-litigated.
+    assert await write_time_divergence(pid, f"{comp}/Trash.vue", [("sym", "onTrash")], stamped=[]) == []
+    assert await write_time_divergence(
+        pid, f"{comp}/New.vue", [("sym", "onNew")],
+        stamped=[{"symbol": "onNew", "kind": "sym", "snippet_id": sid}],
+    ) == []
+    # A directory with no dominant canon is silent.
+    assert await write_time_divergence(pid, "src/other.py", [("sym", "thing")], stamped=[]) == []
+
+    # The judgment answers the question and clears the flag.
+    await classify_shapes(owner, pid, [
+        {"path": f"{comp}/Danger.vue", "symbol": "confirmDanger", "status": "variant",
+         "snippet_id": sid, "reason": "native confirm is fine in the dev-only panel"},
+    ])
+    rows, total = await list_project_shapes(owner, pid, flag="divergence")
+    assert total == 0
+
+
+@pytest.mark.integration
+async def test_history_records_what_was_used_when_and_drift_asks_for_a_recheck(seeded):
+    from scribe.services.shape_ledger import shape_history
+
+    owner, other, pid, sid = (
+        seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
+    )
+    v1 = _defs(("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n    return factory()"))
+    await sync_repo_shapes(pid, REPO, v1, seen_marker="c1")
+    await classify_shapes(owner, pid, [
+        {"path": "src/app.py", "symbol": "make_app", "status": "instance", "snippet_id": sid},
+    ])
+    # The body moves under the judgment → drifted + recheck; re-judging clears it.
+    v2 = _defs(("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n    return factory(debug=True)"))
+    await sync_repo_shapes(pid, REPO, v2, seen_marker="c2")
+    rows, total = await list_project_shapes(owner, pid, flag="recheck")
+    assert total == 1 and rows[0].symbol == "make_app" and rows[0].status == "instance"
+    await classify_shapes(owner, pid, [
+        {"path": "src/app.py", "symbol": "make_app", "status": "variant", "snippet_id": sid,
+         "reason": "debug flag is deliberate here"},
+    ])
+    rows, total = await list_project_shapes(owner, pid, flag="recheck")
+    assert total == 0
+    # Then it vanishes from the tree.
+    await sync_repo_shapes(pid, REPO, [], seen_marker="c3")
+
+    history = await shape_history(owner, pid, "src/app.py", symbol="make_app")
+    shape = history["shapes"][0]
+    assert shape["status"] == "variant" and shape["vanished_at"] is not None
+    # The seeded fixture synced this row first (marker "main"); v1/v2 are
+    # later sightings — first_seen keeps the first.
+    assert shape["first_seen_commit"] == "main" and shape["last_seen_commit"] == "c2"
+    timeline = [(e["event"], e["status"], e["snippet_id"], e["commit"]) for e in history["events"]]
+    assert timeline == [
+        ("classified", "instance", sid, "c1"),
+        ("drifted", "instance", sid, "c2"),
+        ("classified", "variant", sid, "c2"),
+        ("vanished", "variant", sid, "c2"),
+    ]
+    assert history["events"][2]["reason"] == "debug flag is deliberate here"
+    assert history["events"][0]["classified_by"] == "agent"
+    # Directory-wide read works (the empty sync also vanished the seeded
+    # Config and helper rows under src/ — two more events); an outsider
+    # reads nothing.
+    assert len((await shape_history(owner, pid, "src"))["events"]) == 6
+    assert await shape_history(other, pid, "src/app.py") == {}

tests/test_pattern_coverage.py

@@ -456,3 +456,70 @@ async def test_unservable_binding_measures_nothing(seeded):
     await set_binding(uid, "https://github.com/somebody/else.git", other_pid)
 
     assert await compute_coverage(uid, other_pid, selector=_selector(_tarball(TREE))) is None
+
+
+# --- #2792: fingerprints and the proposer's readout --------------------------
+
+
+def test_extract_definitions_fingerprints_each_block():
+    """The block rule across the language families the extractor knows: a
+    Python def ends at the next top-level statement, a braces/CSS block keeps
+    its closer, and comments/decorators don't move the hash."""
+    from scribe.services.coverage import extract_definitions
+
+    text = (
+        "import os\n\n"
+        "def a(x):\n    # comment\n    return x + 1\n\n\n"
+        "class B:\n    def m(self):\n        return 2\n\n"
+        ".btn {\n  color: red;\n}\n"
+        "export const f = (x) => {\n  return x;\n};\n"
+    )
+    defs = {d.name: d for d in extract_definitions(text)}
+    assert set(defs) == {"a", "B", "m", "btn", "f"}
+    assert defs["a"].signature == "def a(x):"
+    assert defs["a"].body.startswith("def a(x):\n    # comment\n    return x + 1")
+    assert "class B" not in defs["a"].body
+    assert defs["B"].body.rstrip().endswith("return 2")
+    assert defs["btn"].body == ".btn {\n  color: red;\n}"
+    assert defs["f"].body == "export const f = (x) => {\n  return x;\n};"
+    assert all(len(d.body_sha) == 16 for d in defs.values())
+    # Comment changes don't change what the shape IS; code changes do.
+    again = {d.name: d for d in extract_definitions(text.replace("# comment", "# other"))}
+    assert again["a"].body_sha == defs["a"].body_sha
+    changed = {d.name: d for d in extract_definitions(text.replace("x + 1", "x + 2"))}
+    assert changed["a"].body_sha != defs["a"].body_sha
+    # And the identity view is unchanged for the hook mirror.
+    from scribe.services.coverage import extract_shapes
+    assert extract_shapes(text) == [(d.kind, d.name) for d in extract_definitions(text)]
+
+
+def test_coverage_line_names_the_proposers_standing():
+    from scribe.services.coverage import coverage_line
+
+    base = {
+        "total": 100, "accounted": 10, "unclassified": 90,
+        "counts": {"canonical": 10, "instance": 0, "variant": 0, "exempt": 0},
+        "computed_at": "2026-08-21T00:00:00+00:00",
+        "largest_gaps": [{"dir": "src", "unclassified": 90, "total": 90}],
+    }
+    assert coverage_line(base).endswith("; 90 unclassified, largest: src")
+    line = coverage_line({**base, "proposed": 40, "derive_groups": [{"group": "a"}, {"group": "b"}]})
+    assert "; 90 unclassified (40 proposed, 2 derive groups), largest: src" in line
+    line = coverage_line({**base, "proposed": 0, "derive_groups": [{"group": "a"}]})
+    assert "(1 derive group)" in line
+
+
+def test_coverage_line_names_divergence_and_recheck():
+    from scribe.services.coverage import coverage_line
+
+    base = {
+        "total": 100, "accounted": 40, "unclassified": 60,
+        "counts": {"canonical": 10, "instance": 30, "variant": 0, "exempt": 0},
+        "computed_at": "2026-08-21T00:00:00+00:00",
+        "largest_gaps": [{"dir": "src", "unclassified": 60, "total": 60}],
+    }
+    line = coverage_line({**base, "divergent": 2, "recheck": 1, "proposed": 5})
+    assert "; 60 unclassified (5 proposed, 2 DIVERGENT), largest: src" in line
+    assert line.endswith("; 1 judged shape changed since judged — recheck")
+    assert "DIVERGENT" not in coverage_line(base)
+    assert "recheck" not in coverage_line(base)

tests/test_services_backup.py

@@ -13,11 +13,12 @@ import pytest
 from scribe.services import backup
 
 
-def test_backup_version_is_v7():
-    """v7 added code_shapes (#2787). The bump is the point of the test —
-    a payload section added without moving the version produces backups that
-    are structurally different and indistinguishable by inspection."""
-    assert backup.BACKUP_VERSION == 7
+def test_backup_version_is_v8():
+    """v7 added code_shapes (#2787), v8 its history (#2793). The bump is the
+    point of the test — a payload section added without moving the version
+    produces backups that are structurally different and indistinguishable
+    by inspection."""
+    assert backup.BACKUP_VERSION == 8
 
 
 def test_not_included_lists_the_known_gaps():
@@ -114,7 +115,7 @@ async def test_export_full_backup_contains_every_declared_section():
                 "topic_suppressions",
                 "systems", "record_systems", "design_systems",
                 "design_tokens", "note_usage_events", "repo_bindings",
-                "note_supersessions", "code_shapes"):
+                "note_supersessions", "code_shapes", "code_shape_events"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []
 

tests/test_shape_ledger.py

@@ -5,6 +5,8 @@ token-free serialisation. The sync pass (step 2) and the classification
 surface (step 3) grow their tests here; DB-backed behavior lands in the
 integration lane once there is behavior to exercise.
 """
+import pytest
+
 from scribe.models import Base
 from scribe.models.code_shape import SHAPE_CLASSIFIERS, SHAPE_STATUSES, CodeShape
 
@@ -82,3 +84,261 @@ def test_classify_and_list_are_mounted_as_mcp_tools():
     mcp = build_mcp_server()
     for name in ("classify_shapes", "list_shapes", "refresh_pattern_coverage"):
         assert mcp._tool_manager.get_tool(name) is not None
+
+
+# --- step 5: the write-path feed's evidence tests (pure) ---------------------
+
+
+def test_symbol_reference_is_word_bounded_and_kind_aware():
+    from scribe.services.shape_ledger import references_symbol as ref
+
+    code = "const ok = await confirmed({ title: 'x' });\nif (!ok) return;"
+    assert ref(code, "confirmed", "sym")
+    assert not ref(code, "confirm", "sym")          # prefix never claims the call
+    assert not ref("", "confirmed", "sym")
+    assert not ref(code, "", "sym")
+    # CSS: the class as a selector or inside a class attribute; dashes are part
+    # of the name, so `btn` must not claim `btn-primary`.
+    html = '<button class="btn btn-primary">Go</button>'
+    assert ref(html, ".btn-primary", "css")
+    assert ref(html, "btn-primary", "css")
+    assert ref(".btn-primary { color: red }", ".btn-primary", "css")
+    assert not ref('<button class="btn-primary">', ".btn", "css")
+    assert ref('<button class="btn-primary btn">', ".btn", "css")
+
+
+def test_snippet_kind_reads_the_symbol_then_the_language():
+    from scribe.services.shape_ledger import snippet_kind
+
+    assert snippet_kind(".btn-primary", "css") == "css"
+    assert snippet_kind(".btn-primary", "") == "css"
+    assert snippet_kind("confirmed", "typescript") == "sym"
+    assert snippet_kind("", "scss") == "css"      # whole-stylesheet record
+    assert snippet_kind("", "python") == "sym"
+
+
+def test_route_shapes_param_parses_capped_and_deduped():
+    from scribe.routes.plugin import _SHAPES_CAP, _parse_shapes
+
+    assert _parse_shapes("css:btn-primary,sym:onTrash") == [
+        ("css", "btn-primary"), ("sym", "onTrash"),
+    ]
+    assert _parse_shapes(" sym:a , sym:a ,bogus:x,sym:,:,") == [("sym", "a")]
+    assert _parse_shapes("") == []
+    many = ",".join(f"sym:f{i}" for i in range(40))
+    assert len(_parse_shapes(many)) == _SHAPES_CAP
+
+
+def test_hook_is_a_server_internal_classifier():
+    """`hook` is in the status vocabulary but NOT a via a caller may claim —
+    a classify_shapes call saying via="hook" would launder judgment as
+    evidence (the reverse of the stamping rule's point)."""
+    from scribe.services.shape_ledger import _CALLER_VIAS
+
+    assert "hook" in SHAPE_CLASSIFIERS
+    assert "hook" not in _CALLER_VIAS
+
+
+# --- step 6: the mechanical proposer (pure) ---------------------------------
+
+
+def test_proposal_columns_and_vocabulary_are_pinned():
+    """Fingerprints + the proposer's standing suggestion live on the row; the
+    basis vocabulary is fixed, with `derive` the odd one out (a group, not a
+    snippet)."""
+    from scribe.models.code_shape import PROPOSAL_BASES
+
+    cols = CodeShape.__table__.c
+    for name in ("signature", "body_sha", "proposed_snippet_id", "proposal_basis",
+                 "proposal_score", "proposal_group", "proposed_at", "proposed_sha"):
+        assert name in cols, name
+    fk = next(iter(cols.proposed_snippet_id.foreign_keys))
+    assert fk.ondelete == "SET NULL" and fk.column.table.name == "notes"
+    assert "ix_code_shapes_proposed" in {ix.name for ix in CodeShape.__table__.indexes}
+    assert PROPOSAL_BASES == ("symbol", "text", "reference", "signature", "semantic", "derive")
+
+
+def test_row_proposal_property_is_one_object_or_none():
+    row = CodeShape(project_id=1, repo_key="r", path="a.py", symbol="f", kind="sym")
+    assert row.proposal is None
+    row.proposed_snippet_id, row.proposal_basis, row.proposal_score = 9, "symbol", 1.0
+    assert row.proposal == {"basis": "symbol", "score": 1.0, "snippet_id": 9}
+    row.proposed_snippet_id = None
+    row.proposal_basis, row.proposal_group, row.proposal_score = "derive", "dup:abc", 3.0
+    assert row.proposal == {"basis": "derive", "score": 3.0, "group": "dup:abc"}
+
+
+def test_signature_similarity_blanks_the_names():
+    from scribe.services.shape_ledger import signature_similarity as sim
+
+    a = "def move_event(project_id: int, event_id: int, after_id: int | None):"
+    b = "def move_beat(project_id: int, beat_id: int, after_id: int | None):"
+    assert sim(a, "move_event", b, "move_beat") > 0.85
+    assert sim(a, "move_event", "def export_pdf(manuscript, design, fonts):", "export_pdf") < 0.6
+    assert sim("", "x", b, "move_beat") == 0.0
+    # Trivial signatures resemble everything and mean nothing — floored out.
+    assert sim("def helper(x):", "helper", "def make_app():", "make_app") == 0.0
+
+
+def test_text_containment_is_whitespace_insensitive_with_a_floor():
+    from scribe.services.shape_ledger import text_contains
+
+    code = "const ok = await confirmed({ title: 'Delete?', confirmLabel: 'Delete' });\nif (!ok) return;"
+    body = "async function onDelete() {\n  const ok = await confirmed({\n    title: 'Delete?',\n    confirmLabel: 'Delete'\n  });\n  if (!ok) return;\n}"
+    assert text_contains(body, code)
+    assert text_contains(code, body)
+    assert not text_contains("x = 1", "x = 1")          # below the substance floor
+
+
+def _canon(sid, kind="sym", symbol="", locations=(), signature="", code="", project_id=0):
+    from scribe.services.shape_ledger import Canon, _norm_text
+    return Canon(sid, kind, symbol, tuple(locations), signature, _norm_text(code), project_id)
+
+
+def test_match_canon_prefers_the_shapes_own_project_on_a_tie():
+    """The same helper recorded in two projects: the shape's own project's
+    record is its canon; family canon elsewhere is the fallback."""
+    from scribe.services.shape_ledger import match_canon
+    family = _canon(3, "sym", "slugify", [("lib/text.py", "slugify")], project_id=1)
+    own = _canon(4, "sym", "slugify", [("src/util/text.py", "slugify")], project_id=2)
+    assert match_canon("sym", "src/other.py", "slugify", "def slugify(t):", "",
+                       [family, own], project_id=2) == (4, "symbol", 1.0)
+    assert match_canon("sym", "src/other.py", "slugify", "def slugify(t):", "",
+                       [family, own], project_id=1) == (3, "symbol", 1.0)
+    # No project given → first-best stands; nothing breaks.
+    assert match_canon("sym", "src/other.py", "slugify", "def slugify(t):", "",
+                       [family, own])[1] == "symbol"
+
+
+def test_match_canon_orders_bases_strongest_first_and_respects_kind():
+    from scribe.services.shape_ledger import match_canon
+
+    confirmed = _canon(
+        7, "sym", "confirmed", [("frontend/src/composables/useConfirm.ts", "confirmed")],
+        "export async function confirmed(opts: ConfirmOptions): Promise<boolean> {",
+        "export async function confirmed(opts: ConfirmOptions): Promise<boolean> { /* singleton */ }",
+    )
+    mover = _canon(
+        8, "sym", "move_beat", [("src/forge/plot.py", "move_beat")],
+        "def move_beat(project_id: int, beat_id: int, after_id: int | None) -> None:",
+    )
+    btn = _canon(9, "css", ".btn-primary", [("web/buttons.css", ".btn-primary")],
+                 ".btn-primary {", ".btn-primary { color: var(--action-primary); padding: 4px 8px; border-radius: 4px; }")
+    canons = [confirmed, mover, btn]
+
+    # symbol: a second `confirmed` defined elsewhere answers to #7 — but the
+    # canon's own location never does (that row is canonical, not a proposal).
+    assert match_canon("sym", "src/other.ts", "confirmed", "function confirmed() {", "", canons) == (7, "symbol", 1.0)
+    assert match_canon("sym", "frontend/src/composables/useConfirm.ts", "confirmed",
+                       "export async function confirmed(", "", canons) is None
+    # reference: a call site of the canon.
+    body = "async function onTrash() {\n  const ok = await confirmed({ title: 'x' });\n  if (!ok) return;\n}"
+    assert match_canon("sym", "c.vue", "onTrash", "async function onTrash() {", body, canons) == (7, "reference", 0.9)
+    # signature: the family shape, names blanked.
+    hit = match_canon("sym", "src/forge/timeline.py", "move_event",
+                      "def move_event(project_id: int, event_id: int, after_id: int | None) -> None:",
+                      "    pass", canons)
+    assert hit and hit[0] == 8 and hit[1] == "signature" and hit[2] >= 0.8
+    # text: the canon's code contains the shape's body (a css copy), kind-matched —
+    # the same text as a `sym` shape matches no css canon.
+    css_body = ".btn-primary { color: var(--action-primary); padding: 4px 8px; border-radius: 4px; }"
+    assert match_canon("css", "web/other.css", "btn-big", ".btn-big {", css_body, canons) == (9, "text", 0.95)
+    assert match_canon("sym", "web/other.css", "btn-big", ".btn-big {", css_body, canons) is None
+    # nothing in play
+    assert match_canon("sym", "x.py", "unrelated", "def unrelated(a, b, c, d, e):", "return 1", canons) is None
+
+
+def test_match_canon_symbol_beats_everything_including_css_copies():
+    """The previous test's css `btn-primary`-elsewhere case, stated plainly:
+    a second definition of the canon's own name is the symbol basis."""
+    from scribe.services.shape_ledger import match_canon
+    btn = _canon(9, "css", ".btn-primary", [("web/buttons.css", ".btn-primary")], ".btn-primary {", ".btn-primary { color: red; padding: 4px 8px; border-radius: 4px; }")
+    assert match_canon("css", "web/other.css", "btn-primary", ".btn-primary {", ".btn-primary { color: blue }", [btn]) == (9, "symbol", 1.0)
+
+
+def test_derive_groups_copy_before_name_with_floors():
+    from scribe.services.shape_ledger import derive_groups
+
+    rows = [
+        ("a.py", "sym", "helper", "sha1"), ("b.py", "sym", "helper", "sha1"),   # identical copies
+        ("c.py", "sym", "helper", "sha9"),                                        # same name, 3rd file
+        ("d.css", "css", "btn", "s1"), ("e.css", "css", "btn", "s2"), ("f.css", "css", "btn", "s3"),
+        ("g.py", "sym", "main", "s4"), ("h.py", "sym", "main", "s5"),            # only 2 files → no name group
+        ("i.py", "sym", "one", "s6"),
+    ]
+    g = derive_groups(rows)
+    assert g[("a.py", "sym", "helper")] == "dup:sha1" == g[("b.py", "sym", "helper")]
+    assert g[("c.py", "sym", "helper")] == "name:sym:helper"
+    assert g[("d.css", "css", "btn")] == "name:css:btn"
+    assert ("g.py", "sym", "main") not in g
+    assert ("i.py", "sym", "one") not in g
+
+
+def test_confirm_requires_a_named_scope():
+    import asyncio
+
+    from scribe.services.shape_ledger import confirm_proposals
+
+    with pytest.raises(ValueError) as err:
+        asyncio.run(confirm_proposals(1, 2))
+    assert "name what you reviewed" in str(err.value)
+
+
+def test_proposer_tools_are_mounted():
+    from scribe.mcp.server import build_mcp_server
+
+    mcp = build_mcp_server()
+    assert mcp._tool_manager.get_tool("confirm_shape_proposals") is not None
+    tool = mcp._tool_manager.get_tool("list_shapes")
+    assert "proposal" in tool.parameters.get("properties", {})
+
+
+# --- step 7: the divergence readout (pure) ----------------------------------
+
+
+def _row(path, kind="sym", status="unclassified", snippet_id=None):
+    r = CodeShape(project_id=1, repo_key="r", path=path, symbol=path.rsplit("/", 1)[-1], kind=kind)
+    r.status, r.snippet_id = status, snippet_id
+    return r
+
+
+def test_dominant_canon_needs_enough_judged_siblings_and_a_clear_majority():
+    from scribe.services.shape_ledger import dominant_canon
+
+    dense = [_row(f"c/{i}", status="instance", snippet_id=7) for i in range(4)] + [
+        _row("c/x", status="instance", snippet_id=8), _row("c/y")]
+    assert dominant_canon(dense) == (7, 4, 5)
+    sparse = [_row("c/a", status="instance", snippet_id=7), _row("c/b", status="instance", snippet_id=7)]
+    assert dominant_canon(sparse) is None                     # 2 judged < floor
+    split = [_row(f"c/{i}", status="instance", snippet_id=7) for i in range(2)] + [
+        _row(f"c/{i+5}", status="instance", snippet_id=8) for i in range(2)]
+    assert dominant_canon(split) is None                      # 50% < 60% share
+    # Variants are departures, not votes; canonical counts like an instance.
+    mixed = [_row("c/a", status="canonical", snippet_id=7)] + [
+        _row(f"c/{i}", status="instance", snippet_id=7) for i in range(2)] + [
+        _row("c/v", status="variant", snippet_id=9)]
+    assert dominant_canon(mixed) == (7, 3, 3)
+
+
+def test_history_and_readout_tools_are_mounted_and_shape_history_is_read_only():
+    from scribe.mcp.server import _READ_ONLY_TOOLS, build_mcp_server
+
+    mcp = build_mcp_server()
+    assert mcp._tool_manager.get_tool("shape_history") is not None
+    assert "shape_history" in _READ_ONLY_TOOLS
+    assert "flag" in mcp._tool_manager.get_tool("list_shapes").parameters.get("properties", {})
+
+
+def test_history_and_divergence_columns_are_pinned():
+    from scribe.models.code_shape import SHAPE_EVENTS, CodeShapeEvent
+
+    cols = CodeShape.__table__.c
+    for name in ("classified_sha", "recheck_at", "diverges_from"):
+        assert name in cols, name
+    assert "ix_code_shapes_diverges" in {ix.name for ix in CodeShape.__table__.indexes}
+    ev = CodeShapeEvent.__table__
+    fk = next(iter(ev.c.shape_id.foreign_keys))
+    assert fk.ondelete == "CASCADE" and fk.column.table.name == "code_shapes"
+    assert not ev.c.snippet_id.foreign_keys           # history outlives the snippet
+    assert SHAPE_EVENTS == ("classified", "vanished", "reappeared", "drifted")
+    assert "code_shape_events" in Base.metadata.tables

tests/test_write_path_trigger.py

@@ -778,11 +778,13 @@ def test_route_reads_every_arg_the_hook_sends():
 
     from scribe.routes import plugin as routes
     src = inspect.getsource(routes.write_path_prior_art)
-    for arg in ("path", "code", "repo", "project_id", "exclude_ids", "exclude_sync_ids"):
+    for arg in ("path", "code", "repo", "project_id", "exclude_ids",
+                "exclude_sync_ids", "shapes"):
         assert f'request.args.get("{arg}"' in src, f"route ignores {arg}"
 
     hook = HOOK.read_text()
-    for arg in ("path=", "code=", "repo=", "exclude_ids=", "exclude_sync_ids="):
+    for arg in ("path=", "code=", "repo=", "exclude_ids=", "exclude_sync_ids=",
+                "shapes="):
         assert arg in hook, f"hook never sends {arg}"
 
 
@@ -1009,3 +1011,279 @@ def test_local_arm_finds_duplicates_in_every_language_family(
     ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
     assert "already defined" in ctx
     assert "create_snippet" in ctx
+
+
+# --- #2791: the write-path feed — hook evidence lands as ledger rows ----------
+
+
+def _ts():
+    from datetime import datetime, timezone
+    return datetime.now(timezone.utc)
+
+
+@pytest.mark.asyncio
+async def test_stamping_needs_named_shapes_and_a_recent_pull():
+    """Offered-but-ignored stamps nothing: without a PULLED event there is no
+    evidence, and without the hook naming shapes there is nothing to stamp.
+    Neither case may even read the pull stream."""
+    from scribe.services import plugin_context as pc
+    pulls = AsyncMock(return_value={})
+    stamp = AsyncMock(return_value=[])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc.shape_ledger_svc, "recent_pulls", pulls), \
+         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp):
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, project_id=4)
+        assert out["stamped"] == []
+        pulls.assert_not_awaited()                       # no shapes → no read
+        out = await pc.build_write_path_hint(
+            1, "src/x.py", code=REAL_CODE, project_id=4,
+            stamp_shapes=[("sym", "debounce")],
+        )
+    pulls.assert_awaited_once()
+    stamp.assert_not_awaited()                           # shapes, but no pull
+    assert out["stamped"] == []
+
+
+@pytest.mark.asyncio
+async def test_a_pulled_snippet_already_seen_is_evidence_not_menu():
+    """The pulled-then-written flow IS the dedup-excluded flow: the hint offered
+    #7 earlier (so it sits in exclude_ids), the session pulled it, and now
+    writes code resembling it. #7 must be scored for this payload — and handed
+    to the stamp as resemblance — without being re-listed in the menu."""
+    from scribe.services import plugin_context as pc
+    search = AsyncMock(return_value=[(0.91, _note(7, "pulled")), (0.80, _note(8, "fresh"))])
+    stamp = AsyncMock(return_value=[{
+        "path": "src/x.py", "symbol": "debounce", "kind": "sym",
+        "snippet_id": 7, "reason": "hook: pulled #7; payload resembles it (0.91)",
+    }])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
+         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={7: _ts()})), \
+         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp):
+        out = await pc.build_write_path_hint(
+            1, "src/x.py", code=REAL_CODE, project_id=4, exclude_ids=[7],
+            stamp_shapes=[("sym", "debounce")], repo_key="git.example.com/a/b",
+        )
+    # The query kept #7 eligible (and widened the budget by one for it)...
+    kw = search.call_args.kwargs
+    assert 7 not in kw["exclude_ids"]
+    assert kw["limit"] == 4
+    # ...but the menu still honours the session dedup.
+    assert out["note_ids"] == [8]
+    assert "#7" not in "\n".join(
+        line for line in out["context"].splitlines() if "[similar" in line
+    )
+    # The stamp saw the pull and the resemblance score for this payload.
+    skw = stamp.call_args.kwargs
+    assert skw["pulled"] == {7: skw["pulled"][7]}
+    assert skw["resembles"] == {7: 0.91}
+    assert skw["shapes"] == [("sym", "debounce")]
+    assert skw["repo_key"] == "git.example.com/a/b"
+    assert out["stamped"][0]["snippet_id"] == 7
+    # And the session is told what landed, with the way to correct it.
+    assert "Shape accounting" in out["context"]
+    assert "`debounce` → instance of #7" in out["context"]
+    assert "classify_shapes" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_a_stamp_renders_even_when_the_hint_is_otherwise_silent():
+    """After dedup the common case is an empty hint; the stamp must still run
+    and still be reported — silence about accounting is how hook rows would
+    become invisible."""
+    from scribe.services import plugin_context as pc
+    stamped = [{"path": "web/b.css", "symbol": "btn-primary", "kind": "css",
+                "snippet_id": 5, "reason": "hook: pulled #5; payload references `btn-primary`"}]
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
+         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={5: _ts()})), \
+         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances",
+                      AsyncMock(return_value=stamped)):
+        out = await pc.build_write_path_hint(
+            1, "web/b.css", code=".btn-primary { color: red; }" * 4, project_id=4,
+            stamp_shapes=[("css", "btn-primary")],
+        )
+    assert out["note_ids"] == []
+    assert out["stamped"] == stamped
+    assert "`.btn-primary` → instance of #5" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_a_failing_stamp_does_not_sink_the_hint():
+    from scribe.services import plugin_context as pc
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets",
+                      AsyncMock(return_value=([_snippet_item(12, "records me")], 1))), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
+         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={12: _ts()})), \
+         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances",
+                      AsyncMock(side_effect=RuntimeError("ledger down"))):
+        out = await pc.build_write_path_hint(
+            1, "src/x.py", code=REAL_CODE, project_id=4, stamp_shapes=[("sym", "f")],
+        )
+    assert out["sync_note_ids"] == [12]
+    assert out["stamped"] == []
+
+
+def test_route_stamps_only_for_a_caller_allowed_to_write():
+    """A read-scoped key gets the hint — every plugin hook works on a read key
+    — but a GET must never change accounting for it. The route passes the
+    shapes through only when the key is write-scoped (or it's a session)."""
+    import inspect
+
+    from scribe.routes import plugin as routes
+    src = inspect.getsource(routes.write_path_prior_art)
+    assert 'request.args.get("shapes"' in src
+    assert '== "write"' in src
+    assert "stamp_shapes=shapes if may_stamp else None" in src
+    assert "normalize_repo_key(repo)" in src
+    hook = HOOK.read_text()
+    assert "&shapes=" in hook
+
+
+def test_hook_names_the_shapes_being_written():
+    """The feed's two inputs: every definition in the payload, or — for an Edit
+    that changes a body, not a signature — the definition enclosing the edit,
+    found by walking the target file upward from the edited lines."""
+    src = HOOK.read_text()
+    assert "scribe_defs()" in src                      # one extractor, two consumers
+    assert ".tool_input.old_string" in src             # the Edit's anchor
+    assert "| tac | scribe_defs | head -1" in src      # nearest definition above
+    # The ledger feed sends NAMES, never bodies, and stays on the one GET.
+    assert src.count("/api/plugin/prior-art?") == 1
+
+
+def _run_hook_against_sink(tmp_path, payload):
+    """Run the hook with SCRIBE_URL pointed at a throwaway local listener and
+    return the query the hook sent. Lets the shell be tested end to end —
+    the extraction, the encoding, the URL — without a Scribe instance."""
+    import http.server
+    import threading
+    import urllib.parse
+
+    seen: dict = {}
+
+    class _Sink(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            seen.update(urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query))
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(b'{"context":"","note_ids":[]}')
+
+        def log_message(self, *a):
+            pass
+
+    server = http.server.HTTPServer(("127.0.0.1", 0), _Sink)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{server.server_port}")
+        out = subprocess.run(
+            ["bash", str(HOOK)], input=json.dumps(payload),
+            capture_output=True, text=True, env=env,
+        )
+        assert out.returncode == 0, out.stderr
+    finally:
+        server.shutdown()
+        server.server_close()
+    return seen
+
+
+def test_hook_sends_every_definition_in_a_write(tmp_path):
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    env = _hook_runtime_env()
+    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
+    seen = _run_hook_against_sink(tmp_path, {
+        "session_id": "s-feed-w", "cwd": str(repo), "tool_name": "Write",
+        "tool_input": {
+            "file_path": str(repo / "new.ts"),
+            "content": "export async function onDelete(): Promise<void> {\n"
+                       "  const ok = await confirmed({ title: 'x' });\n}\n"
+                       ".btn-primary {\n  color: red;\n}\n",
+        },
+    })
+    assert seen["path"] == ["new.ts"]
+    assert seen["shapes"] == ["css:btn-primary,sym:onDelete"]
+
+
+def test_hook_sends_the_enclosing_definition_for_a_body_edit(tmp_path):
+    """An Edit to the inside of onTrash names no definition itself; the hook
+    must walk the file upward from the edited line and send onTrash."""
+    import shutil
+    if shutil.which("tac") is None:
+        pytest.skip("the enclosing-definition walk needs tac")
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    env = _hook_runtime_env()
+    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
+    target = repo / "comp.vue"
+    target.write_text(
+        "<script setup lang=\"ts\">\n"
+        "async function onTrash(): Promise<void> {\n"
+        "  const ok = await confirmed({ title: 'Move to the trash?' });\n"
+        "  if (!ok) return;\n"
+        "}\n"
+        "const other = () => {\n  return 1;\n};\n"
+        "</script>\n"
+    )
+    seen = _run_hook_against_sink(tmp_path, {
+        "session_id": "s-feed-e", "cwd": str(repo), "tool_name": "Edit",
+        "tool_input": {
+            "file_path": str(target),
+            "old_string": "  if (!ok) return;",
+            "new_string": "  if (!ok) return;\n  await guarded(() => store.trashNode(id));",
+        },
+    })
+    assert seen["shapes"] == ["sym:onTrash"]
+
+
+@pytest.mark.asyncio
+async def test_the_write_time_divergence_check_is_named_in_band():
+    """#2793: the hook named a shape at a path whose directory a canon
+    dominates, and the stamp didn't make it that canon's instance — the hint
+    must say so at the write, even when nothing else renders."""
+    from scribe.services import plugin_context as pc
+    div = [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": 2761,
+            "instances": 20, "judged": 21}]
+    check = AsyncMock(return_value=div)
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
+         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
+         patch.object(pc.shape_ledger_svc, "write_time_divergence", check):
+        out = await pc.build_write_path_hint(
+            1, "frontend/src/components/Danger.vue", code=REAL_CODE, project_id=24,
+            stamp_shapes=[("sym", "confirmDanger")],
+        )
+    check.assert_awaited_once_with(24, "frontend/src/components/Danger.vue",
+                                   [("sym", "confirmDanger")], [])
+    assert out["divergence"] == div
+    assert "Divergence check at `frontend/src/components/Danger.vue`" in out["context"]
+    assert "`confirmDanger` → #2761 (20 of 21 judged siblings are its instances)" in out["context"]
+    assert "variant" in out["context"]
+    # No project → no check at all.
+    check.reset_mock()
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
+         patch.object(pc.shape_ledger_svc, "write_time_divergence", check):
+        out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, stamp_shapes=[("sym", "f")])
+    check.assert_not_awaited()
+    assert out["divergence"] == []