feat(backup): code_shape_uses travels (v9); consumer-map test expects uses (#2870)

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

src/scribe/services/backup.py

@@ -11,7 +11,7 @@ from scribe.models.note_supersession import NoteSupersession
 from scribe.models.note_version import NoteVersion
 from scribe.models.design_system import DesignSystem, DesignToken
 from scribe.models.note_usage import NoteUsageEvent
-from scribe.models.code_shape import CodeShape, CodeShapeEvent
+from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
 from scribe.models.project import Project
 from scribe.models.repo_binding import RepoBinding
 from scribe.models.rulebook import (
@@ -42,8 +42,11 @@ logger = logging.getLogger(__name__)
 # 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.
+# v9 (2026-08) added code_shape_uses — the ledger's consumption edges (#2870):
+# judgment-grade edges (agent/audit/import) are operator records; mechanical
+# ones (reference/hook) travel too, cheaply, and the next refresh refreshes them.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 8
+BACKUP_VERSION = 9
 
 # 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
@@ -62,7 +65,7 @@ _BACKED_UP = [
     "systems", "record_systems", "design_systems", "design_tokens",
     "note_usage_events", "repo_bindings", "note_supersessions",
     # v7 (2026-08): the shape ledger (#2787); v8: its history (#2793).
-    "code_shapes", "code_shape_events",
+    "code_shapes", "code_shape_events", "code_shape_uses",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -182,6 +185,10 @@ def _code_shape_event_rows(rows) -> list[dict]:
     return [r.to_dict() for r in rows]
 
 
+def _code_shape_use_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}
@@ -361,6 +368,9 @@ async def export_full_backup() -> dict:
         code_shape_events = (await session.execute(
             select(CodeShapeEvent).order_by(CodeShapeEvent.at, CodeShapeEvent.id)
         )).scalars().all()
+        code_shape_uses = (await session.execute(
+            select(CodeShapeUse).order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_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()
@@ -406,6 +416,7 @@ async def export_full_backup() -> dict:
         "note_supersessions": _note_supersession_rows(supersessions),
         "code_shapes": _code_shape_rows(code_shapes),
         "code_shape_events": _code_shape_event_rows(code_shape_events),
+        "code_shape_uses": _code_shape_use_rows(code_shape_uses),
     }
 
 
@@ -482,6 +493,11 @@ async def export_user_backup(user_id: int) -> dict:
             select(CodeShapeEvent).where(CodeShapeEvent.project_id.in_(project_ids))
             .order_by(CodeShapeEvent.at, CodeShapeEvent.id)
         )).scalars().all() if project_ids else []
+        code_shape_uses = (await session.execute(
+            select(CodeShapeUse).join(CodeShape, CodeShape.id == CodeShapeUse.shape_id)
+            .where(CodeShape.project_id.in_(project_ids))
+            .order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
+        )).scalars().all() if project_ids else []
         rulebooks = (await session.execute(
             select(Rulebook).where(Rulebook.owner_user_id == user_id)
         )).scalars().all()
@@ -553,6 +569,7 @@ async def export_user_backup(user_id: int) -> dict:
         "note_supersessions": _note_supersession_rows(supersessions),
         "code_shapes": _code_shape_rows(code_shapes),
         "code_shape_events": _code_shape_event_rows(code_shape_events),
+        "code_shape_uses": _code_shape_use_rows(code_shape_uses),
     }
 
 
@@ -657,6 +674,7 @@ async def _restore_v2(data: dict) -> dict:
         "systems": 0, "record_systems": 0, "design_systems": 0,
         "design_tokens": 0, "note_usage_events": 0, "repo_bindings": 0,
         "note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
+        "code_shape_uses": 0,
     }
 
     async with async_session() as session:
@@ -1105,6 +1123,20 @@ async def _restore_v2(data: dict) -> dict:
             ))
             stats["code_shape_events"] += 1
 
+        # v9: consumption edges (#2870) ride their shape AND their snippet —
+        # both ends must have survived, or the edge is no longer a fact.
+        for use in data.get("code_shape_uses", []):
+            new_shape_id = shape_id_map.get(use.get("shape_id") or 0)
+            new_sid = note_id_map.get(use.get("snippet_id") or 0)
+            if new_shape_id is None or new_sid is None:
+                continue
+            session.add(CodeShapeUse(
+                shape_id=new_shape_id, snippet_id=new_sid,
+                basis=use.get("basis", "import"), evidence=use.get("evidence"),
+                created_at=_dt(use.get("created_at")),
+            ))
+            stats["code_shape_uses"] += 1
+
         await session.commit()
 
     logger.info("Restored v2/v3 backup: %s", stats)

tests/test_integration_shape_classify.py

@@ -278,7 +278,7 @@ async def test_get_snippet_carries_the_structured_consumer_map(seeded):
     # The map is caller-scoped: an outsider asking the service directly gets
     # silence, not another project's file layout.
     consumers = await snippet_consumers(seeded["other"], sid)
-    assert consumers == {"instances": [], "variants": []}
+    assert consumers == {"instances": [], "variants": [], "uses": []}
 
 
 @pytest.mark.integration

tests/test_services_backup.py

@@ -18,7 +18,7 @@ def test_backup_version_is_v8():
     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
+    assert backup.BACKUP_VERSION == 9
 
 
 def test_not_included_lists_the_known_gaps():
@@ -115,7 +115,8 @@ 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", "code_shape_events"):
+                "note_supersessions", "code_shapes", "code_shape_events",
+                "code_shape_uses"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []