feat(shapes): the write-path hook suggests and no longer stamps (milestone 439 step 4)
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / TypeScript typecheck (push) Successful in 53s
CI & Build / integration (push) Successful in 54s
CI & Build / Python tests (push) Successful in 1m49s
CI & Build / Build & push image (push) Successful in 27s

The hook used to land its evidence as `instance` rows, classified_by=hook —
a permanent verdict nobody read, and the source of the weak stamps that
poisoned two ledgers (#4608). Now the agent that wrote the code judges it at
the end of the turn, and the hook's evidence is what it is shown.

- shape_ledger.suggest_write_path_instances (was stamp_write_path_instances):
  same evidence and form gate, but it writes a PROPOSAL — proposed_snippet_id,
  basis "reference" (named) or "semantic" (resembles), score — and never a
  status. A judged row is left alone; a new shape gets an unclassified
  provisional row so the suggestion reaches the end-of-turn question as
  "looks like #N". By-name uses edges stay: a fact, not a verdict.
- The prior-art line says "looks like #N … when you judge this turn's
  shapes, say whether it is", and the result key is `suggested`.
- write_time_divergence takes `suggested`; _RESEMBLE_MIN re-documented as the
  suggestion bar and the weak-stamp line.
- reusing-code skill no longer says the pull stamps an instance.
- Tests moved to the new contract, plus a guard that nothing the hook does
  writes classified_by="hook".

Plugin version minted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-01 08:42:10 -04:00
co-authored by Claude Opus 5.5
parent f1fbdf746a
commit 4cf1c6f042
7 changed files with 150 additions and 110 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
{ {
"name": "scribe", "name": "scribe",
"description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).", "description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).",
"version": "2026.10.01.1239", "version": "2026.10.01.1242",
"author": { "author": {
"name": "Bryan Van Deusen" "name": "Bryan Van Deusen"
}, },
+3 -3
View File
@@ -34,9 +34,9 @@ through recall/auto-inject; this skill is the active reflex around that.
to ask both at once. to ask both at once.
- If a snippet fits, pull it in full with `get_snippet(id)` and reuse it — its - If a snippet fits, pull it in full with `get_snippet(id)` and reuse it — its
`location` points at the reference implementation. Adapt, don't re-derive. `location` points at the reference implementation. Adapt, don't re-derive.
The pull also does the accounting: the code you then write that references The pull also feeds the accounting: the code you then write that references
or resembles it is stamped an `instance` of that canon in the shape ledger or resembles it is offered back to you as "looks like #N" when you judge the
(classified_by=hook) — reuse from memory leaves no row. turn's shapes — reuse from memory leaves no such evidence.
- If auto-inject already surfaced a snippet title that looks relevant, that's - If auto-inject already surfaced a snippet title that looks relevant, that's
your cue to `get_snippet` it rather than start from scratch. your cue to `get_snippet` it rather than start from scratch.
- **Prior art offered beside a write is not noise — read it.** When Scribe notes - **Prior art offered beside a write is not noise — read it.** When Scribe notes
+6 -4
View File
@@ -270,10 +270,12 @@ async def write_path_prior_art():
css|sym. The shape ledger's write-path feed css|sym. The shape ledger's write-path feed
(#2791): when the session recently PULLED a (#2791): when the session recently PULLED a
snippet this payload references or resembles, snippet this payload references or resembles,
these land as instance rows (classified_by=hook). these carry it as a proposal — evidence for the
Honoured only for a caller allowed to write — a agent's own end-of-turn judgment, never a
read-scoped key still gets the hint, and never verdict (milestone 439). Honoured only for a
changes accounting on a GET. caller allowed to write — a read-scoped key still
gets the hint, and never changes the ledger on a
GET.
""" """
path = (request.args.get("path") or "").strip() path = (request.args.get("path") or "").strip()
code = request.args.get("code") or "" code = request.args.get("code") or ""
+29 -26
View File
@@ -2083,16 +2083,18 @@ async def build_write_path_hint(
``stamp_shapes`` turns the same request into the ledger's write-path feed ``stamp_shapes`` turns the same request into the ledger's write-path feed
(#2791): the (kind, name) definitions the hook saw in — or enclosing — (#2791): the (kind, name) definitions the hook saw in — or enclosing —
the payload. When the session has PULLED a snippet recently and this the payload. When the session has PULLED a snippet recently and this
payload references or resembles it, those shapes land as `instance` rows payload references or resembles it, those shapes carry it as a PROPOSAL
(classified_by=hook, see shape_ledger.stamp_write_path_instances) and (see shape_ledger.suggest_write_path_instances) and the result's
the result's ``stamped`` lists them. The route passes it only for a ``suggested`` lists them. Never a verdict since milestone 439: the agent
caller allowed to write — a read-scoped key gets the hint, never the that wrote the code judges it at the end of the turn, shown this as
stamp. ``repo_key`` (the hook's remote, normalised) homes a provisional evidence. The route passes it only for a caller allowed to write — a
row for a shape the ledger has not synced yet. read-scoped key gets the hint, never a ledger write. ``repo_key`` (the
hook's remote, normalised) homes a provisional row for a shape the
ledger has not synced yet.
""" """
cfg = await get_writepath_config(user_id) cfg = await get_writepath_config(user_id)
empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg, empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg,
"stamped": [], "divergence": [], "derive": [], "derive_keys": [], "suggested": [], "divergence": [], "derive": [], "derive_keys": [],
"rule_ids": []} "rule_ids": []}
path = (path or "").strip() path = (path or "").strip()
if not cfg["enabled"] or not path: if not cfg["enabled"] or not path:
@@ -2356,18 +2358,18 @@ async def build_write_path_hint(
menu = (placed + scored)[:max(0, top_k - len(synced))] menu = (placed + scored)[:max(0, top_k - len(synced))]
# The stamp runs whether or not anything is rendered — after dedup, the # The suggestion runs whether or not anything is rendered — after dedup,
# common case is a silent hint and a pulled canon being instantiated. # the common case is a silent hint and a pulled canon being instantiated.
stamped: list[dict] = [] suggested: list[dict] = []
if stamp_shapes and pulled: if stamp_shapes and pulled:
try: try:
stamped = await shape_ledger_svc.stamp_write_path_instances( suggested = await shape_ledger_svc.suggest_write_path_instances(
user_id, project_id, path=path, shapes=stamp_shapes, user_id, project_id, path=path, shapes=stamp_shapes,
code=code or "", pulled=pulled, resembles=resembles, code=code or "", pulled=pulled, resembles=resembles,
repo_key=repo_key, repo_key=repo_key,
) )
except Exception: except Exception:
logger.warning("Write-path ledger stamping failed", exc_info=True) logger.warning("Write-path ledger suggestion failed", exc_info=True)
# The in-band button-B check (#2793): the hook named the shapes being # The in-band button-B check (#2793): the hook named the shapes being
# written; if this directory+kind is canon-dense and a named shape isn't # written; if this directory+kind is canon-dense and a named shape isn't
# (about to be) an instance of that canon, say so NOW — at the write, # (about to be) an instance of that canon, say so NOW — at the write,
@@ -2376,7 +2378,7 @@ async def build_write_path_hint(
if stamp_shapes and project_id: if stamp_shapes and project_id:
try: try:
divergence = await shape_ledger_svc.write_time_divergence( divergence = await shape_ledger_svc.write_time_divergence(
project_id, path, stamp_shapes, stamped, code or "", project_id, path, stamp_shapes, suggested, code or "",
) )
except Exception: except Exception:
logger.warning("write-time divergence check failed", exc_info=True) logger.warning("write-time divergence check failed", exc_info=True)
@@ -2430,7 +2432,7 @@ async def build_write_path_hint(
design_text, design_dedup = await _design_arm( design_text, design_dedup = await _design_arm(
user_id, project_id, path, set(exclude_derive or []), user_id, project_id, path, set(exclude_derive or []),
) )
if not staleness and not synced and not menu and not stamped and not divergence and not derive: if not staleness and not synced and not menu and not suggested and not divergence and not derive:
if design_text: if design_text:
return {**empty, "context": design_text, "derive_keys": [design_dedup]} return {**empty, "context": design_text, "derive_keys": [design_dedup]}
return empty return empty
@@ -2531,8 +2533,8 @@ async def build_write_path_hint(
if passage: if passage:
lines.append(f"> ↳ {passage}") lines.append(f"> ↳ {passage}")
if stamped: if suggested:
lines.append(_stamp_line(path, stamped)) lines.append(_suggest_line(path, suggested))
if divergence: if divergence:
lines.append(_divergence_line(path, divergence)) lines.append(_divergence_line(path, divergence))
if derive: if derive:
@@ -2694,7 +2696,7 @@ async def build_write_path_hint(
"note_ids": note_ids, "note_ids": note_ids,
"sync_note_ids": sync_note_ids, "sync_note_ids": sync_note_ids,
"config": cfg, "config": cfg,
"stamped": stamped, "suggested": suggested,
"divergence": divergence, "divergence": divergence,
"derive": derive, "derive": derive,
"derive_keys": [d["key"] for d in derive] + ( "derive_keys": [d["key"] for d in derive] + (
@@ -3017,22 +3019,23 @@ def _divergence_line(path: str, divergence: list[dict]) -> str:
) )
def _stamp_line(path: str, stamped: list[dict]) -> str: def _suggest_line(path: str, suggested: list[dict]) -> str:
"""One line saying what the ledger just recorded, so the session can """One line naming the recorded shape this code looks like, as evidence
correct a wrong stamp in the moment rather than an audit finding it.""" for the judgment the agent gives at the end of the turn (milestone 439)
— not a record of one already made."""
by_snippet: dict[int, list[str]] = {} by_snippet: dict[int, list[str]] = {}
for row in stamped: for row in suggested:
label = f".{row['symbol']}" if row["kind"] == "css" else row["symbol"] label = f".{row['symbol']}" if row["kind"] == "css" else row["symbol"]
by_snippet.setdefault(int(row["snippet_id"]), []).append(f"`{label}`") by_snippet.setdefault(int(row["snippet_id"]), []).append(f"`{label}`")
parts = [ parts = [
f"{', '.join(names)} → instance of #{sid}" f"{', '.join(names)} → looks like #{sid}"
for sid, names in by_snippet.items() for sid, names in by_snippet.items()
] ]
return ( return (
f"> Shape accounting: recorded at `{path}` — {'; '.join(parts)} " f"> Shape accounting: at `{path}` — {'; '.join(parts)} "
"(classified_by=hook: you pulled that snippet this session and this " "(you pulled that snippet this session and this code names or resembles "
"code references/resembles it). Not an instance? `classify_shapes` " "it). When you judge this turn's shapes, say whether it is: `instance` "
"overrides a hook stamp." "of it, `variant` with the why, or something else."
) )
+45 -25
View File
@@ -908,11 +908,13 @@ async def snippet_consumers(user_id: int, note_id: int) -> dict:
# semantic arm scored it above the write-path threshold for this # semantic arm scored it above the write-path threshold for this
# very payload. Either is evidence; the pull alone is not. # very payload. Either is evidence; the pull alone is not.
# Both hold → every shape the hook named at that path, of the snippet's kind, # Both hold → every shape the hook named at that path, of the snippet's kind,
# becomes instance-of-N with classified_by="hook" and the evidence as reason. # carries N as a PROPOSAL (milestone 439). Until then it became instance-of-N
# with classified_by="hook" — a verdict nobody read, and the rows written that
# way are what `stamps_to_review` still lists.
# #
# A hook row is EVIDENCE, not judgment: it only ever lands on rows nobody has # A proposal is EVIDENCE, not judgment: it lands only on rows nobody has
# judged (unclassified) or rows an earlier hook stamped, never on a canonical # judged, never touches a status, and is what the end-of-turn question shows
# row or an agent/audit/import judgment. Re-judge with classify_shapes. # the agent as "looks like #N". The agent's classify_shapes is the verdict.
# "The write path actually pulled it": a working session's reach. The # "The write path actually pulled it": a working session's reach. The
# precision comes from the in-play test above, not from this window. # precision comes from the in-play test above, not from this window.
@@ -1354,7 +1356,7 @@ def _stamp_allowed(rank: int, mine: str, canon: str) -> bool:
return forms_agree(mine, canon) return forms_agree(mine, canon)
async def stamp_write_path_instances( async def suggest_write_path_instances(
user_id: int, user_id: int,
project_id: int, project_id: int,
*, *,
@@ -1365,22 +1367,35 @@ async def stamp_write_path_instances(
resembles: dict[int, float] | None = None, resembles: dict[int, float] | None = None,
repo_key: str = "", repo_key: str = "",
) -> list[dict]: ) -> list[dict]:
"""Land hook evidence as `instance` rows for the shapes being written. """Offer hook evidence as a PROPOSAL on the shapes being written.
This used to land the evidence as `instance` rows, classified_by=hook —
a permanent classification written with nobody reading it. That is what
poisoned two ledgers (132 and 488 rows below today's floor, #4608), and
what milestone 439 retires: the agent that wrote the code says what it is,
at the end of the turn, and this evidence is what it is shown when it
does. So the row keeps its status; the candidate canon goes into the
proposal fields (`proposed_snippet_id`, `proposal_basis` "reference" for a
by-name hit or "semantic" for a resemblance, `proposal_score`), where the
end-of-turn question reads it as "looks like #N". A judged row is left
alone entirely.
``shapes`` is the hook's (kind, name) list for ``path``; ``pulled`` is ``shapes`` is the hook's (kind, name) list for ``path``; ``pulled`` is
recent_pulls(); ``resembles`` maps snippet ids the semantic arm scored recent_pulls(); ``resembles`` maps snippet ids the semantic arm scored
for this payload to their score. Returns the rows stamped, each for this payload to their score. Returns the suggestions, each
{path, symbol, kind, snippet_id, reason} — empty in the common case. {path, symbol, kind, snippet_id, reason} — empty in the common case.
A shape the ledger has no live row for yet (it is being written right A shape the ledger has no live row for yet (it is being written right
now) gets a PROVISIONAL row under ``repo_key`` — first/last-seen empty — now) gets a PROVISIONAL unclassified row under ``repo_key`` — first/
so the stamp is not lost to the next sync, which either confirms the last-seen empty — so the suggestion survives to the end of the turn and
shape (sets its seen marker) or stamps it vanished. No repo key → only the next sync either confirms the shape or stamps it vanished. No repo
existing rows are stamped. key → only existing rows carry a suggestion.
Every pulled canon the payload NAMES is still recorded as a `uses` edge:
that is a fact about the code, not a verdict about what it is.
When more than one pulled snippet is in play for a shape, a by-name When more than one pulled snippet is in play for a shape, a by-name
reference beats resemblance and the most recent pull breaks ties: a row reference beats resemblance and the most recent pull breaks ties.
holds one canon (the known model limit logged on #2790).
""" """
from scribe.services import access from scribe.services import access
from scribe.services import snippets as snippets_svc from scribe.services import snippets as snippets_svc
@@ -1433,8 +1448,7 @@ async def stamp_write_path_instances(
for bucket in in_play.values(): for bucket in in_play.values():
bucket.sort(key=lambda t: (t[0], t[1]), reverse=True) bucket.sort(key=lambda t: (t[0], t[1]), reverse=True)
now = datetime.now(timezone.utc) suggested: list[dict] = []
stamped: list[dict] = []
async with async_session() as session: async with async_session() as session:
rows = ( rows = (
await session.execute( await session.execute(
@@ -1486,9 +1500,12 @@ async def stamp_write_path_instances(
by_key[(name, kind)] = row by_key[(name, kind)] = row
elif not (row.status in _MECHANICAL_TODO or row.classified_by == "hook"): elif not (row.status in _MECHANICAL_TODO or row.classified_by == "hook"):
continue # a judgment — or the canon itself — stands continue # a judgment — or the canon itself — stands
await _judge(session, row, status="instance", snippet_id=sid, by="hook", # A proposal, never a status: the writer decides (milestone 439).
reason=why, at=now) row.proposed_snippet_id = sid
stamped.append({ row.proposal_basis = "reference" if _rank >= 2 else "semantic"
row.proposal_score = resembles.get(sid) if _rank < 2 else 1.0
row.proposal_group = None
suggested.append({
"path": path, "symbol": name, "kind": kind, "path": path, "symbol": name, "kind": kind,
"snippet_id": sid, "reason": why, "snippet_id": sid, "reason": why,
}) })
@@ -1504,9 +1521,9 @@ async def stamp_write_path_instances(
[t[2] for t in bucket if t[0] == 2], [t[2] for t in bucket if t[0] == 2],
basis="hook", evidence="write path: pulled the snippet, payload names its symbol", basis="hook", evidence="write path: pulled the snippet, payload names its symbol",
) )
if stamped: if suggested:
await session.commit() await session.commit()
return stamped return suggested
# --- the mechanical proposer (#2792): the machine proposes, judgment classifies # --- the mechanical proposer (#2792): the machine proposes, judgment classifies
@@ -2210,8 +2227,11 @@ async def confirm_proposals(
# A canon dominates a directory+kind when at least this many siblings are # A canon dominates a directory+kind when at least this many siblings are
# judged (canonical/instance) and this share of them answer to one snippet. # judged (canonical/instance) and this share of them answer to one snippet.
# How much a payload must resemble a canon before the hook may assert, with # How much a payload must resemble a canon before the hook SUGGESTS it as the
# nobody watching, that a shape IS an instance of it. # shape's canon — and, for the rows stamped before milestone 439, the bar below
# which a stored stamp is weak (`is_weak_stamp`). Until milestone 439 this was
# the bar for the hook to ASSERT, unattended, that a shape IS an instance; the
# hook no longer asserts anything, and the history below is why.
# #
# WHY A FLOOR AT ALL. There was none: any score the semantic arm produced # WHY A FLOOR AT ALL. There was none: any score the semantic arm produced
# counted, so 0.69 asserted as confidently as 0.95, and the score went into # counted, so 0.69 asserted as confidently as 0.95, and the score went into
@@ -2364,15 +2384,15 @@ async def canon_density(
async def write_time_divergence( async def write_time_divergence(
project_id: int, path: str, shapes: list[tuple[str, str]], stamped: list[dict], project_id: int, path: str, shapes: list[tuple[str, str]], suggested: list[dict],
code: str = "", code: str = "",
) -> list[dict]: ) -> list[dict]:
"""The in-band check for the shapes the hook named at ``path``: for each """The in-band check for the shapes the hook named at ``path``: for each
kind whose directory has a dominant canon, the named shapes that are kind whose directory has a dominant canon, the named shapes that are
not (already or just now) that canon's instance/canonical — new or not that canon's instance/canonical (judged, or just suggested as it) — new or
unclassified rows only; a judged shape is not re-litigated at every unclassified rows only; a judged shape is not re-litigated at every
edit. Returns [{symbol, kind, canon_snippet_id, instances, judged}].""" edit. Returns [{symbol, kind, canon_snippet_id, instances, judged}]."""
just_stamped = {(s["symbol"], s["kind"]): s["snippet_id"] for s in stamped} just_stamped = {(s["symbol"], s["kind"]): s["snippet_id"] for s in suggested}
out: list[dict] = [] out: list[dict] = []
if not shapes: if not shapes:
return out return out
+52 -37
View File
@@ -322,15 +322,16 @@ async def test_sync_refiles_rows_whose_snippet_was_purged(seeded):
@pytest.mark.integration @pytest.mark.integration
async def test_write_path_stamp_is_evidence_that_yields_to_judgment(seeded): async def test_write_path_evidence_is_a_proposal_never_a_verdict(seeded):
"""Pulled + referenced → every named shape of the snippet's kind becomes """Milestone 439. Pulled + referenced → every named shape of the
an instance row, classified_by=hook, carrying the evidence as reason. A snippet's kind CARRIES the snippet as a proposal (basis "reference"),
later agent judgment on one of them stands against a re-stamp; the hook and its status is untouched: the agent that wrote it says what it is.
may only overwrite nobody's judgment or its own. The outsider stamps A judgment is never touched, not even by a proposal. The outsider
nothing (write-gated like every other ledger write).""" writes nothing (write-gated like every other ledger write). The uses
edge — a fact about the code — is still recorded."""
from datetime import datetime, timezone from datetime import datetime, timezone
from scribe.services.shape_ledger import stamp_write_path_instances from scribe.services.shape_ledger import suggest_write_path_instances
owner, other, pid, sid = ( owner, other, pid, sid = (
seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"] seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
@@ -338,84 +339,98 @@ async def test_write_path_stamp_is_evidence_that_yields_to_judgment(seeded):
pulled = {sid: datetime.now(timezone.utc)} pulled = {sid: datetime.now(timezone.utc)}
code = "app = factory()\nreturn app\n" # references the snippet's symbol code = "app = factory()\nreturn app\n" # references the snippet's symbol
assert await stamp_write_path_instances( assert await suggest_write_path_instances(
other, pid, path="src/app.py", shapes=[("sym", "make_app")], other, pid, path="src/app.py", shapes=[("sym", "make_app")],
code=code, pulled=pulled, code=code, pulled=pulled,
) == [] ) == []
stamped = await stamp_write_path_instances( suggested = await suggest_write_path_instances(
owner, pid, path="src/app.py", owner, pid, path="src/app.py",
shapes=[("sym", "make_app"), ("sym", "Config"), ("css", "nope")], shapes=[("sym", "make_app"), ("sym", "Config"), ("css", "nope")],
code=code, pulled=pulled, code=code, pulled=pulled,
) )
assert {s["symbol"] for s in stamped} == {"make_app", "Config"} # css skipped: no css canon assert {s["symbol"] for s in suggested} == {"make_app", "Config"} # css skipped: no css canon
rows, _ = await list_project_shapes(owner, pid, snippet_id=sid) rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
by_symbol = {r.symbol: r for r in rows} by_symbol = {r.symbol: r for r in rows}
assert by_symbol["make_app"].status == "instance" for name in ("make_app", "Config"):
assert by_symbol["make_app"].classified_by == "hook" row = by_symbol[name]
assert by_symbol["make_app"].reason == f"hook: pulled #{sid}; payload references `factory`" assert row.status == "unclassified" and row.classified_by is None, name
assert (row.proposed_snippet_id, row.proposal_basis) == (sid, "reference"), name
# The by-name reference is still recorded as a call-site fact.
users, _ = await list_project_shapes(owner, pid, uses=sid)
assert {"make_app", "Config"} <= {r.symbol for r in users}
# A judgment lands; the next stamp must leave it alone but may re-stamp # A judgment stands: the next suggestion leaves the judged row alone.
# its own earlier row.
await classify_shapes(owner, pid, [ await classify_shapes(owner, pid, [
{"path": "src/app.py", "symbol": "make_app", "status": "exempt", {"path": "src/app.py", "symbol": "make_app", "status": "exempt",
"reason": "the app factory is its own thing"}, "reason": "the app factory is its own thing"},
]) ])
again = await stamp_write_path_instances( again = await suggest_write_path_instances(
owner, pid, path="src/app.py", owner, pid, path="src/app.py",
shapes=[("sym", "make_app"), ("sym", "Config")], code=code, pulled=pulled, shapes=[("sym", "make_app"), ("sym", "Config")], code=code, pulled=pulled,
) )
assert {s["symbol"] for s in again} == {"Config"} assert {s["symbol"] for s in again} == {"Config"}
rows, _ = await list_project_shapes(owner, pid, path="src/app.py") rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
by_symbol = {r.symbol: r for r in rows} assert {r.symbol: r.status for r in rows}["make_app"] == "exempt"
assert by_symbol["make_app"].status == "exempt"
assert by_symbol["Config"].status == "instance"
# Neither pulled nor in play → nothing, even with shapes named. # Neither pulled nor in play → nothing, even with shapes named.
assert await stamp_write_path_instances( assert await suggest_write_path_instances(
owner, pid, path="src/util.py", shapes=[("sym", "helper")], owner, pid, path="src/util.py", shapes=[("sym", "helper")],
code="print('unrelated')", pulled=pulled, code="print('unrelated')", pulled=pulled,
) == [] ) == []
# And nothing the hook does writes classified_by="hook" any more.
async with async_session() as s:
hook_rows = (await s.execute(select(CodeShape).where(
CodeShape.project_id == pid, CodeShape.classified_by == "hook",
))).scalars().all()
assert hook_rows == []
@pytest.mark.integration @pytest.mark.integration
async def test_a_brand_new_shape_gets_a_provisional_row_the_sync_settles(seeded): async def test_a_brand_new_shape_gets_a_provisional_row_carrying_the_suggestion(seeded):
"""The shape being written right now has no ledger row yet. With the """The shape being written right now has no ledger row yet. With the
hook's repo key it gets a provisional one — seen markers empty — so the hook's repo key it gets a provisional, UNCLASSIFIED one — seen markers
stamp survives until the next sync, which confirms it (sets the marker) empty — carrying the suggestion to the end-of-turn question; the sync
or stamps it vanished. Without a repo key only existing rows are touched.""" confirms it or stamps it vanished. Without a repo key only existing rows
carry a suggestion."""
from datetime import datetime, timezone from datetime import datetime, timezone
from scribe.services.shape_ledger import stamp_write_path_instances from scribe.services.shape_ledger import suggest_write_path_instances, unjudged_shapes
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"] owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
pulled = {sid: datetime.now(timezone.utc)} pulled = {sid: datetime.now(timezone.utc)}
code = "def build():\n return factory()\n" code = "def build():\n return factory()\n"
assert await stamp_write_path_instances( assert await suggest_write_path_instances(
owner, pid, path="src/new.py", shapes=[("sym", "build")], owner, pid, path="src/new.py", shapes=[("sym", "build")],
code=code, pulled=pulled, # no repo_key code=code, pulled=pulled, # no repo_key
) == [] ) == []
stamped = await stamp_write_path_instances( suggested = await suggest_write_path_instances(
owner, pid, path="src/new.py", shapes=[("sym", "build")], owner, pid, path="src/new.py", shapes=[("sym", "build")],
code=code, pulled=pulled, repo_key=REPO, code=code, pulled=pulled, repo_key=REPO,
) )
assert [s["symbol"] for s in stamped] == ["build"] assert [s["symbol"] for s in suggested] == ["build"]
async with async_session() as s: async with async_session() as s:
row = (await s.execute(select(CodeShape).where( row = (await s.execute(select(CodeShape).where(
CodeShape.project_id == pid, CodeShape.path == "src/new.py", CodeShape.project_id == pid, CodeShape.path == "src/new.py",
))).scalar_one() ))).scalar_one()
assert row.status == "instance" and row.classified_by == "hook" assert row.status == "unclassified" and row.classified_by is None
assert row.proposed_snippet_id == sid
# "Unset" is the column's empty default — the markers are non-null # "Unset" is the column's empty default — the markers are non-null
# Text, and the sync is what first fills them. # Text, and the sync is what first fills them.
assert row.first_seen_commit == "" and row.last_seen_commit == "" assert row.first_seen_commit == "" and row.last_seen_commit == ""
# The sync sees the shape in the tree → confirmed, stamp intact. # The end-of-turn question shows the suggestion as evidence.
asked = await unjudged_shapes(pid, [("src/new.py", "sym", "build")], repo_key=REPO)
assert asked[0]["evidence"]["looks_like"]["snippet_id"] == sid
# The sync sees the shape in the tree → confirmed, still unjudged.
await sync_repo_shapes( await sync_repo_shapes(
pid, REPO, SHAPES + [("src/new.py", "sym", "build")], seen_marker="abc123", pid, REPO, SHAPES + [("src/new.py", "sym", "build")], seen_marker="abc123",
) )
rows, _ = await list_project_shapes(owner, pid, path="src/new.py") rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
assert rows[0].status == "instance" and rows[0].last_seen_commit == "abc123" assert rows[0].status == "unclassified" and rows[0].last_seen_commit == "abc123"
# The sync no longer sees it → vanished, out of the live accounting. # The sync no longer sees it → vanished, out of the live accounting.
await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="def456") await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="def456")
@@ -866,19 +881,19 @@ async def test_a_second_confirm_dialog_is_detected_and_named(seeded):
# In-band: the hook names the shape at write time → the check names the canon. # In-band: the hook names the shape at write time → the check names the canon.
named = await write_time_divergence( named = await write_time_divergence(
pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], stamped=[] pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], suggested=[]
) )
assert named == [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": sid, assert named == [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": sid,
"instances": 4, "judged": 4}] "instances": 4, "judged": 4}]
# ...but an already-judged shape, or one just stamped as the canon's # ...but an already-judged shape, or one just stamped as the canon's
# instance, is not re-litigated. # instance, is not re-litigated.
assert await write_time_divergence(pid, f"{comp}/Trash.vue", [("sym", "onTrash")], stamped=[]) == [] assert await write_time_divergence(pid, f"{comp}/Trash.vue", [("sym", "onTrash")], suggested=[]) == []
assert await write_time_divergence( assert await write_time_divergence(
pid, f"{comp}/New.vue", [("sym", "onNew")], pid, f"{comp}/New.vue", [("sym", "onNew")],
stamped=[{"symbol": "onNew", "kind": "sym", "snippet_id": sid}], suggested=[{"symbol": "onNew", "kind": "sym", "snippet_id": sid}],
) == [] ) == []
# A directory with no dominant canon is silent. # A directory with no dominant canon is silent.
assert await write_time_divergence(pid, "src/other.py", [("sym", "thing")], stamped=[]) == [] assert await write_time_divergence(pid, "src/other.py", [("sym", "thing")], suggested=[]) == []
# The judgment answers the question and clears the flag. # The judgment answers the question and clears the flag.
await classify_shapes(owner, pid, [ await classify_shapes(owner, pid, [
@@ -956,7 +971,7 @@ async def test_weak_stamps_elect_no_canon_and_their_flags_are_withdrawn(seeded):
rows, total = await list_project_shapes(owner, pid, flag="divergence") rows, total = await list_project_shapes(owner, pid, flag="divergence")
assert total == 0, "a flag whose canon no longer dominates is withdrawn" assert total == 0, "a flag whose canon no longer dominates is withdrawn"
assert await write_time_divergence( assert await write_time_divergence(
pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], stamped=[] pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], suggested=[]
) == [] ) == []
# Nothing was un-judged: the stamps stand until someone reads them. # Nothing was un-judged: the stamps stand until someone reads them.
async with async_session() as session: async with async_session() as session:
+14 -14
View File
@@ -1291,9 +1291,9 @@ async def test_stamping_needs_named_shapes_and_a_recent_pull():
patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
patch.object(pc, "record_retrieval", MagicMock()), \ patch.object(pc, "record_retrieval", MagicMock()), \
patch.object(pc.shape_ledger_svc, "recent_pulls", pulls), \ patch.object(pc.shape_ledger_svc, "recent_pulls", pulls), \
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp): patch.object(pc.shape_ledger_svc, "suggest_write_path_instances", stamp):
out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, project_id=4) out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, project_id=4)
assert out["stamped"] == [] assert out["suggested"] == []
pulls.assert_not_awaited() # no shapes → no read pulls.assert_not_awaited() # no shapes → no read
out = await pc.build_write_path_hint( out = await pc.build_write_path_hint(
1, "src/x.py", code=REAL_CODE, project_id=4, 1, "src/x.py", code=REAL_CODE, project_id=4,
@@ -1301,7 +1301,7 @@ async def test_stamping_needs_named_shapes_and_a_recent_pull():
) )
pulls.assert_awaited_once() pulls.assert_awaited_once()
stamp.assert_not_awaited() # shapes, but no pull stamp.assert_not_awaited() # shapes, but no pull
assert out["stamped"] == [] assert out["suggested"] == []
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -1330,7 +1330,7 @@ async def test_a_pulled_snippet_already_seen_is_evidence_not_menu():
patch.object(pc, "record_retrieval", MagicMock()), \ patch.object(pc, "record_retrieval", MagicMock()), \
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \ patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={7: _ts()})), \ patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={7: _ts()})), \
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp): patch.object(pc.shape_ledger_svc, "suggest_write_path_instances", stamp):
out = await pc.build_write_path_hint( out = await pc.build_write_path_hint(
1, "src/x.py", code=REAL_CODE, project_id=4, exclude_ids=[7], 1, "src/x.py", code=REAL_CODE, project_id=4, exclude_ids=[7],
stamp_shapes=[("sym", "debounce")], repo_key="git.example.com/a/b", stamp_shapes=[("sym", "debounce")], repo_key="git.example.com/a/b",
@@ -1355,11 +1355,11 @@ async def test_a_pulled_snippet_already_seen_is_evidence_not_menu():
assert skw["resembles"] == {7: 0.91} assert skw["resembles"] == {7: 0.91}
assert skw["shapes"] == [("sym", "debounce")] assert skw["shapes"] == [("sym", "debounce")]
assert skw["repo_key"] == "git.example.com/a/b" assert skw["repo_key"] == "git.example.com/a/b"
assert out["stamped"][0]["snippet_id"] == 7 assert out["suggested"][0]["snippet_id"] == 7
# And the session is told what landed, with the way to correct it. # And the session is shown it as evidence for its own end-of-turn judgment.
assert "Shape accounting" in out["context"] assert "Shape accounting" in out["context"]
assert "`debounce` → instance of #7" in out["context"] assert "`debounce` → looks like #7" in out["context"]
assert "classify_shapes" in out["context"] assert "When you judge this turn's shapes" in out["context"]
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -1376,15 +1376,15 @@ async def test_a_stamp_renders_even_when_the_hint_is_otherwise_silent():
patch.object(pc, "record_retrieval", MagicMock()), \ patch.object(pc, "record_retrieval", MagicMock()), \
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \ patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={5: _ts()})), \ patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={5: _ts()})), \
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", patch.object(pc.shape_ledger_svc, "suggest_write_path_instances",
AsyncMock(return_value=stamped)): AsyncMock(return_value=stamped)):
out = await pc.build_write_path_hint( out = await pc.build_write_path_hint(
1, "web/b.css", code=".btn-primary { color: red; }" * 4, project_id=4, 1, "web/b.css", code=".btn-primary { color: red; }" * 4, project_id=4,
stamp_shapes=[("css", "btn-primary")], stamp_shapes=[("css", "btn-primary")],
) )
assert out["note_ids"] == [] assert out["note_ids"] == []
assert out["stamped"] == stamped assert out["suggested"] == stamped
assert "`.btn-primary` → instance of #5" in out["context"] assert "`.btn-primary` → looks like #5" in out["context"]
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -1397,13 +1397,13 @@ async def test_a_failing_stamp_does_not_sink_the_hint():
patch.object(pc, "record_retrieval", MagicMock()), \ patch.object(pc, "record_retrieval", MagicMock()), \
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \ patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={12: _ts()})), \ patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={12: _ts()})), \
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", patch.object(pc.shape_ledger_svc, "suggest_write_path_instances",
AsyncMock(side_effect=RuntimeError("ledger down"))): AsyncMock(side_effect=RuntimeError("ledger down"))):
out = await pc.build_write_path_hint( out = await pc.build_write_path_hint(
1, "src/x.py", code=REAL_CODE, project_id=4, stamp_shapes=[("sym", "f")], 1, "src/x.py", code=REAL_CODE, project_id=4, stamp_shapes=[("sym", "f")],
) )
assert out["sync_note_ids"] == [12] assert out["sync_note_ids"] == [12]
assert out["stamped"] == [] assert out["suggested"] == []
def test_route_stamps_only_for_a_caller_allowed_to_write(): def test_route_stamps_only_for_a_caller_allowed_to_write():
@@ -1760,7 +1760,7 @@ async def test_a_record_this_arm_withheld_itself_is_not_a_near_miss():
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \ patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
patch.object(pc.shape_ledger_svc, "recent_pulls", patch.object(pc.shape_ledger_svc, "recent_pulls",
AsyncMock(return_value={7: _ts()})), \ AsyncMock(return_value={7: _ts()})), \
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", patch.object(pc.shape_ledger_svc, "suggest_write_path_instances",
AsyncMock(return_value=[])), \ AsyncMock(return_value=[])), \
patch.object(pc, "semantic_search_notes", patch.object(pc, "semantic_search_notes",
_search_reporting(0.9, fake_note( _search_reporting(0.9, fake_note(