feat(ledger): uses edges — consumption is its own relation, conformance keeps one snippet_id (#2870, milestone 294)

A shape can follow one convention canon AND call several helper canons; the
row's single snippet_id made the 2026-08 audit pick (hash_token won, the
service-function convention lost), and hook evidence — pulled a snippet, then
wrote code naming it — was stamped as instance when it is a uses fact.

- code_shape_uses (migration 0084): shape → snippet, basis, evidence; unique
  per pair; cascades with both ends. USE_BASES: reference | hook | agent |
  audit | import. A judgment-grade basis overwrites a mechanical one, never
  the reverse.
- classify_shapes items and classify_shapes_by_rule take uses=[snippet ids]
  (targets validated like snippet_id; all-or-nothing).
- The write-path hook writes a uses edge for every pulled canon the payload
  names (the instance stamp is unchanged); the proposer writes a uses edge
  for every canon a body names (reference_canons: kind + language family +
  stoplist, same rules as the reference basis) — the mechanical form of
  "auto-confirm own-import references" deferred from #2871.
- list_shapes(uses=N) lists the consumers of a canon; get_snippet's consumer
  map gains `uses` beside instances/variants.

Operator decision on #2870 (2026-08-21): keep one snippet_id, add uses edges.

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

src/scribe/mcp/tools/shapes.py

@@ -44,6 +44,12 @@ async def classify_shapes(
             one-off-handler, test-helper, convention-plumbing, pure-helper,
             generated, script, typed-record) so the ledger can be filtered
             and aggregated by kind of one-off — the prose stays the record.
