feat(shapes): the agent judges what it wrote, at the end of the turn (milestone 439 steps 1-3)

Recording used to be decided by machinery — the only "record it" prompt
fired when a same-named copy already existed (#2664), so a first instance of
a reusable piece was never asked about, and judgment arrived only through
audits. Now the question is asked where the knowledge is: the end of the
turn that wrote the code, of the agent that wrote it.

- Write hooks keep `<sid>.written.ids` (path, kind, name) for every
  definition a write names; a new file adds a `file` line for its stem — a
  candidate in any language without a framework rule (scribe_written_append).
- Stop hook scribe_shape_check.sh sends the ledger to GET
  /api/plugin/shape-check and blocks once, in the server's words, when
  anything is unjudged. Same discipline as the report check: never twice,
  never without a recorded check, another hook's loop left alone; the ledger
  is kept when the instance cannot be reached.
- shape_ledger.unjudged_shapes: no row, unclassified, scoped and hook stamps
  are unjudged; an agent/audit/import verdict is not. A snippet recorded at
  the shape answers for it until the refresh stamps it canonical.
- services/shape_check owns the reason text and records every outcome in
  app_logs (passed / blocked / judged_after_block / left_after_block).
- classify_shapes(repo=…) judges a shape the ledger has not synced yet via a
  provisional row under a bound repo; the sync confirms it, or vanishes and
  revives it with the verdict intact. An unbound repo is refused.

Plugin version minted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_integration_shape_classify.py

@@ -1138,3 +1138,81 @@ async def test_history_records_what_was_used_when_and_drift_asks_for_a_recheck(s
     # reads nothing.
     assert len((await shape_history(owner, pid, "src"))["events"]) == 6
     assert await shape_history(other, pid, "src/app.py") == {}
+
+
+# --- milestone 439: the writer judges what it wrote, before the sync has seen it
+
+@pytest.mark.integration
+async def test_a_shape_written_this_turn_is_judged_before_the_sync_sees_it(seeded):
+    """The end-of-turn verdict lands on a shape no sync has read yet: with a
+    bound `repo` it becomes a provisional row and is judged, the sync that
+    then reads the shape keeps the judgment, and a sync that does NOT have it
+    yet (the code has not reached the bound ref) keeps it through the vanish
+    and the reappearance. Without `repo` the old contract holds."""
+    from scribe.services import repo_bindings as repo_bindings_svc
+
+    owner, pid = seeded["owner"], seeded["pid"]
+    await repo_bindings_svc.set_binding(owner, REPO, pid)
+    verdict = [{"path": "web/src/lib/StatusChip.svelte", "symbol": "tone",
+                "kind": "sym", "status": "exempt",
+                "reason": "the chip's own colour switch", "reason_code": "pure-helper"}]
+
+    out = await classify_shapes(owner, pid, verdict)
+    assert out["unmatched"] == [{"path": "web/src/lib/StatusChip.svelte", "symbol": "tone"}]
+
+    with pytest.raises(ValueError, match="not bound"):
+        await classify_shapes(owner, pid, verdict, repo="git.example.com/alice/elsewhere")
+
+    out = await classify_shapes(owner, pid, verdict, repo=REPO)
+    assert out == {"classified": 1, "unmatched": [], "provisional": 1}
+
+    # A sync that has not got the code yet: the row vanishes, judgment intact…
+    await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="c1")
+    # …and the sync that has it brings the row back with the verdict.
+    await sync_repo_shapes(
+        pid, REPO, SHAPES + [("web/src/lib/StatusChip.svelte", "sym", "tone")], seen_marker="c2",
+    )
+    rows, _ = await list_project_shapes(owner, pid, path="web/src/lib/StatusChip.svelte")
+    assert len(rows) == 1
+    row = rows[0]
+    assert (row.status, row.classified_by, row.reason_code) == ("exempt", "agent", "pure-helper")
+    assert row.vanished_at is None and row.last_seen_commit == "c2"
+
+
+@pytest.mark.integration
+async def test_unjudged_is_what_nobody_read(seeded):
+    """The end-of-turn question: a shape with no row, an unclassified one, a
+    sync-scoped one and a hook stamp are all asked about; an agent's or an
+    audit's verdict is not. Evidence travels; a verdict is never inferred."""
+    from sqlalchemy import update
+
+    from scribe.services.shape_ledger import _RESEMBLE_REASON, unjudged_shapes
+
+    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": "src/util.py", "symbol": "helper", "status": "exempt", "reason": "x"},
+    ])
+    async with async_session() as session:
+        await session.execute(
+            update(CodeShape)
+            .where(CodeShape.project_id == pid, CodeShape.symbol == "helper")
+            .values(classified_by="hook", status="instance", snippet_id=sid,
+                    reason=_RESEMBLE_REASON.format(sid=sid, score=0.7))
+        )
+        await session.commit()
+
+    written = [
+        ("src/app.py", "sym", "make_app"),      # judged by an agent → not asked
+        ("src/app.py", "sym", "Config"),        # unclassified → asked
+        ("src/util.py", "sym", "helper"),       # a hook stamp → asked, with its evidence
+        ("src/new.py", "sym", "fresh"),         # no row → asked as new
+        ("src/new.py", "sym", "fresh"),         # duplicates collapse
+    ]
+    got = await unjudged_shapes(pid, written, repo_key=REPO)
+    assert [(g["symbol"], g["status"]) for g in got] == [
+        ("Config", "unclassified"), ("helper", "instance"), ("fresh", "new"),
+    ]
+    assert got[1]["evidence"]["stamped"] == {"snippet_id": sid, "score": 0.7}
+    assert got[2]["evidence"] == {}
+    assert await unjudged_shapes(pid, []) == []

