From d4c7b0e48d8513441fc1a92767f35db3a3490722 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 21 Aug 2026 15:18:25 -0400 Subject: [PATCH] =?UTF-8?q?feat(ledger):=20`uses`=20edges=20=E2=80=94=20co?= =?UTF-8?q?nsumption=20is=20its=20own=20relation,=20conformance=20keeps=20?= =?UTF-8?q?one=20snippet=5Fid=20(#2870,=20milestone=20294)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- alembic/versions/0084_code_shape_uses.py | 40 ++++++ src/scribe/mcp/tools/shapes.py | 21 +++- src/scribe/mcp/tools/snippets.py | 4 + src/scribe/models/__init__.py | 2 +- src/scribe/models/code_shape.py | 50 +++++++- src/scribe/services/shape_ledger.py | 147 +++++++++++++++++++++-- tests/test_integration_shape_classify.py | 35 ++++++ tests/test_shape_ledger.py | 29 +++++ 8 files changed, 314 insertions(+), 14 deletions(-) create mode 100644 alembic/versions/0084_code_shape_uses.py diff --git a/alembic/versions/0084_code_shape_uses.py b/alembic/versions/0084_code_shape_uses.py new file mode 100644 index 0000000..1da91e6 --- /dev/null +++ b/alembic/versions/0084_code_shape_uses.py @@ -0,0 +1,40 @@ +"""code_shape_uses — consumption edges, separate from conformance (#2870, milestone 294) + +Revision ID: 0084 +Revises: 0083 +Create Date: 2026-08-21 + +A ledger row carries ONE snippet_id: what shape this is (instance/variant of +a canon). But a shape can also CALL several canonical helpers — a service +function that is an instance of the service-function convention and a +consumer of hash_token. The 2026-08 audit had to pick one; hook evidence +("pulled #N then wrote code referencing it") was stamped as instance when it +is a uses fact. This table holds the many-valued relation: shape → snippet, +with the basis and the evidence. Cascades with the shape and the snippet. +""" +import sqlalchemy as sa +from alembic import op + +revision = "0084" +down_revision = "0083" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "code_shape_uses", + 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("snippet_id", sa.Integer(), sa.ForeignKey("notes.id", ondelete="CASCADE"), nullable=False), + sa.Column("basis", sa.Text(), nullable=False), + sa.Column("evidence", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")), + sa.UniqueConstraint("shape_id", "snippet_id", name="uq_code_shape_uses_shape_snippet"), + ) + op.create_index("ix_code_shape_uses_snippet", "code_shape_uses", ["snippet_id"]) + + +def downgrade() -> None: + op.drop_index("ix_code_shape_uses_snippet", table_name="code_shape_uses") + op.drop_table("code_shape_uses") diff --git a/src/scribe/mcp/tools/shapes.py b/src/scribe/mcp/tools/shapes.py index c5e9349..753a597 100644 --- a/src/scribe/mcp/tools/shapes.py +++ b/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)} diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py index 748438e..78425ce 100644 --- a/src/scribe/mcp/tools/snippets.py +++ b/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 diff --git a/src/scribe/models/__init__.py b/src/scribe/models/__init__.py index 56ec114..f5e11b8 100644 --- a/src/scribe/models/__init__.py +++ b/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 diff --git a/src/scribe/models/code_shape.py b/src/scribe/models/code_shape.py index 4d3010a..4929086 100644 --- a/src/scribe/models/code_shape.py +++ b/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") diff --git a/src/scribe/services/shape_ledger.py b/src/scribe/services/shape_ledger.py index 806c0e0..3bd1956 100644 --- a/src/scribe/services/shape_ledger.py +++ b/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 diff --git a/tests/test_integration_shape_classify.py b/tests/test_integration_shape_classify.py index c35b505..93c63a7 100644 --- a/tests/test_integration_shape_classify.py +++ b/tests/test_integration_shape_classify.py @@ -201,6 +201,41 @@ async def test_sync_stamps_scoped_rows_and_unstamps_when_they_become_reachable(s assert by_symbol["card"].status == "instance" and by_symbol["card"].snippet_id == sid +@pytest.mark.integration +async def test_uses_edges_are_the_consumer_map(seeded): + """#2870: a shape keeps ONE snippet_id (what it is) and any number of + uses edges (what it calls); the snippet's consumer map lists them, + list_shapes(uses=N) finds them, and a sweep can write them.""" + from scribe.services import snippets as snippets_svc + owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] + helper = await snippets_svc.create_snippet( + owner, name="cls_hash_helper", code="def hash_token(raw):\n return raw\n", + language="python", repo="Widget", path="src/hash.py", symbol="hash_token", + project_id=pid, + ) + hid = int(helper.id) + out = await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "instance", + "snippet_id": sid, "uses": [hid]}, + ], via="audit") + assert out["classified"] == 1 + rows, total = await list_project_shapes(owner, pid, uses=hid) + assert total == 1 and rows[0].symbol == "make_app" and rows[0].snippet_id == sid + consumers = await snippet_consumers(owner, hid) + assert consumers["instances"] == [] and len(consumers["uses"]) == 1 + assert consumers["uses"][0]["symbol"] == "make_app" and consumers["uses"][0]["basis"] == "audit" + # A sweep writes uses too; an unknown snippet in uses applies nothing. + out = await classify_shapes_where( + owner, pid, path="src/util.py", status="exempt", reason="local", uses=[hid], + ) + assert out["classified"] == 1 + assert (await list_project_shapes(owner, pid, uses=hid))[1] == 2 + with pytest.raises(ValueError): + await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "Config", "status": "exempt", "reason": "x", "uses": [999999]}, + ]) + + @pytest.mark.integration async def test_list_filters_compose(seeded): owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] diff --git a/tests/test_shape_ledger.py b/tests/test_shape_ledger.py index fd9a8c1..cb07e65 100644 --- a/tests/test_shape_ledger.py +++ b/tests/test_shape_ledger.py @@ -391,6 +391,35 @@ def test_compact_row_carries_identity_standing_and_the_proposers_word_only(): assert noisy not in compact +def test_uses_edges_table_and_validation(): + """#2870: consumption is its own relation — a table that cascades with + both ends, and `uses` on a classification must be a list of ids.""" + from scribe.models import Base + from scribe.models.code_shape import USE_BASES, CodeShapeUse + from scribe.services.shape_ledger import validate_classifications + assert "code_shape_uses" in Base.metadata.tables + cols = CodeShapeUse.__table__.c + assert next(iter(cols.shape_id.foreign_keys)).ondelete == "CASCADE" + assert next(iter(cols.snippet_id.foreign_keys)).ondelete == "CASCADE" + assert set(USE_BASES) == {"reference", "hook", "agent", "audit", "import"} + ok = [{"path": "a.py", "symbol": "f", "status": "instance", "snippet_id": 9, "uses": [3, 4]}] + assert validate_classifications(ok) is None + bad = [{"path": "a.py", "symbol": "f", "status": "instance", "snippet_id": 9, "uses": "3"}] + assert "uses must be a list" in validate_classifications(bad) + + +def test_reference_canons_names_every_used_canon_not_just_the_best(): + from scribe.services.shape_ledger import Canon, _norm_text, reference_canons + a = Canon(1, "sym", "hash_token", (("src/x.py", "hash_token"),), "def hash_token(raw):", _norm_text("x"), 2, "python") + b = Canon(2, "sym", "rules_payload", (("src/y.py", "rules_payload"),), "def rules_payload(r):", _norm_text("y"), 2, "python") + ts = Canon(3, "sym", "fmtDate", (("f/d.ts", "fmtDate"),), "export function fmtDate(iso: string): string {", _norm_text("z"), 2, "typescript") + body = "def create_invitation(email):\n h = hash_token(raw)\n return rules_payload(h)\n" + assert reference_canons("sym", "src/scribe/services/auth.py", "create_invitation", body, [a, b, ts]) == [1, 2] + # the shape's own name and the other language family are never "uses" + assert reference_canons("sym", "src/x.py", "hash_token", body, [a]) == [] + assert reference_canons("sym", "f/v.vue", "show", "fmtDate(x); hash_token(y)", [a, ts]) == [3] + + def test_reason_codes_are_a_fixed_catalogue_and_validated(): """#2874: an optional index beside the prose reason; unknown codes are a structural error (the batch applies nothing)."""