feat(ledger): classify_shapes + list_shapes MCP tools; get_snippet carries the consumer map (#2789, milestone 294 step 3)

The judgment write path. classify_shapes applies a batch of classifications
to a project's live ledger rows — all-or-nothing (#2709's lesson: the whole
batch is validated, write-ACL'd, and every snippet target proven readable
before any row is touched); rows match by exact (path, symbol), kind narrows,
and shapes no live row matches come back as 'unmatched' rather than errors.
variant/exempt REQUIRE the reason — the why is the record (note 2786) — and
'unclassified' deliberately withdraws a judgment back to the todo. The 'via'
channel is caller-restricted to agent|audit|import; hook and mechanical stay
server-internal so a caller can't launder judgment as machinery.

list_shapes is the todo query (status=unclassified) with composable filters:
path is exact-or-under like recorded locations, snippet_id reads a consumer
map, include_vanished reads history; paged with the true total.

get_snippet now attaches  and  — the structured consumer
map, filtered to projects the CALLER can read so a shared snippet never side-
channels another project's file layout; attached only when non-empty (#2483).

Integration tests pin the batch atomicity, ACL gates, filter composition, the
consumer map on the MCP pull, and the SET NULL companion: a judgment whose
snippet was purged rejoins the todo on the next sync.

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

src/scribe/services/shape_ledger.py

@@ -174,3 +174,237 @@ async def live_rows(project_id: int) -> list[CodeShape]:
                 )
             ).scalars().all()
         )
