feat(ledger): mechanical proposer — every refresh proposes instances against canon and groups derive-first candidates; agents confirm in batches (#2792, milestone 294 step 6)

Shapes now carry a content fingerprint (signature + whitespace/comment-
insensitive body_sha; migration 0080) and the proposer runs inside the
coverage refresh, the one moment bodies exist: symbol elsewhere → textual
containment → body references the canon → signature resemblance → semantic
(capped per refresh, unreached rows stay unexamined for the next). A hit is
a proposal on the row (proposed_snippet_id/basis/score), never a
classification; rows with no canon hit group by the derive-first rule
(identical body in ≥2 places, same name in ≥3 files) as proposal_basis=
derive + a group key. list_shapes(proposal=any|canon|derive|<basis>) is the
queue; confirm_shape_proposals(project_id, snippet_id|path|basis) confirms
in batches as agent instances; any classify_shapes/hook stamp retires the
proposal. Readout carries proposed + derive_groups (line, payload, card).
Plugin 0.1.35 (skill: the machine proposes, judgment classifies).

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

src/scribe/services/shape_ledger.py

@@ -21,9 +21,11 @@ an extracted shape.
 """
 from __future__ import annotations
 
+import difflib
 import logging
 import re
 from datetime import datetime, timedelta, timezone
+from typing import Iterable, NamedTuple
 
 from sqlalchemy import select
 
@@ -65,15 +67,17 @@ def location_covers(loc_path: str, loc_symbol: str, path: str, name: str) -> boo
 async def sync_repo_shapes(
     project_id: int,
     repo_key: str,
-    shapes: list[tuple[str, str, str]],
+    shapes: list,
     *,
     seen_marker: str,
 ) -> None:
-    """Upsert one repo's extracted (path, kind, name) shapes into the ledger.
+    """Upsert one repo's extracted shapes into the ledger.
 
-    ``seen_marker`` is the commit the archive was read at when the forge can
-    say, else the ref name — provenance sugar; the row timestamps carry the
-    when.
+    ``shapes`` are (path, kind, name) triples, or the richer ArchiveShape
+    records (#2792) whose 4th/5th fields — signature, body_sha — refresh the
+    row's content fingerprint. ``seen_marker`` is the commit the archive was
+    read at when the forge can say, else the ref name — provenance sugar;
+    the row timestamps carry the when.
     """
     now = datetime.now(timezone.utc)
     async with async_session() as session:
@@ -87,7 +91,10 @@ async def sync_repo_shapes(
         ).scalars().all()
         by_key = {(r.path, r.symbol, r.kind): r for r in rows}
         seen: set[tuple[str, str, str]] = set()
-        for path, kind, name in shapes:
+        for shape in shapes:
+            path, kind, name = shape[0], shape[1], shape[2]
+            signature = shape[3] if len(shape) > 3 else ""
+            body_sha = shape[4] if len(shape) > 4 else ""
             key = (path, name, kind)
             if key in seen:
                 continue
@@ -98,9 +105,14 @@ async def sync_repo_shapes(
                     project_id=project_id, repo_key=repo_key,
                     path=path, symbol=name, kind=kind,
                     first_seen_commit=seen_marker, last_seen_commit=seen_marker,
+                    signature=signature, body_sha=body_sha,
                 ))
                 continue
             row.last_seen_commit = seen_marker
+            if signature:
+                row.signature = signature
+            if body_sha:
+                row.body_sha = body_sha
             # A shape that vanished and came back is live again — the vanish
             # stays visible in history via updated_at, not as a dead flag.
             row.vanished_at = None
@@ -297,6 +309,10 @@ async def classify_shapes(
             status = item["status"]
             for row in matches:
                 row.status = status
+                # The machine proposes, judgment classifies: any judgment
+                # retires the standing proposal; a withdrawal also forgets
+                # the examination so the next refresh proposes afresh.
+                _clear_proposal(row, reexamine=(status == "unclassified"))
                 if status == "unclassified":
                     row.snippet_id = None
                     row.reason = None
@@ -324,12 +340,16 @@ async def list_project_shapes(
     include_vanished: bool = False,
     limit: int = 100,
     offset: int = 0,
+    proposal: str = "",
 ) -> tuple[list[CodeShape], int]:
     """A filtered page of a project's ledger, with the unfiltered-match total.
 
     ([], 0) when the caller can't read the project — the same silence every
     other project list gives. ``path`` matches the exact file or anything
-    beneath it, mirroring recorded-location semantics.
+    beneath it, mirroring recorded-location semantics. ``proposal`` narrows
+    to rows the proposer has spoken about: "any", "canon" (an instance-of-#N
+    suggestion), "derive" (a repeats-with-no-canon group), or one basis
+    name (symbol/reference/text/signature/semantic).
     """
     from sqlalchemy import func, or_
 
@@ -349,6 +369,17 @@ async def list_project_shapes(
         ))
     if snippet_id:
         conds.append(CodeShape.snippet_id == snippet_id)