+            uses is an OPTIONAL list of snippet ids this shape CALLS (#2870):
+            conformance (status + snippet_id) says what shape it is, uses
+            says which canonical helpers it consumes — a service function
+            can be an instance of the service-function convention AND use
+            hash_token. Consumer maps are uses edges; list_shapes(uses=N)
+            and get_snippet's `uses` read them.
         via: Who is judging — "agent" (default), "audit" (a sweep), or
             "import" (carrying maps recorded elsewhere).
 
@@ -69,6 +75,7 @@ async def list_shapes(
     proposal: str = "",
     flag: str = "",
     compact: bool = False,
+    uses: int = 0,
 ) -> dict:
     """Read a project's shape ledger — `status="unclassified"` IS the todo.
 
@@ -84,7 +91,12 @@ async def list_shapes(
             classify_shapes judgment. The human todo is `unclassified`.
         path: exact file, or a directory — matches everything beneath it
             (the coverage line's "largest" dirs go straight in here).
-        snippet_id: rows classified against this snippet — a consumer map.
+        snippet_id: rows classified against this snippet (instance/variant
+            of it — conformance).
+        uses: rows that CALL this snippet (#2870) — the consumer map proper,
+            whatever shape each row is itself; edges come from judgments
+            (classify_shapes uses=), the write-path hook, and the proposer's
+            by-name reference hits.
         include_vanished: include shapes no longer in the tree (history).
         limit/offset: page through big ledgers (limit caps at 500).
         compact: rows as `path · symbol · kind · status · signature` plus
@@ -131,7 +143,7 @@ async def list_shapes(
         uid, project_id,
         status=status, path=path, snippet_id=snippet_id,
         include_vanished=include_vanished, limit=limit, offset=offset,
-        proposal=proposal, flag=flag,
+        proposal=proposal, flag=flag, uses=uses,
     )
     return {
         "shapes": [r.to_compact() if compact else r.to_dict() for r in rows],
@@ -150,6 +162,7 @@ async def classify_shapes_by_rule(
     via: str = "agent",
     include_judged: bool = False,
     reason_code: str = "",
+    uses: list[int] | None = None,
 ) -> dict:
     """The sweep form of classify_shapes: ONE judgment applied to every
     unclassified shape under a directory whose symbol matches a glob.
@@ -174,6 +187,8 @@ async def classify_shapes_by_rule(
         via: "agent" (default) | "audit" | "import".
         reason_code: Optional catalogue code beside the reason (see
             classify_shapes) — a sweep is exactly where one applies.
+        uses: Optional snippet ids every matched shape CALLS (#2870) — e.g.
+            "every *_scheduler.py symbol uses ScheduledJob".
         include_judged: By default only unjudged rows are touched —
             `unclassified` and the sync's mechanical `scoped` stamp — a
             sweep never silently overwrites a judgment. True re-judges every
@@ -191,7 +206,7 @@ async def classify_shapes_by_rule(
             uid, project_id, path=path, status=status, pattern=pattern,
             kind=kind, snippet_id=snippet_id or None, reason=reason or None,
             via=via, include_judged=include_judged,
-            reason_code=reason_code or None,
+            reason_code=reason_code or None, uses=uses or None,
         )
     except ValueError as exc:
         return {"error": str(exc)}

src/scribe/mcp/tools/snippets.py

@@ -245,6 +245,10 @@ async def get_snippet(snippet_id: int) -> dict:
         data["instances"] = consumers["instances"]
     if consumers["variants"]:
         data["variants"] = consumers["variants"]
+    if consumers.get("uses"):
+        # The call sites (#2870): shapes that use this snippet, whatever
+        # shape they are themselves.
+        data["uses"] = consumers["uses"]
     return data
 
 

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, CodeShapeEvent  # noqa: E402, F401
+from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse  # 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

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 
 from sqlalchemy import (
     BigInteger,
@@ -217,6 +217,54 @@ class CodeShape(Base, TimestampMixin):
         return out
 
 
+# How a uses edge was established (#2870): who/what said "this shape calls
+# that canon". `reference` is the proposer's mechanical by-name hit on the
+# body (language-gated, #2871); `hook` is write-path evidence (pulled the
+# snippet, then wrote code naming its symbol); agent/audit/import are
+# judgments carried on classify_shapes(..., uses=[...]).
+USE_BASES = ("reference", "hook", "agent", "audit", "import")
+
+
+class CodeShapeUse(Base):
+    """One consumption edge: shape → canonical snippet it calls/uses (#2870).
+
+    Conformance (CodeShape.status/snippet_id) answers "what shape is this";
+    this table answers "what does it use" — many per shape. A service function
+    that is an instance of the service-function convention AND a consumer of
+    hash_token has one snippet_id and one uses edge. Cascades with both ends:
+    a use of a deleted snippet is no longer a fact worth keeping.
+    """
+
+    __tablename__ = "code_shape_uses"
+    __table_args__ = (
+        UniqueConstraint("shape_id", "snippet_id", name="uq_code_shape_uses_shape_snippet"),
+        Index("ix_code_shape_uses_snippet", "snippet_id"),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    shape_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("code_shapes.id", ondelete="CASCADE"), nullable=False
+    )
+    snippet_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notes.id", ondelete="CASCADE"), nullable=False
+    )
+    basis: Mapped[str] = mapped_column(Text, nullable=False)
+    evidence: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "shape_id": self.shape_id,
+            "snippet_id": self.snippet_id,
+            "basis": self.basis,
+            "evidence": self.evidence,
+            "created_at": iso(self.created_at),
+        }
+
+
 # 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")

src/scribe/services/shape_ledger.py

@@ -30,7 +30,7 @@ from typing import Iterable, NamedTuple
 from sqlalchemy import select
 
 from scribe.models import async_session
-from scribe.models.code_shape import REASON_CODES, CodeShape, CodeShapeEvent
+from scribe.models.code_shape import REASON_CODES, CodeShape, CodeShapeEvent, CodeShapeUse
 from scribe.models.base import iso
 
 logger = logging.getLogger(__name__)
@@ -209,6 +209,59 @@ async def _judge(
     session.add(_event(row, "classified", at))
 
 
+async def record_uses(
+    session, row: CodeShape, snippet_ids, *, basis: str, evidence: str | None = None,
+) -> int:
+    """Upsert consumption edges shape → snippet (#2870). A judgment-grade
+    basis (agent/audit/import) overwrites a mechanical one (reference/hook)
+    on the same edge; mechanical never overwrites a judgment. Returns the
+    number of edges written or refreshed. The row must be persisted (flushed)
+    so it has an id."""
+    wanted = {int(x) for x in (snippet_ids or []) if x}
+    if not wanted:
+        return 0
+    if row.id is None:
+        session.add(row)
+        await session.flush()
+    existing = {
+        e.snippet_id: e
+        for e in (
+            await session.execute(
+                select(CodeShapeUse).where(CodeShapeUse.shape_id == row.id)
+            )
+        ).scalars().all()
+    }
+    judged = basis in _CALLER_VIAS
+    n = 0
+    for sid in wanted:
+        edge = existing.get(sid)
+        if edge is None:
+            session.add(CodeShapeUse(shape_id=row.id, snippet_id=sid, basis=basis, evidence=evidence))
+            n += 1
+        elif judged or edge.basis not in _CALLER_VIAS:
+            edge.basis, edge.evidence = basis, evidence
+            n += 1
+    return n
+
+
+async def uses_of(shape_ids) -> dict[int, list[CodeShapeUse]]:
+    """{shape_id: [edges]} for a set of rows — the read side of record_uses."""
+    ids = [int(x) for x in shape_ids if x]
+    if not ids:
+        return {}
+    async with async_session() as session:
+        edges = (
+            await session.execute(
+                select(CodeShapeUse).where(CodeShapeUse.shape_id.in_(ids))
+                .order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
+            )
+        ).scalars().all()
+    out: dict[int, list[CodeShapeUse]] = {}
+    for e in edges:
+        out.setdefault(e.shape_id, []).append(e)
+    return out
+
+
 async def mark_canonicals(
     project_id: int, recorded: list[tuple[int, str, str]]
 ) -> None:
@@ -321,6 +374,11 @@ def validate_classifications(items: list[dict]) -> str | None:
                 f"classifications[{i}]: unknown reason_code {code!r} "
                 f"(one of: {', '.join(REASON_CODES)})"
             )
+        uses = item.get("uses")
+        if uses is not None and (
+            not isinstance(uses, list) or not all(isinstance(u, int) and u > 0 for u in uses)
+        ):
+            return f"classifications[{i}]: uses must be a list of snippet ids"
     return None
 
 
@@ -359,6 +417,8 @@ async def classify_shapes(
         for item in classifications
         if item.get("status") in _NEEDS_TARGET
     }
+    for item in classifications:
+        target_ids.update(int(u) for u in (item.get("uses") or []))
     for sid in sorted(target_ids):
         if await snippets_svc.get_snippet(user_id, sid) is None:
             raise ValueError(f"snippet {sid} not found (or not readable)")
@@ -398,6 +458,9 @@ async def classify_shapes(
                     by=via, reason=item.get("reason"), at=now,
                     reason_code=item.get("reason_code"),
                 )
+                if item.get("uses"):
+                    await record_uses(session, row, item["uses"], basis=via,
+                                      evidence=item.get("reason"))
                 classified += 1
         await session.commit()
     return {"classified": classified, "unmatched": unmatched}
@@ -433,6 +496,7 @@ async def classify_shapes_where(
     via: str = "agent",
     include_judged: bool = False,
     reason_code: str | None = None,
+    uses: list[int] | None = None,
 ) -> dict:
     """The sweep form of classify_shapes (#2868): one judgment applied to
     every live row under ``path`` whose symbol matches ``pattern`` (and
@@ -453,7 +517,7 @@ async def classify_shapes_where(
         raise ValueError("canonical is the sync's stamp on a snippet's own location — a sweep cannot set it")
     probe = {"path": path, "symbol": "*", "status": status,
              "snippet_id": snippet_id or 0, "reason": reason or "",
-             "reason_code": reason_code or ""}
+             "reason_code": reason_code or "", "uses": uses}
     error = validate_classifications([probe])
     if error:
         raise ValueError(error.replace("classifications[0]", "rule"))
@@ -461,6 +525,9 @@ async def classify_shapes_where(
         raise ValueError(f"project {project_id} not found or no write access")
     if status in _NEEDS_TARGET and await snippets_svc.get_snippet(user_id, int(snippet_id)) is None:
         raise ValueError(f"snippet {snippet_id} not found (or not readable)")
+    for sid in sorted({int(u) for u in (uses or [])}):
+        if await snippets_svc.get_snippet(user_id, sid) is None:
+            raise ValueError(f"snippet {sid} not found (or not readable)")
 
     now = datetime.now(timezone.utc)
     judged: list[str] = []
@@ -477,6 +544,8 @@ async def classify_shapes_where(
                 snippet_id=int(snippet_id) if status in _NEEDS_TARGET else None,
                 by=via, reason=reason, at=now, reason_code=reason_code,
             )
+            if uses:
+                await record_uses(session, row, uses, basis=via, evidence=reason)
             judged.append(f"{row.path}::{row.symbol}")
         await session.commit()
     return {"classified": len(judged), "sample": sorted(judged)[:12]}
@@ -494,6 +563,7 @@ async def list_project_shapes(
     offset: int = 0,
     proposal: str = "",
     flag: str = "",
+    uses: int = 0,
 ) -> tuple[list[CodeShape], int]:
     """A filtered page of a project's ledger, with the unfiltered-match total.
 
@@ -539,6 +609,12 @@ async def list_project_shapes(
         conds.append(CodeShape.diverges_from.isnot(None))
     elif flag == "recheck":
         conds.append(CodeShape.recheck_at.isnot(None))
+    if uses:
+        # Consumers of a canon (#2870): rows with a uses edge to it, whatever
+        # shape they themselves are.
+        conds.append(CodeShape.id.in_(
+            select(CodeShapeUse.shape_id).where(CodeShapeUse.snippet_id == uses)
+        ))
     async with async_session() as session:
         total = (
             await session.execute(
@@ -589,18 +665,38 @@ async def snippet_consumers(user_id: int, note_id: int) -> dict:
                 )
             )
         ).scalars().all()
-    readable: dict[int, bool] = {}
-    out: dict[str, list[dict]] = {"instances": [], "variants": []}
-    for row in rows:
-        if row.project_id not in readable:
-            readable[row.project_id] = await access.can_read_project(
-                user_id, row.project_id
+        # The consumption edges (#2870): rows that USE this canon, whatever
+        # shape they are themselves — the call-site map.
+        using = (
+            await session.execute(
+                select(CodeShape, CodeShapeUse.basis, CodeShapeUse.evidence)
+                .join(CodeShapeUse, CodeShapeUse.shape_id == CodeShape.id)
+                .where(CodeShapeUse.snippet_id == note_id, CodeShape.vanished_at.is_(None))
             )
-        if not readable[row.project_id]:
+        ).all()
+    readable: dict[int, bool] = {}
+
+    async def can_read(pid: int) -> bool:
+        if pid not in readable:
+            readable[pid] = await access.can_read_project(user_id, pid)
+        return readable[pid]
+
+    out: dict[str, list[dict]] = {"instances": [], "variants": [], "uses": []}
+    for row in rows:
+        if not await can_read(row.project_id):
             continue
         out["instances" if row.status == "instance" else "variants"].append(
             _consumer_dict(row)
         )
+    for row, basis, evidence in using:
+        if not await can_read(row.project_id):
+            continue
+        d = _consumer_dict(row)
+        d["basis"] = basis
+        if evidence:
+            d["evidence"] = evidence
+        d.pop("reason", None)
+        out["uses"].append(d)
     return out
 
 
@@ -780,6 +876,13 @@ async def stamp_write_path_instances(
                 "path": path, "symbol": name, "kind": kind,
                 "snippet_id": sid, "reason": why,
             })
+            # Every pulled canon the payload NAMES is a uses edge (#2870) — the
+            # call-site fact, independent of which one the row is judged to be.
+            await record_uses(
+                session, row,
+                [s_id for rank, _at, s_id, _why in bucket if rank == 2],
+                basis="hook", evidence="write path: pulled the snippet, payload names its symbol",
+            )
         if stamped:
             await session.commit()
     return stamped
@@ -940,6 +1043,26 @@ def text_contains(body: str, code: str) -> bool:
     return a in b or b in a
 
 
+def reference_canons(kind: str, path: str, symbol: str, body: str, canons: Iterable[Canon]) -> list[int]:
+    """Every canon this body NAMES (#2870) — the uses edges the proposer can
+    write mechanically: same kind, same language family, symbol not in the
+    generic-verb stoplist, and not the shape's own name."""
+    norm_sym = _norm_symbol(symbol)
+    out: list[int] = []
+    for c in canons:
+        if c.kind != kind or not c.symbol:
+            continue
+        if kind == "sym" and not same_family(path, c.language):
+            continue
+        if _norm_symbol(c.symbol) == norm_sym:
+            continue
+        if _norm_symbol(c.symbol).lower() in _REFERENCE_STOPLIST:
+            continue
+        if references_symbol(body, c.symbol, kind):
+            out.append(c.snippet_id)
+    return out
+
+
 def match_canon(
     kind: str, path: str, symbol: str, signature: str, body: str,
     canons: Iterable[Canon], *, project_id: int = 0,
@@ -1126,6 +1249,12 @@ async def propose_for_repo(
             row.proposal_group = group
             row.proposed_at = now
             row.proposed_sha = examined_as
+            # Consumption is recorded for every canon the body names (#2870),
+            # whatever the row is then judged to be.
+            used = reference_canons(row.kind, row.path, row.symbol, body, canons)
+            if used:
+                await record_uses(session, row, used, basis="reference",
+                                  evidence="proposer: body names the canon's symbol")
             if hit:
                 row.proposed_snippet_id, row.proposal_basis, row.proposal_score = hit
                 row.proposal_group = None