feat(ledger): audit surfaces — list_shapes(compact=True) and classify_shapes_by_rule, the sweep form of a judgment (#2868, milestone 294)

The 2026-08 audit judged 3,427 rows in 14 hand-driven batches through a raw
MCP client because a list_shapes page overflowed the tool budget and every
row had to be sent back one by one. Now:

- list_shapes(compact=True): path · symbol · kind · status · signature, plus
  snippet_id / by / proposal / diverges_from / recheck only when set. A full
  500-row page fits. CodeShape.to_compact() is the row shape.
- classify_shapes_by_rule(project_id, path, status, pattern=, kind=,
  snippet_id=, reason=, via=, include_judged=): ONE judgment over every
  unclassified live row under a directory whose symbol matches a glob;
  judged rows are untouched unless include_judged; canonical is refused;
  same gates as the row form; one transaction; returns count + sample.
  shape_ledger.classify_shapes_where / rule_matches (pure) carry it.

Tests: compact row pinned, rule_matches directory/glob/kind semantics, the
tool mount, and an integration sweep (unclassified-only, include_judged,
gates).

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

src/scribe/mcp/tools/shapes.py

@@ -64,6 +64,7 @@ async def list_shapes(
     offset: int = 0,
     proposal: str = "",
     flag: str = "",
+    compact: bool = False,
 ) -> dict:
     """Read a project's shape ledger — `status="unclassified"` IS the todo.
 
@@ -77,6 +78,11 @@ async def list_shapes(
         snippet_id: rows classified against this snippet — a consumer map.
         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
+            snippet_id / by / proposal / diverges_from / recheck only when
+            set — no commits, shas or timestamps. THE form for an audit:
+            a full 500-row page fits the tool budget. The default rows carry
+            everything (shape_history-grade bookkeeping).
         proposal: the proposer's queue (#2792) — "any", "canon" (rows the
             machine thinks are an instance of a snippet: `proposal` carries
             snippet_id, basis, score), "derive" (rows that repeat with NO
@@ -118,7 +124,63 @@ async def list_shapes(
         include_vanished=include_vanished, limit=limit, offset=offset,
         proposal=proposal, flag=flag,
     )
-    return {"shapes": [r.to_dict() for r in rows], "total": total}
+    return {
+        "shapes": [r.to_compact() if compact else r.to_dict() for r in rows],
+        "total": total,
+    }
+
+
+async def classify_shapes_by_rule(
+    project_id: int,
+    path: str,
+    status: str,
+    pattern: str = "",
+    kind: str = "",
+    snippet_id: int = 0,
+    reason: str = "",
+    via: str = "agent",
+    include_judged: bool = False,
+) -> dict:
+    """The sweep form of classify_shapes: ONE judgment applied to every
+    unclassified shape under a directory whose symbol matches a glob.
+
+    For the long tail an audit judges by family, not by row — "every scoped
+    rule under frontend/src/views is exempt: styles one element of its view",
+    "every `*_scheduler.py` symbol is an instance of ScheduledJob" — where
+    listing 900 rows and sending them back is the whole cost. The row form
+    stays the precise tool; reach for it when each row gets its own reason.
+
+    Args:
+        project_id: The project whose ledger is being judged.
+        path: A file, or a directory and everything beneath it. Required —
+            a sweep names what it judges.
+        status: instance | variant | exempt | unclassified (canonical is the
+            sync's stamp, not a sweep's).
+        pattern: Shell glob on the symbol (`*_rows`, `_*`, `modal-*`, `*`);
+            "" = every symbol under path.
+        kind: "sym" or "css" to narrow; "" = both.
+        snippet_id: Required for instance/variant — the canon judged against.
+        reason: Required for variant/exempt — the why, recorded on every row.
+        via: "agent" (default) | "audit" | "import".
+        include_judged: By default only `unclassified` rows are touched — a
+            sweep never silently overwrites a judgment. True re-judges every
+            matching live row (use to re-confirm after a recheck, or to
+            revise a family you judged earlier).
+
+    One transaction: applies whole or not at all. Returns
+    {"classified": N, "sample": ["path::symbol", ...]} (first 12, sorted)
+    so you can see what the rule reached; N = 0 means the rule matched
+    nothing live and unclassified — widen the pattern or refresh coverage.
+    """
+    uid = current_user_id()
+    try:
+        return await shape_ledger_svc.classify_shapes_where(
+            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,
+        )
+    except ValueError as exc:
+        return {"error": str(exc)}
 
 
 async def shape_history(
@@ -217,7 +279,7 @@ async def refresh_pattern_coverage(project_id: int) -> dict:
 
 def register(mcp) -> None:
     for fn in (
-        classify_shapes, list_shapes, refresh_pattern_coverage,
-        confirm_shape_proposals, shape_history,
+        classify_shapes, classify_shapes_by_rule, list_shapes,
+        refresh_pattern_coverage, confirm_shape_proposals, shape_history,
     ):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/models/code_shape.py

@@ -167,6 +167,31 @@ class CodeShape(Base, TimestampMixin):
             "updated_at": iso(self.updated_at),
         }
 
+    def to_compact(self) -> dict:
+        """The row as an audit reads it (#2868): identity, standing, the
+        definition line and the proposer's word — none of the bookkeeping
+        (commits, shas, timestamps). A 500-row page of these fits the tool
+        budget; a page of to_dict() does not."""
+        out = {
+            "path": self.path,
+            "symbol": self.symbol,
+            "kind": self.kind,
+            "status": self.status,
+            "signature": self.signature,
+        }
+        if self.snippet_id is not None:
+            out["snippet_id"] = self.snippet_id
+        if self.classified_by:
+            out["by"] = self.classified_by
+        proposal = self.proposal
+        if proposal:
+            out["proposal"] = proposal
+        if self.diverges_from is not None:
+            out["diverges_from"] = self.diverges_from
+        if self.recheck_at is not None:
+            out["recheck"] = True
+        return out
+
 
 # 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:

src/scribe/services/shape_ledger.py

@@ -370,6 +370,82 @@ async def classify_shapes(
     return {"classified": classified, "unmatched": unmatched}
 
 
+def rule_matches(row: CodeShape, *, path: str, pattern: str, kind: str) -> bool:
+    """Does a ledger row fall under a rule-form classification (#2868)?
+    ``path`` is a file or a directory (everything beneath it), ``pattern``
+    a shell glob on the symbol (``""`` = every symbol), ``kind`` narrows to
+    sym/css. Pure, so the sweep's reach can be tested without a database."""
+    import fnmatch
+
+    clean = (path or "").strip().strip("/")
+    if clean and not (row.path == clean or row.path.startswith(clean + "/")):
+        return False
+    if kind and row.kind != kind:
+        return False
+    if pattern and not fnmatch.fnmatchcase(_norm_symbol(row.symbol), pattern):
+        return False
+    return True
+
+
+async def classify_shapes_where(
+    user_id: int,
+    project_id: int,
+    *,
+    path: str,
+    status: str,
+    pattern: str = "",
+    kind: str = "",
+    snippet_id: int | None = None,
+    reason: str | None = None,
+    via: str = "agent",
+    include_judged: bool = False,
+) -> dict:
+    """The sweep form of classify_shapes (#2868): one judgment applied to
+    every live row under ``path`` whose symbol matches ``pattern`` (and
+    ``kind``). By default only `unclassified` rows are touched — a sweep
+    must never silently overwrite a judgment; ``include_judged`` opts in.
+    Same gates as the row form (status vocabulary, snippet target, reason
+    for variant/exempt, write access); one transaction, so it applies whole
+    or not at all. Returns the count and a sample of what it judged."""
+    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)}")
+    if not (path or "").strip():
+        raise ValueError("path is required — a sweep names the directory it judges")
+    if status == "canonical":
+        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 ""}
+    error = validate_classifications([probe])
+    if error:
+        raise ValueError(error.replace("classifications[0]", "rule"))
+    if not await access.can_write_project(user_id, project_id):
+        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)")
+
+    now = datetime.now(timezone.utc)
+    judged: list[str] = []
+    async with async_session() as session:
+        conds = [CodeShape.project_id == project_id, CodeShape.vanished_at.is_(None)]
+        if not include_judged:
+            conds.append(CodeShape.status == "unclassified")
+        rows = (await session.execute(select(CodeShape).where(*conds))).scalars().all()
+        for row in rows:
+            if not rule_matches(row, path=path, pattern=pattern, kind=kind):
+                continue
+            await _judge(
+                session, row, status=status,
+                snippet_id=int(snippet_id) if status in _NEEDS_TARGET else None,
+                by=via, reason=reason, at=now,
+            )
+            judged.append(f"{row.path}::{row.symbol}")
+        await session.commit()
+    return {"classified": len(judged), "sample": sorted(judged)[:12]}
+
+
 async def list_project_shapes(
     user_id: int,
     project_id: int,