+
+
+# --- classification (#2789): the judgment write path --------------------------
+
+# Statuses an explicit classification may set. All five: setting a row back to
+# `unclassified` is how a judgment is deliberately withdrawn.
+_SETTABLE = ("canonical", "instance", "variant", "exempt", "unclassified")
+
+# Who may appear as the classifier on this path. `hook` and `mechanical` are
+# server-internal feeds (steps 5-6) — a caller claiming them would launder a
+# judgment as machinery.
+_CALLER_VIAS = ("agent", "audit", "import")
+
+
+def validate_classifications(items: list[dict]) -> str | None:
+    """The structural error a classification batch would earn, or None.
+
+    Pure and checked BEFORE anything is touched: a batch either applies or
+    errors whole — the StrictArgs lesson (#2709), a caller must never learn
+    later that half a batch silently happened.
+    """
+    if not items:
+        return "classifications is empty — nothing to apply"
+    for i, item in enumerate(items):
+        if not isinstance(item, dict):
+            return f"classifications[{i}] is not an object"
+        path = (item.get("path") or "").strip()
+        symbol = (item.get("symbol") or "").strip()
+        if not path or not symbol:
+            return f"classifications[{i}] needs both path and symbol"
+        status = item.get("status") or ""
+        if status not in _SETTABLE:
+            return (
+                f"classifications[{i}] has unknown status {status!r} "
+                f"(one of: {', '.join(_SETTABLE)})"
+            )
+        snippet_id = item.get("snippet_id") or 0
+        if status in _NEEDS_TARGET and not snippet_id:
+            return (
+                f"classifications[{i}]: status {status!r} needs snippet_id — "
+                "the snippet this shape is (or departs from)"
+            )
+        if status in ("variant", "exempt") and not (item.get("reason") or "").strip():
+            return (
+                f"classifications[{i}]: status {status!r} needs a reason — "
+                "the WHY is the record (note 2786)"
+            )
+    return None
+
+
+async def classify_shapes(
+    user_id: int,
+    project_id: int,
+    classifications: list[dict],
+    *,
+    via: str = "agent",
+) -> dict:
+    """Apply a batch of judgments to a project's live ledger rows.
+
+    All-or-nothing on errors: the whole batch is validated (structure, write
+    access, every snippet target readable by the caller) before any row is
+    touched. Rows are matched by exact (path, symbol) — plus kind when the
+    item carries one — and a target no live row matches is reported in
+    ``unmatched``, not an error: the tree may simply have moved since the
+    caller listed. Idempotent by construction.
+    """
+    from scribe.services import access
+    from scribe.services import snippets as snippets_svc
+
+    if via not in _CALLER_VIAS:
+        raise ValueError(f"via must be one of: {', '.join(_CALLER_VIAS)}")
+    error = validate_classifications(classifications)
+    if error:
+        raise ValueError(error)
+    if not await access.can_write_project(user_id, project_id):
+        raise ValueError(f"project {project_id} not found or no write access")
+
+    # Snippet targets resolve through the caller's own read access — a
+    # family-canon snippet in another project counts (note 2786), a snippet
+    # the caller cannot read does not exist for them.
+    target_ids = {
+        int(item["snippet_id"])
+        for item in classifications
+        if item.get("status") in _NEEDS_TARGET
+    }
+    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)")
+
+    now = datetime.now(timezone.utc)
+    classified = 0
+    unmatched: list[dict] = []
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                select(CodeShape).where(
+                    CodeShape.project_id == project_id,
+                    CodeShape.vanished_at.is_(None),
+                )
+            )
+        ).scalars().all()
+        by_key: dict[tuple[str, str], list[CodeShape]] = {}
+        for row in rows:
+            by_key.setdefault((row.path, row.symbol), []).append(row)
+        for item in classifications:
+            matches = by_key.get(
+                ((item.get("path") or "").strip(), (item.get("symbol") or "").strip())
+            ) or []
+            kind = (item.get("kind") or "").strip()
+            if kind:
+                matches = [r for r in matches if r.kind == kind]
+            if not matches:
+                unmatched.append({
+                    "path": item.get("path"), "symbol": item.get("symbol"),
+                })
+                continue
+            status = item["status"]
+            for row in matches:
+                row.status = status
+                if status == "unclassified":
+                    row.snippet_id = None
+                    row.reason = None
+                    row.classified_by = None
+                    row.classified_at = None
+                else:
+                    row.snippet_id = (
+                        int(item["snippet_id"]) if status in _NEEDS_TARGET else None
+                    )
+                    row.reason = (item.get("reason") or "").strip() or None
+                    row.classified_by = via
+                    row.classified_at = now
+                classified += 1
+        await session.commit()
+    return {"classified": classified, "unmatched": unmatched}
+
+
+async def list_project_shapes(
+    user_id: int,
+    project_id: int,
+    *,
+    status: str = "",
+    path: str = "",
+    snippet_id: int = 0,
+    include_vanished: bool = False,
+    limit: int = 100,
+    offset: int = 0,
+) -> 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.
+    """
+    from sqlalchemy import func, or_
+
+    from scribe.services import access
+
+    if not await access.can_read_project(user_id, project_id):
+        return [], 0
+    conds = [CodeShape.project_id == project_id]
+    if not include_vanished:
+        conds.append(CodeShape.vanished_at.is_(None))
+    if status:
+        conds.append(CodeShape.status == status)
+    if path:
+        clean = path.strip().strip("/")
+        conds.append(or_(
+            CodeShape.path == clean, CodeShape.path.like(clean + "/%")
+        ))
+    if snippet_id:
+        conds.append(CodeShape.snippet_id == snippet_id)
+    async with async_session() as session:
+        total = (
+            await session.execute(
+                select(func.count()).select_from(CodeShape).where(*conds)
+            )
+        ).scalar_one()
+        rows = (
+            await session.execute(
+                select(CodeShape).where(*conds)
+                .order_by(CodeShape.path, CodeShape.symbol, CodeShape.kind)
+                .limit(max(1, min(limit, 500))).offset(max(0, offset))
+            )
+        ).scalars().all()
+    return list(rows), int(total)
+
+
+def _consumer_dict(row: CodeShape) -> dict:
+    """The compact shape a snippet's consumer map carries — enough to open
+    the file, none of the ledger bookkeeping."""
+    out = {
+        "project_id": row.project_id,
+        "repo": row.repo_key,
+        "path": row.path,
+        "symbol": row.symbol,
+        "kind": row.kind,
+        "classified_by": row.classified_by,
+    }
+    if row.reason:
+        out["reason"] = row.reason
+    return out
+
+
+async def snippet_consumers(user_id: int, note_id: int) -> dict:
+    """The structured consumer map for one snippet (#2789): its `instances`
+    (rows judged to conform) and `variants` (named departures, each carrying
+    its why). Rows are filtered to projects the CALLER can read — a shared
+    snippet must not become a side channel into someone else's project
+    layout. Empty lists mean "attach nothing" (#2483)."""
+    from scribe.services import access
+
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                select(CodeShape).where(
+                    CodeShape.snippet_id == note_id,
+                    CodeShape.status.in_(("instance", "variant")),
+                    CodeShape.vanished_at.is_(None),
+                )
+            )
+        ).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
+            )
+        if not readable[row.project_id]:
+            continue
+        out["instances" if row.status == "instance" else "variants"].append(
+            _consumer_dict(row)
+        )
+    return out