From 942edd1eb57db41a89eba9750a4602a9fd73388e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 19 Aug 2026 19:55:47 -0400 Subject: [PATCH] feat(ledger): classify_shapes + list_shapes MCP tools; get_snippet carries the consumer map (#2789, milestone 294 step 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- src/scribe/mcp/tools/__init__.py | 5 +- src/scribe/mcp/tools/shapes.py | 95 +++++++++ src/scribe/mcp/tools/snippets.py | 16 ++ src/scribe/services/shape_ledger.py | 234 +++++++++++++++++++++++ tests/test_integration_shape_classify.py | 221 +++++++++++++++++++++ tests/test_shape_ledger.py | 32 ++++ 6 files changed, 601 insertions(+), 2 deletions(-) create mode 100644 src/scribe/mcp/tools/shapes.py create mode 100644 tests/test_integration_shape_classify.py diff --git a/src/scribe/mcp/tools/__init__.py b/src/scribe/mcp/tools/__init__.py index f53e73f..e8b0b1f 100644 --- a/src/scribe/mcp/tools/__init__.py +++ b/src/scribe/mcp/tools/__init__.py @@ -5,8 +5,8 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called from `mcp.server.build_mcp_server`. """ from scribe.mcp.tools import ( - design_systems, milestones, notes, processes, projects, recent, repos, rulebooks, search, snippets, - systems, tags, tasks, trash, + design_systems, milestones, notes, processes, projects, recent, repos, rulebooks, search, shapes, + snippets, systems, tags, tasks, trash, ) @@ -24,5 +24,6 @@ def register_all(mcp) -> None: repos.register(mcp) processes.register(mcp) snippets.register(mcp) + shapes.register(mcp) rulebooks.register(mcp) trash.register(mcp) diff --git a/src/scribe/mcp/tools/shapes.py b/src/scribe/mcp/tools/shapes.py new file mode 100644 index 0000000..6643ed1 --- /dev/null +++ b/src/scribe/mcp/tools/shapes.py @@ -0,0 +1,95 @@ +"""Shape-ledger MCP tools — the classification write/read surface (#2789). + +The accounting model (note 2786): the snippet library records CANON (small); +the ledger accounts for EVERY extracted shape (total). These tools are how +agents move shapes out of `unclassified` — the todo state — and how they read +what still needs judgment. The ledger rows themselves are fed by the coverage +refresh; these tools only ever judge what the sync has seen. +""" +from __future__ import annotations + +from scribe.mcp._context import current_user_id +from scribe.services import shape_ledger as shape_ledger_svc + + +async def classify_shapes( + project_id: int, classifications: list[dict], via: str = "agent" +) -> dict: + """Record judgments for a project's code shapes — in batch, as rows. + + EVERY shape in a bound repo should end up classified (note 2786): + - `instance` of snippet N — it conforms to recorded canon (family-level + canon in another project counts; that fully accounts for the shape). + - `variant` of snippet N — a deliberate, named departure. `reason` + (the why) is REQUIRED; it is the record. + - `exempt` — judged genuinely one-off. `reason` REQUIRED. + - `canonical` of snippet N — this row IS the snippet's reference + (rarely set by hand; the coverage sync stamps these mechanically). + - `unclassified` — withdraw a judgment; the shape rejoins the todo. + + Consumer maps belong HERE, not in prose: when an audit enumerates call + sites of a canonical helper, each call site's defining shape is an + `instance` row — a sentence in a verification detail cannot be sorted, + queried, or diffed. + + Args: + project_id: The project whose ledger is being judged. + classifications: Objects of {path, symbol, status, kind?, snippet_id?, + reason?}. path+symbol name the shape exactly as list_shapes shows + it; kind ("sym"/"css") narrows when one file defines both. + snippet_id is required for canonical/instance/variant; reason is + required for variant/exempt. + via: Who is judging — "agent" (default), "audit" (a sweep), or + "import" (carrying maps recorded elsewhere). + + All-or-nothing: a structural error, a missing snippet target, or no write + access applies NOTHING. Returns {"classified": N, "unmatched": [...]} — + unmatched names shapes no live ledger row matches (the tree may have + moved since you listed; re-run the project's coverage refresh to re-sync). + """ + uid = current_user_id() + return await shape_ledger_svc.classify_shapes( + uid, project_id, classifications, via=via + ) + + +async def list_shapes( + project_id: int, + status: str = "", + path: str = "", + snippet_id: int = 0, + include_vanished: bool = False, + limit: int = 100, + offset: int = 0, +) -> dict: + """Read a project's shape ledger — `status="unclassified"` IS the todo. + + Every extracted definition in the project's bound repos has a row here + (fed by the coverage refresh). Filters compose: + + Args: + status: canonical | instance | variant | exempt | 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. + include_vanished: include shapes no longer in the tree (history). + limit/offset: page through big ledgers (limit caps at 500). + + Returns {"shapes": [...], "total": N} — total counts every match, not + just this page. Classify what you can judge with classify_shapes; a + repeating shape with NO recorded canon is a derive-one-first moment + (consolidate onto a reference, create_snippet it, then classify the + rest against it), never N loose classifications. + """ + uid = current_user_id() + rows, total = await shape_ledger_svc.list_project_shapes( + uid, project_id, + status=status, path=path, snippet_id=snippet_id, + include_vanished=include_vanished, limit=limit, offset=offset, + ) + return {"shapes": [r.to_dict() for r in rows], "total": total} + + +def register(mcp) -> None: + for fn in (classify_shapes, list_shapes): + mcp.tool(name=fn.__name__)(fn) diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py index cdb3b37..1cc2871 100644 --- a/src/scribe/mcp/tools/snippets.py +++ b/src/scribe/mcp/tools/snippets.py @@ -207,6 +207,12 @@ async def get_snippet(snippet_id: int) -> dict: the source moved on — trust the location over the cached body and consider verify_snippet after you look. + When the shape ledger has judgments against this snippet, the response + carries `instances` (shapes classified as conforming to it — the + structured consumer map) and/or `variants` (named departures, each with + its why). Consult them before changing the snippet's contract: they are + the call sites your change lands on (classify_shapes maintains them). + If the record belongs to someone else it carries `shared: true` with the `owner` and your `permission`. Read that as ONE PERSON'S SUGGESTION, not as established practice here: judge it on its merits, say whose it is when you @@ -229,6 +235,16 @@ async def get_snippet(snippet_id: int) -> dict: await systems_tools.attach_systems( uid, note.user_id, data, note.id, note.project_id ) + # The structured consumer map (#2789): ledger rows judged against this + # snippet. Attached only when non-empty (#2483) — and never for projects + # the caller can't read. + from scribe.services import shape_ledger as shape_ledger_svc + + consumers = await shape_ledger_svc.snippet_consumers(uid, int(note.id)) + if consumers["instances"]: + data["instances"] = consumers["instances"] + if consumers["variants"]: + data["variants"] = consumers["variants"] return data diff --git a/src/scribe/services/shape_ledger.py b/src/scribe/services/shape_ledger.py index d464458..d1fcbea 100644 --- a/src/scribe/services/shape_ledger.py +++ b/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 diff --git a/tests/test_integration_shape_classify.py b/tests/test_integration_shape_classify.py new file mode 100644 index 0000000..b8d84a8 --- /dev/null +++ b/tests/test_integration_shape_classify.py @@ -0,0 +1,221 @@ +"""Real-Postgres integration tests for shape classification (#2789). + +What mocks can't prove: the all-or-nothing batch against real rows, the +write-ACL gate, the todo query's filters, and the consumer map riding +get_snippet. Ledger rows are seeded through the same sync the coverage walk +uses — no forge needed, the sync takes extracted shapes directly. +""" +import pytest +import pytest_asyncio +from sqlalchemy import select + +from scribe.models import async_session, engine +from scribe.models.code_shape import CodeShape +from scribe.models.project import Project +from scribe.models.user import User +from scribe.services.shape_ledger import ( + classify_shapes, + list_project_shapes, + snippet_consumers, + sync_repo_shapes, +) + +pytestmark = pytest.mark.integration + +REPO = "git.example.com/alice/widget" +SHAPES = [ + ("src/app.py", "sym", "make_app"), + ("src/app.py", "sym", "Config"), + ("src/util.py", "sym", "helper"), + ("web/button.css", "css", "btn"), +] + + +@pytest_asyncio.fixture(autouse=True) +async def _dispose_engine(): + """Dispose the app's module-level engine after each test. + + The engine pools asyncpg connections per event loop, but pytest-asyncio runs + each test on a fresh loop — so without this, test 2 gets handed test 1's + connection bound to a now-dead loop ("Future attached to a different loop"). + Disposing in the test's own loop teardown clears the pool cleanly. + """ + yield + await engine.dispose() + + +async def _user(session, username: str) -> User: + existing = ( + await session.execute(select(User).where(User.username == username)) + ).scalar_one_or_none() + if existing is not None: + return existing + user = User(username=username) + session.add(user) + await session.flush() + return user + + +@pytest_asyncio.fixture +async def seeded(): + """Owner + outsider, a project with a synced 4-shape ledger, one snippet.""" + from scribe.services import snippets as snippets_svc + + async with async_session() as s: + owner = await _user(s, "classify_owner") + other = await _user(s, "classify_other") + project = Project(user_id=owner.id, title="Classify target") + s.add(project) + await s.flush() + ids = {"owner": owner.id, "other": other.id, "pid": project.id} + await s.commit() + + await sync_repo_shapes(ids["pid"], REPO, SHAPES, seen_marker="main") + snippet = await snippets_svc.create_snippet( + ids["owner"], name="cls_make_app", code="def make_app():\n pass\n", + language="python", repo="Widget", path="src/factory.py", + symbol="factory", project_id=ids["pid"], + ) + ids["snippet"] = int(snippet.id) + return ids + + +@pytest.mark.integration +async def test_classify_applies_judgments_and_reports_unmatched(seeded): + owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] + out = await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "instance", + "snippet_id": sid}, + {"path": "src/util.py", "symbol": "helper", "status": "exempt", + "reason": "test scaffolding, deliberately local"}, + {"path": "web/button.css", "symbol": "btn", "status": "variant", + "snippet_id": sid, "reason": "darker focus ring for the toolbar"}, + {"path": "gone.py", "symbol": "nothing", "status": "exempt", + "reason": "x"}, + ], via="audit") + assert out["classified"] == 3 + assert out["unmatched"] == [{"path": "gone.py", "symbol": "nothing"}] + + rows, total = await list_project_shapes(owner, pid) + by_symbol = {r.symbol: r for r in rows} + assert total == 4 + assert by_symbol["make_app"].status == "instance" + assert by_symbol["make_app"].snippet_id == sid + assert by_symbol["make_app"].classified_by == "audit" + assert by_symbol["helper"].status == "exempt" + assert by_symbol["helper"].reason == "test scaffolding, deliberately local" + assert by_symbol["btn"].status == "variant" + assert by_symbol["Config"].status == "unclassified" + + # Withdrawing a judgment returns the shape to the todo, fields cleared. + await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "unclassified"}, + ]) + rows, _ = await list_project_shapes(owner, pid, status="unclassified") + assert {r.symbol for r in rows} == {"Config", "make_app"} + make_app = next(r for r in rows if r.symbol == "make_app") + assert make_app.snippet_id is None and make_app.classified_by is None + + +@pytest.mark.integration +async def test_a_bad_batch_applies_nothing(seeded): + """All-or-nothing (#2709's lesson): a caller must never learn later that + half a batch silently happened.""" + owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] + with pytest.raises(ValueError) as err: + await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "instance", + "snippet_id": sid}, + {"path": "src/util.py", "symbol": "helper", "status": "variant", + "snippet_id": sid}, # variant with no reason: structural error + ]) + assert "needs a reason" in str(err.value) + rows, _ = await list_project_shapes(owner, pid, status="unclassified") + assert len(rows) == 4 # including make_app — the valid half did NOT apply + + # A snippet target the caller can't read is the same: nothing applies. + with pytest.raises(ValueError): + await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "instance", + "snippet_id": 999999999}, + ]) + + +@pytest.mark.integration +async def test_classification_is_write_gated_and_listing_read_gated(seeded): + other, pid, sid = seeded["other"], seeded["pid"], seeded["snippet"] + with pytest.raises(ValueError): + await classify_shapes(other, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "exempt", + "reason": "not their call to make"}, + ]) + assert await list_project_shapes(other, pid) == ([], 0) + + +@pytest.mark.integration +async def test_list_filters_compose(seeded): + owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] + await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "instance", + "snippet_id": sid}, + ]) + rows, total = await list_project_shapes(owner, pid, path="src") + assert total == 3 and all(r.path.startswith("src/") for r in rows) + # Directory semantics, not string prefix: "sr" matches nothing. + assert (await list_project_shapes(owner, pid, path="sr"))[1] == 0 + rows, total = await list_project_shapes(owner, pid, snippet_id=sid) + assert total == 1 and rows[0].symbol == "make_app" + rows, total = await list_project_shapes( + owner, pid, status="unclassified", limit=2 + ) + assert total == 3 and len(rows) == 2 # paged, with the true total + + +@pytest.mark.integration +async def test_get_snippet_carries_the_structured_consumer_map(seeded): + from scribe.mcp._context import _user_id_ctx + from scribe.mcp.tools.snippets import get_snippet + + owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] + await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "instance", + "snippet_id": sid}, + {"path": "web/button.css", "symbol": "btn", "status": "variant", + "snippet_id": sid, "reason": "darker focus ring"}, + ]) + token = _user_id_ctx.set(owner) + try: + data = await get_snippet(snippet_id=sid) + finally: + _user_id_ctx.reset(token) + assert [i["path"] for i in data["instances"]] == ["src/app.py"] + assert data["variants"][0]["reason"] == "darker focus ring" + + # 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": []} + + +@pytest.mark.integration +async def test_sync_refiles_rows_whose_snippet_was_purged(seeded): + """The SET NULL companion (#2787): a judgment whose target is hard-deleted + rejoins the todo on the next sync instead of dangling target-less.""" + owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] + await classify_shapes(owner, pid, [ + {"path": "src/app.py", "symbol": "make_app", "status": "instance", + "snippet_id": sid}, + ]) + from scribe.models.note import Note + + async with async_session() as s: + note = await s.get(Note, sid) + await s.delete(note) # hard delete, as purge_trash would + await s.commit() + await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="main") + async with async_session() as s: + row = (await s.execute(select(CodeShape).where( + CodeShape.project_id == pid, CodeShape.symbol == "make_app", + ))).scalar_one() + assert row.status == "unclassified" + assert row.snippet_id is None diff --git a/tests/test_shape_ledger.py b/tests/test_shape_ledger.py index 07ecbd4..56b0134 100644 --- a/tests/test_shape_ledger.py +++ b/tests/test_shape_ledger.py @@ -50,3 +50,35 @@ def test_status_queries_have_an_index(): names = {ix.name for ix in CodeShape.__table__.indexes} assert "ix_code_shapes_project_status" in names assert "ix_code_shapes_snippet" in names + + +# --- step 3: the classification batch validator (pure, checked before ACL) --- + + +def test_batch_validation_names_the_failing_item(): + from scribe.services.shape_ledger import validate_classifications as v + + ok = {"path": "src/a.py", "symbol": "f", "status": "exempt", "reason": "one-off"} + assert v([ok]) is None + assert "empty" in v([]) + assert "classifications[1]" in v([ok, {"symbol": "f", "status": "exempt"}]) + assert "unknown status" in v([{**ok, "status": "covered"}]) + # A judgment that references canon must name the canon... + assert "needs snippet_id" in v( + [{"path": "a", "symbol": "f", "status": "instance"}] + ) + # ...and a departure/exemption must carry its why — the why IS the record. + assert "needs a reason" in v( + [{"path": "a", "symbol": "f", "status": "variant", "snippet_id": 3}] + ) + assert "needs a reason" in v([{"path": "a", "symbol": "f", "status": "exempt"}]) + # Withdrawing a judgment needs neither target nor reason. + assert v([{"path": "a", "symbol": "f", "status": "unclassified"}]) is None + + +def test_classify_and_list_are_mounted_as_mcp_tools(): + from scribe.mcp.server import build_mcp_server + + mcp = build_mcp_server() + for name in ("classify_shapes", "list_shapes"): + assert mcp._tool_manager.get_tool(name) is not None