tests/test_services_shape_check.py

@@ -0,0 +1,81 @@
+"""The server half of the end-of-turn shape check (milestone 439): what the
+agent is asked, what the hook's ledger may say, and the outcome record."""
+import json
+from unittest.mock import patch
+
+import pytest
+
+from tests.helpers import make_mock_session
+
+
+def test_the_ledger_lines_parse_and_garbage_is_dropped():
+    from scribe.services.shape_check import parse_written
+
+    raw = ("a.py\tsym\thelper\n"
+           "a.py\tsym\thelper\n"                 # duplicate
+           "web/A.svelte\tfile\tA\n"
+           "x.py\tbogus\tthing\n"                 # unknown kind
+           "only-two\tsym\n"                      # malformed
+           "\tsym\tnameless-path\n")
+    assert parse_written(raw) == [("a.py", "sym", "helper"), ("web/A.svelte", "file", "A")]
+    assert len(parse_written("".join(f"p\tsym\tn{i}\n" for i in range(500)), cap=7)) == 7
+
+
+def test_a_first_stop_blocks_or_passes_and_the_next_never_blocks():
+    from scribe.services.shape_check import outcome_for
+
+    one = [{"path": "a", "kind": "sym", "symbol": "b"}]
+    assert outcome_for("check", one) == "blocked"
+    assert outcome_for("check", []) == "passed"
+    assert outcome_for("after", one) == "left_after_block"
+    assert outcome_for("after", []) == "judged_after_block"
+
+
+def test_the_reason_names_each_shape_its_evidence_and_the_one_call():
+    from scribe.services.shape_check import block_reason
+
+    unjudged = [
+        {"path": "web/A.svelte", "kind": "file", "symbol": "A", "status": "new", "evidence": {}},
+        {"path": "src/x.py", "kind": "sym", "symbol": "load", "status": "unclassified",
+         "evidence": {"looks_like": {"snippet_id": 12, "basis": "semantic", "score": 0.84},
+                      "diverges_from": 9}},
+    ]
+    reason = block_reason(unjudged, project_id=3, repo="git.example.com/a/w")
+    assert "2 definitions" in reason
+    assert "web/A.svelte · A (file) — new" in reason
+    assert "looks like #12 0.84" in reason and "#9 is canon in this directory" in reason
+    assert 'classify_shapes(project_id=3, repo="git.example.com/a/w"' in reason
+    for verdict in ("create_snippet", "`instance`", "`variant`", "`exempt`"):
+        assert verdict in reason, verdict
+
+
+def test_a_long_turn_is_listed_with_a_tail_not_dropped():
+    from scribe.services.shape_check import _LISTED, block_reason
+
+    many = [{"path": "p.py", "kind": "sym", "symbol": f"f{i}", "status": "new", "evidence": {}}
+            for i in range(_LISTED + 3)]
+    reason = block_reason(many, project_id=1, repo="")
+    assert "… and 3 more" in reason
+    assert "repo=" not in reason
+
+
+async def test_the_outcome_is_recorded_as_a_plugin_event():
+    from scribe.services.shape_check import record_shape_check
+
+    session = make_mock_session()
+    with patch("scribe.services.shape_check.async_session", return_value=session):
+        await record_shape_check(7, "blocked", written=5, unjudged=2, project_id=2)
+    row = session.add.call_args.args[0]
+    assert (row.category, row.action, row.user_id) == ("plugin", "shape_check", 7)
+    assert json.loads(row.details) == {"outcome": "blocked", "written": 5,
+                                       "unjudged": 2, "project_id": 2}
+
+
+async def test_an_unknown_outcome_is_refused_before_anything_is_written():
+    from scribe.services.shape_check import record_shape_check
+
+    session = make_mock_session()
+    with patch("scribe.services.shape_check.async_session", return_value=session), \
+         pytest.raises(ValueError):
+        await record_shape_check(7, "skipped", written=1, unjudged=1)
+    session.add.assert_not_called()

tests/test_shape_check_hook.py

@@ -0,0 +1,171 @@
+"""The end-of-turn shape check (milestone 439): the write hooks note what each
+write defined, and the Stop hook asks the instance which of those nobody has
+judged, blocking once in the server's words.
+
+Runs the real shell against the shared HTTP sink. What it pins: the write
+hooks' ledger (a Write of a new file adds a `file` line; a skipped path adds
+nothing); silence with nothing written; a block only on a reason the instance
+returned; the post-block stop recorded and let through; another hook's block
+loop left alone; the ledger kept when the instance could not be reached.
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from tests.helpers import http_sink
+
+HOOKS = Path(__file__).resolve().parents[1] / "plugin" / "hooks"
+CHECK = HOOKS / "scribe_shape_check.sh"
+PRIOR_ART = HOOKS / "scribe_prior_art.sh"
+REASON = "SERVER REASON: judge what you wrote"
+
+
+def _env(tmp_path, url="http://127.0.0.1:9"):
+    for tool in ("curl", "bash", "git"):
+        if shutil.which(tool) is None:
+            pytest.skip(f"hook runtime tool {tool!r} not installed")
+    return {"PATH": os.environ["PATH"], "SCRIBE_URL": url, "SCRIBE_TOKEN": "t",
+            "TMPDIR": str(tmp_path), "HOME": str(tmp_path)}
+
+
+def _ledger(tmp_path, session="s1"):
+    return tmp_path / "scribe-priorart" / f"{session}.written.ids"
+
+
+def _write_ledger(tmp_path, lines, session="s1"):
+    path = _ledger(tmp_path, session)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("".join(f"{line}\n" for line in lines))
+    return path
+
+
+def _stop(env, cwd, active=False, session="s1"):
+    out = subprocess.run(
+        ["bash", str(CHECK)],
+        input=json.dumps({"session_id": session, "transcript_path": "/nonexistent",
+                          "cwd": str(cwd), "hook_event_name": "Stop",
+                          "stop_hook_active": active}),
+        capture_output=True, text=True, env=env, timeout=30,
+    )
+    assert out.returncode == 0, out.stderr
+    return out.stdout.strip()
+
+
+def _repo(tmp_path):
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    subprocess.run(["git", "init", "-q", str(repo)], check=True)
+    subprocess.run(["git", "-C", str(repo), "remote", "add", "origin",
+                    "https://git.example.com/alice/widget.git"], check=True)
+    return repo
+
+
+# ── the write hooks keep the ledger ───────────────────────────────────────
+
+def _pre_write(env, repo, rel, content, tool="Write"):
+    field = "content" if tool == "Write" else "new_string"
+    out = subprocess.run(
+        ["bash", str(PRIOR_ART)],
+        input=json.dumps({"session_id": "s1", "cwd": str(repo), "tool_name": tool,
+                          "tool_input": {"file_path": str(repo / rel), field: content}}),
+        capture_output=True, text=True, env=env, timeout=30,
+    )
+    assert out.returncode == 0, out.stderr
+
+
+def test_a_write_notes_what_it_defined_and_a_new_file_is_a_candidate(tmp_path):
+    with http_sink(b'{"note_ids":[]}') as (port, _seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        repo = _repo(tmp_path)
+        (repo / "web").mkdir()
+        _pre_write(env, repo, "web/StatusChip.svelte",
+                   "<script>\nfunction tone(s) {\n  return s;\n}\n</script>\n"
+                   "<style>\n.chip {\n  color: red;\n}\n</style>\n")
+        lines = _ledger(tmp_path).read_text().splitlines()
+        assert "web/StatusChip.svelte\tfile\tStatusChip" in lines
+        assert "web/StatusChip.svelte\tsym\ttone" in lines
+        assert "web/StatusChip.svelte\tcss\tchip" in lines
+
+
+def test_an_edit_to_an_existing_file_adds_no_file_line(tmp_path):
+    with http_sink(b'{"note_ids":[]}') as (port, _seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        repo = _repo(tmp_path)
+        (repo / "lib.py").write_text("x = 1\n")
+        _pre_write(env, repo, "lib.py", "def helper():\n    return 1\n", tool="Edit")
+        lines = _ledger(tmp_path).read_text().splitlines()
+        assert lines == ["lib.py\tsym\thelper"]
+
+
+def test_a_skipped_path_notes_nothing(tmp_path):
+    with http_sink(b'{"note_ids":[]}') as (port, _seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        repo = _repo(tmp_path)
+        _pre_write(env, repo, "README.md", "# def helper():\n")
+        assert not _ledger(tmp_path).exists()
+
+
+# ── the Stop hook asks ────────────────────────────────────────────────────
+
+def test_nothing_written_is_silent_and_asks_nothing(tmp_path):
+    with http_sink(b'{"status":"ok","reason":"x"}') as (port, seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        assert _stop(env, _repo(tmp_path)) == ""
+        assert seen == []
+
+
+def test_all_judged_passes_silently_and_empties_the_ledger(tmp_path):
+    with http_sink(b'{"status":"ok","outcome":"passed","unjudged":[]}') as (port, seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        ledger = _write_ledger(tmp_path, ["a.py\tsym\thelper", "a.py\tsym\thelper"])
+        assert _stop(env, _repo(tmp_path)) == ""
+        assert seen[0]["phase"] == ["check"]
+        assert seen[0]["written"] == ["a.py\tsym\thelper"], "duplicates collapse"
+        assert seen[0]["repo"] == ["https://git.example.com/alice/widget.git"]
+        assert not ledger.exists()
+
+
+def test_unjudged_blocks_once_in_the_servers_words_then_records_the_answer(tmp_path):
+    reply = json.dumps({"status": "ok", "outcome": "blocked", "reason": REASON}).encode()
+    with http_sink(reply) as (port, seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        repo = _repo(tmp_path)
+        ledger = _write_ledger(tmp_path, ["web/A.svelte\tfile\tA"])
+        assert json.loads(_stop(env, repo)) == {"decision": "block", "reason": REASON}
+        assert ledger.exists(), "kept, so the stop after the block asks about the same set"
+
+        # The stop after the block: recorded, never blocked, ledger consumed.
+        assert _stop(env, repo, active=True) == ""
+        assert [q["phase"] for q in seen] == [["check"], ["after"]]
+        assert not ledger.exists()
+        assert _stop(env, repo, active=True) == ""
+        assert len(seen) == 2
+
+
+def test_no_block_without_a_reason_from_the_instance(tmp_path):
+    with http_sink(b'{"status":"ok"}') as (port, _seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        _write_ledger(tmp_path, ["a.py\tsym\thelper"])
+        assert _stop(env, _repo(tmp_path)) == ""
+
+
+def test_another_hooks_block_loop_is_left_alone(tmp_path):
+    with http_sink(b'{"status":"ok","reason":"x"}') as (port, seen):
+        env = _env(tmp_path, f"http://127.0.0.1:{port}")
+        ledger = _write_ledger(tmp_path, ["a.py\tsym\thelper"])
+        assert _stop(env, _repo(tmp_path), active=True) == ""
+        assert seen == []
+        assert ledger.exists(), "this hook's own question is still owed"
+
+
+def test_an_unreachable_instance_keeps_the_question_for_later(tmp_path):
+    env = _env(tmp_path)          # port 9: nothing listens
+    ledger = _write_ledger(tmp_path, ["a.py\tsym\thelper"])
+    assert _stop(env, _repo(tmp_path)) == ""
+    assert ledger.exists()