+    if proposal == "any":
+        conds.append(or_(
+            CodeShape.proposed_snippet_id.isnot(None),
+            CodeShape.proposal_group.isnot(None),
+        ))
+    elif proposal == "canon":
+        conds.append(CodeShape.proposed_snippet_id.isnot(None))
+    elif proposal == "derive":
+        conds.append(CodeShape.proposal_group.isnot(None))
+    elif proposal:
+        conds.append(CodeShape.proposal_basis == proposal)
     async with async_session() as session:
         total = (
             await session.execute(
@@ -589,6 +620,7 @@ async def stamp_write_path_instances(
             row.reason = why
             row.classified_by = "hook"
             row.classified_at = now
+            _clear_proposal(row)
             stamped.append({
                 "path": path, "symbol": name, "kind": kind,
                 "snippet_id": sid, "reason": why,
@@ -596,3 +628,431 @@ async def stamp_write_path_instances(
         if stamped:
             await session.commit()
     return stamped
+
+
+# --- the mechanical proposer (#2792): the machine proposes, judgment classifies
+#
+# Runs inside the coverage refresh — the one moment the shape BODIES exist
+# (the archive is in memory; the ledger stores fingerprints, never code).
+# Over every live unclassified row whose content changed since it was last
+# examined, it tries, strongest first:
+#
+#   symbol     the shape bears a recorded canon's symbol at another location
+#              (a second definition of the canon's name — an instance or a
+#              duplicate to consolidate; either way, it answers to #N);
+#   text       whitespace-insensitive containment either way between the
+#              body and the canon's recorded code (git-grep, in effect —
+#              a verbatim copy, or the canon's own call-site example);
+#   reference  the body calls/uses a canon's symbol — the call-site shape
+#              the P7 backfill classified by hand (#2790);
+#   signature  a sym whose definition line, name blanked, resembles the
+#              canon's (the move_*/resequence family shape);
+#   semantic   the widest net: the body (concept-queried, as the write-path
+#              arm does) scores above the write-path threshold against a
+#              canon — capped per refresh, so the cost is bounded and a big
+#              ledger is worked through across refreshes.
+#
+# A hit is a PROPOSAL on the row (proposed_snippet_id/basis/score), never a
+# classification: an agent confirms in batches (confirm_proposals) or judges
+# otherwise (classify_shapes clears it). Rows with no canon hit are grouped
+# by the derive-first rule (note 2786): the same body fingerprint in ≥2
+# places, or the same name defined in ≥3 files, is "a repeating shape with
+# no canon — derive one first", carried as proposal_basis="derive" + a group
+# key so sessions see the consolidation candidates as one thing.
+
+# Derive-first floors. Identical bodies twice is already a copy; a bare name
+# needs more repetition before it reads as a family (setup/handler/register
+# recur by convention, not by duplication).
+_DERIVE_MIN_DUP = 2
+_DERIVE_MIN_NAME = 3
+# Semantic checks per repo per refresh — an embedding each (local fastembed),
+# bounded so a 4,000-row ledger is worked through over refreshes, not in one.
+_SEMANTIC_CAP = 150
+# Signature resemblance floor, name blanked (difflib ratio) — and a length
+# floor, because `def NAME():` resembles `def NAME(x):` at 0.95 while saying
+# nothing; a family shape has parameters to resemble.
+_SIGNATURE_FLOOR = 0.8
+_SIGNATURE_MIN_LEN = 30
+# Textual containment needs enough substance to mean anything.
+_TEXT_FLOOR = 40
+_BASIS_ORDER = ("symbol", "text", "reference", "signature")
+
+
+class Canon(NamedTuple):
+    snippet_id: int
+    kind: str
+    symbol: str
+    locations: tuple[tuple[str, str], ...]
+    signature: str
+    code_norm: str
+
+
+def _norm_text(text: str) -> str:
+    return " ".join((text or "").split())
+
+
+def signature_similarity(sig_a: str, name_a: str, sig_b: str, name_b: str) -> float:
+    """How alike two definition lines are once their own names are blanked —
+    `def move_event(project_id, event_id, after_id)` against
+    `def move_beat(project_id, beat_id, after_id)` reads high."""
+    a = _norm_text(sig_a)
+    b = _norm_text(sig_b)
+    if name_a:
+        a = a.replace(name_a, "NAME")
+    if name_b:
+        b = b.replace(name_b, "NAME")
+    if len(a) < _SIGNATURE_MIN_LEN or len(b) < _SIGNATURE_MIN_LEN:
+        return 0.0
+    return difflib.SequenceMatcher(None, a, b).ratio()
+
+
+def text_contains(body: str, code: str) -> bool:
+    """Whitespace-insensitive containment, either way, above a substance
+    floor — the proposer's git-grep."""
+    a = _norm_text(body)
+    b = _norm_text(code)
+    if len(a) < _TEXT_FLOOR or len(b) < _TEXT_FLOOR:
+        return False
+    return a in b or b in a
+
+
+def match_canon(
+    kind: str, path: str, symbol: str, signature: str, body: str,
+    canons: Iterable[Canon],
+) -> tuple[int, str, float] | None:
+    """The strongest (snippet_id, basis, score) a shape earns against the
+    canon catalog, by the non-semantic bases — or None. Strongest by
+    basis order, then by score within the basis."""
+    best: dict[str, tuple[float, int]] = {}
+
+    def offer(basis: str, score: float, sid: int) -> None:
+        cur = best.get(basis)
+        if cur is None or score > cur[0]:
+            best[basis] = (score, sid)
+
+    norm_sym = _norm_symbol(symbol)
+    for c in canons:
+        if c.kind != kind:
+            continue
+        if c.symbol and _norm_symbol(c.symbol) == norm_sym:
+            if not any(location_covers(lp, ls, path, symbol) for lp, ls in c.locations):
+                offer("symbol", 1.0, c.snippet_id)
+            continue  # its own location is canonical territory, not a proposal
+        if c.symbol and references_symbol(body, c.symbol, kind):
+            offer("reference", 0.9, c.snippet_id)
+        if c.code_norm and text_contains(body, c.code_norm):
+            offer("text", 0.95, c.snippet_id)
+        if kind == "sym" and c.signature:
+            ratio = signature_similarity(signature, symbol, c.signature, c.symbol)
+            if ratio >= _SIGNATURE_FLOOR:
+                offer("signature", round(ratio, 3), c.snippet_id)
+    for basis in _BASIS_ORDER:
+        if basis in best:
+            score, sid = best[basis]
+            return (sid, basis, score)
+    return None
+
+
+async def canon_catalog(user_id: int) -> list[Canon]:
+    """Every snippet this user can browse, as matchable canon — own projects
+    and shared ones alike, because family canon counts (note 2786)."""
+    from scribe.models.note import Note
+    from scribe.services.access import browsable_notes_clause
+    from scribe.services.coverage import extract_definitions
+    from scribe.services.snippets import SNIPPET_NOTE_TYPE, snippet_fields
+
+    async with async_session() as session:
+        notes = (
+            await session.execute(
+                select(Note).where(
+                    browsable_notes_clause(user_id),
+                    Note.note_type == SNIPPET_NOTE_TYPE,
+                    Note.deleted_at.is_(None),
+                )
+            )
+        ).scalars().all()
+    out: list[Canon] = []
+    for note in notes:
+        fields = snippet_fields(note)
+        symbol = (fields.get("symbol") or "").strip()
+        kind = snippet_kind(symbol, fields.get("language") or "")
+        code = fields.get("code") or ""
+        defs = extract_definitions(code)
+        signature = ""
+        if defs:
+            own = next(
+                (d for d in defs if _norm_symbol(d.name) == _norm_symbol(symbol)), defs[0]
+            )
+            signature = own.signature
+        out.append(Canon(
+            int(note.id), kind, symbol,
+            tuple(
+                ((loc.get("path") or ""), (loc.get("symbol") or ""))
+                for loc in fields.get("locations") or []
+            ),
+            signature, _norm_text(code),
+        ))
+    return out
+
+
+def _clear_proposal(row: CodeShape, *, reexamine: bool = False) -> None:
+    row.proposed_snippet_id = None
+    row.proposal_basis = None
+    row.proposal_score = None
+    row.proposal_group = None
+    if reexamine:
+        row.proposed_at = None
+        row.proposed_sha = ""
+
+
+def _substance(text: str) -> int:
+    return len("".join((text or "").split()))
+
+
+async def _semantic_canon(
+    user_id: int, body: str, allowed: set[int]
+) -> tuple[int, float] | None:
+    from scribe.services.embeddings import semantic_search_notes
+    from scribe.services.plugin_context import (
+        WRITEPATH_DEFAULT_THRESHOLD, WRITEPATH_MIN_CODE_CHARS, concept_query,
+    )
+
+    if _substance(body) < WRITEPATH_MIN_CODE_CHARS or not allowed:
+        return None
+    query = concept_query(body) or body
+    hits = await semantic_search_notes(
+        user_id, query, limit=3, threshold=WRITEPATH_DEFAULT_THRESHOLD,
+        note_type="snippet", scope="browse",
+    )
+    for score, note in hits:
+        if int(note.id) in allowed:
+            return int(note.id), round(float(score), 3)
+    return None
+
+
+async def propose_for_repo(
+    user_id: int,
+    project_id: int,
+    repo_key: str,
+    definitions: list,
+    *,
+    canons: list[Canon] | None = None,
+    semantic_cap: int = _SEMANTIC_CAP,
+) -> dict:
+    """Examine one repo's live unclassified rows against canon and record
+    proposals. ``definitions`` are the ArchiveShape records the sync just
+    upserted (their bodies are the matching material). Rows whose fingerprint
+    is unchanged since their last examination are skipped; rows that only
+    the capped semantic pass could not reach stay unexamined, so the next
+    refresh reaches the next slice. Returns counts."""
+    if canons is None:
+        canons = await canon_catalog(user_id)
+    by_key = {(d[0], d[1], d[2]): d for d in definitions}
+    sym_canon_ids = {c.snippet_id for c in canons if c.kind == "sym"}
+    now = datetime.now(timezone.utc)
+    examined = proposed = checked = 0
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                select(CodeShape).where(
+                    CodeShape.project_id == project_id,
+                    CodeShape.repo_key == repo_key,
+                    CodeShape.status == "unclassified",
+                    CodeShape.vanished_at.is_(None),
+                )
+            )
+        ).scalars().all()
+        semantic_todo: list[tuple[CodeShape, object]] = []
+        for row in rows:
+            d = by_key.get((row.path, row.kind, row.symbol))
+            if d is None:
+                continue
+            signature, body_sha, body = d[3], d[4], d[5]
+            if row.proposed_at is not None and row.proposed_sha == body_sha:
+                continue
+            examined += 1
+            group = row.proposal_group  # derive grouping is reassigned below
+            hit = match_canon(row.kind, row.path, row.symbol, signature, body, canons)
+            _clear_proposal(row)
+            row.proposal_group = group
+            row.proposed_at = now
+            row.proposed_sha = body_sha
+            if hit:
+                row.proposed_snippet_id, row.proposal_basis, row.proposal_score = hit
+                row.proposal_group = None
+                proposed += 1
+            elif row.kind == "sym":
+                semantic_todo.append((row, d))
+        for i, (row, d) in enumerate(semantic_todo):
+            if i >= semantic_cap:
+                # Not reached this refresh: leave it unexamined so the next
+                # refresh picks it up, rather than stamping a false "nothing".
+                row.proposed_at = None
+                row.proposed_sha = ""
+                continue
+            checked += 1
+            try:
+                found = await _semantic_canon(user_id, d[5], sym_canon_ids)
+            except Exception:
+                logger.warning("semantic proposal failed", exc_info=True)
+                found = None
+            if found:
+                row.proposed_snippet_id, row.proposal_score = found
+                row.proposal_basis = "semantic"
+                row.proposal_group = None
+                proposed += 1
+        await session.commit()
+    return {"examined": examined, "proposed": proposed, "semantic_checked": checked}
+
+
+def derive_groups(
+    rows: Iterable[tuple[str, str, str, str]]
+) -> dict[tuple[str, str, str], str]:
+    """The derive-first grouping over (path, kind, symbol, body_sha) rows
+    that matched no canon: {(path, kind, symbol): group_key}. Identical
+    bodies in ≥2 places group as `dup:<sha>`; the same name defined in ≥3
+    files groups as `name:<kind>:<symbol>`; a row joins at most one group,
+    the copy before the name."""
+    by_sha: dict[str, list[tuple[str, str, str]]] = {}
+    by_name: dict[tuple[str, str], list[tuple[str, str, str]]] = {}
+    for path, kind, symbol, sha in rows:
+        key = (path, kind, symbol)
+        if sha:
+            by_sha.setdefault(sha, []).append(key)
+        by_name.setdefault((kind, _norm_symbol(symbol)), []).append(key)
+    out: dict[tuple[str, str, str], str] = {}
+    for sha, keys in by_sha.items():
+        if len(set(keys)) >= _DERIVE_MIN_DUP:
+            for key in keys:
+                out.setdefault(key, f"dup:{sha}")
+    for (kind, symbol), keys in by_name.items():
+        if len({k[0] for k in keys}) >= _DERIVE_MIN_NAME:
+            for key in keys:
+                out.setdefault(key, f"name:{kind}:{symbol}")
+    return out
+
+
+async def apply_derive_groups(project_id: int) -> int:
+    """Recompute derive-first groups over the project's live unclassified
+    rows that carry no canon proposal; returns how many rows are grouped."""
+    now = datetime.now(timezone.utc)
+    grouped = 0
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                select(CodeShape).where(
+                    CodeShape.project_id == project_id,
+                    CodeShape.status == "unclassified",
+                    CodeShape.vanished_at.is_(None),
+                    CodeShape.proposed_snippet_id.is_(None),
+                )
+            )
+        ).scalars().all()
+        groups = derive_groups(
+            (r.path, r.kind, r.symbol, r.body_sha or "") for r in rows
+        )
+        sizes: dict[str, int] = {}
+        for g in groups.values():
+            sizes[g] = sizes.get(g, 0) + 1
+        for row in rows:
+            key = groups.get((row.path, row.kind, row.symbol))
+            if key:
+                row.proposal_basis = "derive"
+                row.proposal_group = key
+                row.proposal_score = float(sizes[key])
+                if row.proposed_at is None:
+                    row.proposed_at = now
+                grouped += 1
+            elif row.proposal_group:
+                row.proposal_basis = None
+                row.proposal_group = None
+                row.proposal_score = None
+        await session.commit()
+    return grouped
+
+
+def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict:
+    """The readout's view of the proposer's standing: how many canon
+    proposals await confirmation, and the largest derive-first groups."""
+    proposed = 0
+    groups: dict[str, dict] = {}
+    for row in rows:
+        if row.status != "unclassified":
+            continue
+        if row.proposed_snippet_id is not None:
+            proposed += 1
+        elif row.proposal_group:
+            g = groups.setdefault(row.proposal_group, {
+                "group": row.proposal_group, "kind": row.kind,
+                "label": (
+                    ("." if row.kind == "css" else "") + row.symbol
+                    if row.proposal_group.startswith("name:")
+                    else f"{row.symbol} (identical body)"
+                ),
+                "size": 0, "paths": [],
+            })
+            g["size"] += 1
+            if len(g["paths"]) < 3:
+                g["paths"].append(row.path)
+    ranked = sorted(groups.values(), key=lambda g: (-g["size"], g["group"]))
+    return {"proposed": proposed, "derive_groups": ranked[:top]}
+
+
+async def confirm_proposals(
+    user_id: int,
+    project_id: int,
+    *,
+    snippet_id: int = 0,
+    path: str = "",
+    basis: str = "",
+    min_score: float = 0.0,
+) -> dict:
+    """Turn reviewed canon proposals into `instance` rows, in one batch.
+
+    At least one of snippet_id / path / basis must narrow the batch — "confirm
+    everything proposed" without having looked is not a judgment. Each row
+    becomes instance-of-its-proposed-snippet, classified_by="agent", reason
+    naming the basis and score; the proposal is retired. Returns
+    {"confirmed": N}."""
+    from scribe.services import access
+
+    if not (snippet_id or path.strip() or basis.strip()):
+        raise ValueError(
+            "name what you reviewed: confirm by snippet_id, path, and/or basis"
+        )
+    if not await access.can_write_project(user_id, project_id):
+        raise ValueError(f"project {project_id} not found or no write access")
+    from sqlalchemy import or_
+
+    conds = [
+        CodeShape.project_id == project_id,
+        CodeShape.status == "unclassified",
+        CodeShape.vanished_at.is_(None),
+        CodeShape.proposed_snippet_id.isnot(None),
+    ]
+    if snippet_id:
+        conds.append(CodeShape.proposed_snippet_id == snippet_id)
+    if path.strip():
+        clean = path.strip().strip("/")
+        conds.append(or_(CodeShape.path == clean, CodeShape.path.like(clean + "/%")))
+    if basis.strip():
+        conds.append(CodeShape.proposal_basis == basis.strip())
+    now = datetime.now(timezone.utc)
+    confirmed = 0
+    async with async_session() as session:
+        rows = (await session.execute(select(CodeShape).where(*conds))).scalars().all()
+        for row in rows:
+            if (row.proposal_score or 0.0) < min_score:
+                continue
+            row.status = "instance"
+            row.snippet_id = row.proposed_snippet_id
+            row.reason = (
+                f"confirmed {row.proposal_basis} proposal"
+                f" ({(row.proposal_score or 0.0):.2f})"
+            )
+            row.classified_by = "agent"
+            row.classified_at = now
+            _clear_proposal(row)
+            confirmed += 1
+        await session.commit()
+    return {"confirmed": confirmed}