feat(ledger): the CSS consumer map — code_shape_consumers edges (shape → file whose markup names the class, count), migration 0086, resolve_consumers (own-file row when the template defines the class, else every other definition), sync_repo_consumers rebuilt from the archive on every refresh, consumers_of; derived, so not backed up (milestone 302 step 2, #2935)

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

src/scribe/services/shape_ledger.py

@@ -30,7 +30,9 @@ 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, CodeShapeUse
+from scribe.models.code_shape import (
+    REASON_CODES, CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse,
+)
 from scribe.models.base import iso
 
 logger = logging.getLogger(__name__)
@@ -262,6 +264,96 @@ async def uses_of(shape_ids) -> dict[int, list[CodeShapeUse]]:
     return out
 
 
+# --- the CSS consumer map (milestone 302) ------------------------------------
+
+
+def resolve_consumers(
+    css_rows: Iterable[tuple[int, str, str]],
+    references: dict[str, dict[str, int]],
+) -> dict[tuple[int, str], int]:
+    """{(shape_id, consumer_path): count} — which CSS rows each file's markup
+    consumes. ``css_rows`` are (id, path, symbol) of the repo's live css rows;
+    ``references`` is scan_archive's path → class token → count.
+
+    Resolution (note 2917): a class named in file F resolves to F's OWN row
+    of that name when F defines it (a scoped rule is consumed by its own
+    template); otherwise to every other file's row of that name — a shared
+    sheet, or, when several files define it, all of them: the map says
+    "ambiguous" by fanning out rather than guessing one."""
+    by_symbol: dict[str, list[tuple[int, str]]] = {}
+    for sid, path, symbol in css_rows:
+        by_symbol.setdefault(symbol, []).append((sid, path))
+    out: dict[tuple[int, str], int] = {}
+    for consumer, tokens in references.items():
+        for token, count in tokens.items():
+            rows = by_symbol.get(token)
+            if not rows:
+                continue
+            own = [sid for sid, path in rows if path == consumer]
+            targets = own or [sid for sid, _path in rows]
+            for sid in targets:
+                out[(sid, consumer)] = out.get((sid, consumer), 0) + int(count)
+    return out
+
+
+async def sync_repo_consumers(
+    project_id: int, repo_key: str, references: dict[str, dict[str, int]]
+) -> int:
+    """Rebuild one repo's consumer edges from its archive's class references:
+    insert the new, refresh changed counts, delete what the tree no longer
+    says (a template rewritten, a class renamed, a file gone). Edges hang on
+    live rows only; a vanished row's edges go with this pass. Returns how
+    many edges stand afterwards."""
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                select(CodeShape.id, CodeShape.path, CodeShape.symbol, CodeShape.vanished_at).where(
+                    CodeShape.project_id == project_id,
+                    CodeShape.repo_key == repo_key,
+                    CodeShape.kind == "css",
+                )
+            )
+        ).all()
+        live = [(r[0], r[1], r[2]) for r in rows if r[3] is None]
+        all_ids = [r[0] for r in rows]
+        wanted = resolve_consumers(live, references)
+        existing = (
+            await session.execute(
+                select(CodeShapeConsumer).where(CodeShapeConsumer.shape_id.in_(all_ids))
+            )
+        ).scalars().all() if all_ids else []
+        have = {(e.shape_id, e.path): e for e in existing}
+        for key, edge in have.items():
+            if key not in wanted:
+                await session.delete(edge)
+            elif edge.count != wanted[key]:
+                edge.count = wanted[key]
+        for (sid, path), count in wanted.items():
+            if (sid, path) not in have:
+                session.add(CodeShapeConsumer(shape_id=sid, path=path, count=count, basis="template"))
+        await session.commit()
+    return len(wanted)
+
+
+async def consumers_of(shape_ids) -> dict[int, list[CodeShapeConsumer]]:
+    """{shape_id: [edges]} for a set of rows — the read side of the map,
+    ordered by path so a readout is stable."""
+    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(CodeShapeConsumer).where(CodeShapeConsumer.shape_id.in_(ids))
+                .order_by(CodeShapeConsumer.shape_id, CodeShapeConsumer.path)
+            )
+        ).scalars().all()
+    out: dict[int, list[CodeShapeConsumer]] = {}
+    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: