plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "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).",
-  "version": "2026.09.20.2244",
+  "version": "2026.09.21.0300",
   "author": {
     "name": "Bryan Van Deusen"
   },

plugin/skills/reporting-back/SKILL.md

@@ -101,11 +101,45 @@ default differ, the preference is what they asked for.
 | Kind | Sections |
 |---|---|
 | **Completion** | Where this sits · What now works · How / why · Needs you · Next |
-| **Finding** (a problem you found and did not fix) | Symptom · Cause · Size of the fix · **Offer to fix it** |
+| **Finding** (a problem you found) | Symptom · Cause · **What you decided and did** — judge it and act; an offer to fix it is the judgment not made (see *You are the judge*) |
 | **Blocked / failed** | What stopped · What you tried · What you need from them |
 | **Progress** (mid-work) | One or two lines: where things are, what's next, any blocker |
 | **Where are we** | The milestone and its progress · Done · Open · Needs you · Next |
 
+## You are the judge
+
+You are the judge of record for the work itself: what a shape is, whether a
+finding holds, whether a record is right, whether something is done. The
+operator reads the decision — they do not make it.
+
+**A finding surfaced and not judged is a finding dropped, not deferred.** This
+is the failure that hides inside a good report: the symptom named, the cause
+traced, the fix sized, and then handed over for someone else to rule on. It
+reads as diligence and functions as a backlog. The tells are "say the word and
+I'll…", "your call", "let me know if you want me to…", and a list of options
+for a decision that was yours.
+
+So: decide, act, and report what you decided and why. If the evidence is
+genuinely balanced, say which way you went and what would change your mind —
+that is still a decision.
+
+**What DOES go to them**, and the distinction is the act, not the difficulty:
+spending their money, reaching their infrastructure, merging to a protected
+branch, anything hard to reverse or facing outward — the **Handoff** and
+**Approval** shapes below. Judging a record is never one of these. A hard call
+is still yours; an irreversible act is still theirs.
+
+**Judging is attended, not automatic.** You judge by reading the evidence and
+recording why. A threshold, a sweep or a rule that reclassifies in bulk with
+nobody reading it is not you judging — it is the thing that usually created
+the mess, wearing your name. When you catch yourself fixing bad unattended
+writes with another unattended write, stop.
+
+**Building a review surface?** Ask who its implied reader is. If the answer is
+"a person works through this queue", it is mis-designed: give the reader the
+evidence needed to decide and a way to record the decision under their own
+name.
+
 ## Asks — the operator needs to act or decide
 
 | Kind | Sections |

plugin/skills/using-scribe/SKILL.md

@@ -376,6 +376,35 @@ to the situation it applies to so a later session meets it there. Both are
 first-class outcomes of noticing something, not what's left when a rule
 proposal fails.
 
+## You are the judge of what the record says
+
+Scribe's shape exists to support how you work, so you are the judge of record
+for the work itself: what a shape is, which canon a thing belongs to, whether
+a finding holds, whether something is done. `classify_shapes`, `update_task`,
+`create_snippet` and the rest are where your judgment lands — under your own
+name, with the reasoning recorded beside it.
+
+**Judging is attended, and that is the whole distinction.** You judge by
+reading the evidence and writing down why. A threshold, a sweep or a hook that
+classifies in bulk with nobody reading it is not judgment; it is the thing
+that fills a ledger with confident nonsense. The write-path hook may offer
+evidence (`classified_by="hook"`) and any judgment of yours overrides it —
+that asymmetry is deliberate. When you are about to correct a pile of bad
+unattended writes with one more unattended write, stop and read them instead.
+
+**A finding you surface and do not judge is a finding dropped.** Handing a
+classification, a verdict or a done/not-done call to the operator "to decide"
+is the judgment not being made, however carefully it is written up. Keep
+escalating the acts that are genuinely theirs — their money, their
+infrastructure, anything hard to reverse or facing outward — and keep the
+decisions. A hard call is still yours; an irreversible act is still theirs.
+
+**A record that is wrong stays wrong until something rewrites it.** A guard
+added at the point of classification does not undo classifications already
+stored, so when you fix how judgments are made, go and look at the ones
+already there. That look is yours to do, and its result is a judgment you
+record, not a queue you leave behind.
+
 ## When a record doesn't reach the moment it should
 
 Retrieval misjudging is ordinary, and it is fixable — but only by whoever

src/scribe/mcp/server.py

@@ -34,36 +34,37 @@ from quart import Quart
 # for space before the ownership split (milestones 317, 333, 409) is in
 # decision #4027 and the notes it supersedes.
 _INSTRUCTIONS = """
-Scribe is the operator's system of record for their work, and yours: recall
+Scribe is the operator's system of record, and yours: recall before acting,
-from it before acting, record in it as you go, and keep one copy here rather
+record as you go, keep one copy here rather than in local memory files.
-than in local memory files.
 
-Every reflex below is stated in full in the using-scribe skill (if your
+Each reflex is stated in full in the using-scribe skill (if your client reads
-client reads Agent Skills) and in each tool's description. The index:
+Agent Skills) and each tool's description. The index:
 - ORIENT: enter_project(id) loads the project, open work, Systems and design
   system. An `inception` key: ask what it inherits, then
   decide_project_inception.
 - RULES: nothing preloads; a rule arrives when your work matches it. Before a
-  consequential act — or before handing work back unsure you may finish it —
+  consequential act, what_might_apply("what you are about to do") — fifty
-  what_might_apply("what you are about to do"): fifty ranked
+  ranked, no bar. search(content_type="rule") reads one you suspect. Silence
-  candidates, no bar. search(content_type="rule") reads one you already
+  means nothing matched, not none. Rules bind; preferences guide and you keep
-  suspect. Silence means nothing matched, not none. Rules bind; preferences
+  them current; lessons inform.
-  guide and you keep them current; lessons inform.
+- MISSED: a rule that missed you is a trigger to fix, not a floor to move
-- MISSED: a rule that missed you is a trigger to fix, not a
+  (retrieval_telemetry).
-  floor to move (retrieval_telemetry).
+- JUDGE: you are the judge of record — what a shape is, whether a finding
+  holds, whether work is done. Surfacing one for them to rule on is the
+  judgment not made. Escalate their acts, not your decisions.
 - RECALL: search before acting, scoped with the active project_id.
 - RECORD: create_task; a fix is kind="issue". add_task_log as you go; status
-  in_progress on start, done on finish. Tag system_ids as you write.
+  in_progress on start, done on finish. Tag system_ids.
-- PLAN work with an arc: find the existing plan first
+- PLAN with an arc: find the existing plan first
   (search(content_type="milestone")) and add steps to it; else
-  start_planning(steps=[...]). The plan is a milestone, each step a task.
+  start_planning(steps=[...]).
-- IDS exist only once a create returns them. Records that cite each other go
+- IDS exist only once a create returns them. Records citing each other go
   through create_records, writing {{ref:N}} for the Nth record.
 - REUSE: search snippets before building; create_snippet what you build.
 - UI: the project's design system binds; resolve_design_system before
   hand-writing a value.
-- REPORT back from the `placement` a task write returns: where the work sits,
+- REPORT from the `placement` a task write returns: where it sits, what
-  what changed, what needs the operator, what comes next.
+  changed, what needs them, what next.
 
 Creates are duplicate-gated: a near-match returns the existing id to update.
 shared:true records are another user's suggestion, not settled practice.
@@ -119,7 +120,7 @@ _READ_ONLY_TOOLS = frozenset({
     "list_repo_bindings",
     # The shape ledger's todo query (#2789). Reads only — classify_shapes is
     # the write, and it is deliberately NOT here.
-    "list_shapes", "shape_history",
+    "list_shapes", "shape_history", "stamps_to_review",
     # The retrieval telemetry readout (#2975). Aggregates two log tables and
     # writes nothing. Listed explicitly because its name carries no read
     # prefix, so the completeness test below cannot derive it — the same

src/scribe/mcp/tools/shapes.py

@@ -247,6 +247,42 @@ async def shape_history(
     )
 
 
+async def stamps_to_review(project_id: int, top: int = 10) -> dict:
+    """Judged rows whose evidence no longer meets the ledger's bar, and canons
+    whose own rows no longer agree what they are. **Read-only — you decide.**
+
+    This tool deliberately cannot fix anything. Every row it lists was written
+    unattended by the write-path hook on a similarity score, and what made
+    that harmful was not one bad score but a machine recording a permanent
+    classification with nobody reading it. Un-asserting them automatically
+    would be the same mistake with a wider blast radius. So: read the
+    evidence, judge, and record the judgment yourself with `classify_shapes`
+    — under your own name, with a reason.
+
+    `weak` — rows the hook stamped on a resemblance below the current floor
+    (`floor` in the response). The floor arrived after they did, and a guard
+    at the point of classification never undoes what is already stored. Rows
+    an agent or an audit judged are NOT listed at any age: a judgment is not
+    weak evidence, it is a different kind of evidence. Each row carries its
+    score, signature and derived form so you can judge rather than trust the
+    threshold.
+
+    `incoherent` — canons whose judged rows do not agree on a form: the
+    `forms` histogram, how many members were hook-stamped weakly, and a
+    sample. A canon asserts that some shapes are the same sort of thing; when
+    its members are a class, three getters and a dozen tests, that assertion
+    has stopped being true and every base-rate reading built on it — the
+    divergence prompt included — is reading noise.
+
+    A row listed here is a question, not a verdict. Some will be correct.
+
+    Read-only; requires read access to the project.
+    """
+    uid = current_user_id()
+    rows = await shape_ledger_svc.live_rows_for(uid, project_id)
+    return shape_ledger_svc.stamps_to_review(rows, top=top)
+
+
 async def confirm_shape_proposals(
     project_id: int,
     snippet_id: int = 0,
@@ -326,5 +362,6 @@ def register(mcp) -> None:
     for fn in (
         classify_shapes, classify_shapes_by_rule, list_shapes,
         refresh_pattern_coverage, confirm_shape_proposals, shape_history,
+        stamps_to_review,
     ):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/services/plugin_context.py

@@ -2028,7 +2028,7 @@ async def build_write_path_hint(
     if stamp_shapes and project_id:
         try:
             divergence = await shape_ledger_svc.write_time_divergence(
-                project_id, path, stamp_shapes, stamped
+                project_id, path, stamp_shapes, stamped, code or "",
             )
         except Exception:
             logger.warning("write-time divergence check failed", exc_info=True)

src/scribe/services/shape_ledger.py

@@ -439,6 +439,20 @@ async def mark_canonicals(
         await session.commit()
 
 
+async def live_rows_for(user_id: int, project_id: int) -> list[CodeShape]:
+    """`live_rows` behind the project read gate — for callers that arrive from
+    outside (an MCP tool, a route) rather than from a job that already
+    established who is asking. Empty for a project the caller cannot read,
+    never a partial answer."""
+    # Deferred, like every other access import in this module: services/access
+    # reaches back here, and a module-level import closes the cycle.
+    from scribe.services import access
+
+    if not await access.can_read_project(user_id, project_id):
+        return []
+    return await live_rows(project_id)
+
+
 async def live_rows(project_id: int) -> list[CodeShape]:
     """Every un-vanished ledger row for a project — the accounting readout's
     input, across ALL its repos (a repo unreachable this refresh still counts;
@@ -868,6 +882,223 @@ async def snippet_consumers(user_id: int, note_id: int) -> dict:
 # precision comes from the in-play test above, not from this window.
 PULL_WINDOW = timedelta(hours=6)
 
+# ── WHAT KIND OF THING a shape is, past `css | sym` (#4204) ──────────────
+#
+# THE PROBLEM THIS EXISTS FOR. `kind` has exactly two values, so a frozen
+# dataclass, a module constant, a sync predicate, a class and an async service
+# function are all siblings of one another. Nothing downstream could possibly
+# discriminate on a distinction the column does not carry — which is how a
+# directory base rate ended up standing alone as the entire divergence
+# argument. It was not chosen over a better signal; it was the only signal
+# there was. Measured consequence: writing a registry module of pure helpers
+# produced five prompts to build them from the `async_session` service canon,
+# and on another project one snippet had been recorded as the canon for a
+# `class`, a plain `def`, an `async def` and a dozen test functions at once.
+#
+# DERIVED FROM THE SIGNATURE ON READ, not stored. `kind` is part of the row
+# identity — ("project_id", "repo_key", "path", "symbol", "kind") — so
+# widening that column rewrites every row's key and needs a migration plus a
+# re-extract. Reading the form off the signature costs nothing, needs no
+# migration, and is reversible. Widening `kind` remains the principled fix and
+# this does not foreclose it: every caller below asks `shape_form`, so the day
+# the column carries the answer, this function returns it instead.
+#
+# DELIBERATELY COARSE, and it must stay that way. It answers "could these
+# plausibly be the same kind of thing", never "what language construct is
+# this". A finer taxonomy would need per-language parsing and would start
+# disagreeing with itself across the four languages this ledger already holds
+# (Python, Go, TypeScript, Vue SFC) — and a classifier that is wrong in a new
+# way is worse than the coarse one it replaced.
+FORM_UNKNOWN = ""
+
+# How much a canon's own instances must agree before it may assert a form.
+# A canon whose rows disagree about what they are cannot tell anyone else what
+# to be — and that disagreement is the SYMPTOM of the bad stamping this same
+# change fixes, so reading it as "no opinion" makes the two halves cooperate:
+# a canon poisoned by loose stamping falls silent instead of flagging.
+_FORM_SHARE = 0.6
+
+# Leading words that say nothing about the form of the thing being declared.
+_FORM_NOISE = ("export ", "default ", "public ", "private ", "static ", "final ")
+
+# A declared TYPE, across the languages in play. Coarse on purpose: a Go
+# struct, a TS interface and a Python class are one bucket because the
+# question is only ever "is the other thing also a type".
+_FORM_TYPE_WORDS = ("class ", "interface ", "type ", "struct ", "enum ")
+_FORM_FN_WORDS = ("def ", "function ", "func ", "fn ", "sub ")
+
+# THE DECLARING KEYWORD IS OPTIONAL, because Python has none. A module
+# constant is written `MAX_BUDGET = 10` — no `const`, no `let` — and a
+# pattern that required one classified every Python constant as unreadable,
+# which then read downstream as "do not assert" and quietly excluded a whole
+# form from both checks. Caught by the payload test, not by review.
+_FORM_BINDING = re.compile(r"^(?:(?:const|let|var)\s+)?[\w$]+\s*(?::[^=]+)?=")
+_FORM_ARROW = re.compile(
+    r"^(?:(?:const|let|var)\s+)?[\w$]+\s*(?::[^=]+)?=\s*(async\s*)?\("
+)
+
+
+def shape_form(signature: str, kind: str = "sym") -> str:
+    """The structural form of a shape: css / type / async-fn / fn / binding.
+
+    Returns FORM_UNKNOWN when the signature does not say, and every caller
+    treats that as "do not assert", never as "no match". An unreadable
+    signature must make this quieter, not more confident.
+    """
+    if kind == "css":
+        return "css"
+    sig = (signature or "").strip()
+    if not sig:
+        return FORM_UNKNOWN
+    changed = True
+    while changed:
+        changed = False
+        for lead in _FORM_NOISE:
+            if sig.startswith(lead):
+                sig, changed = sig[len(lead):].lstrip(), True
+    if sig.startswith(_FORM_TYPE_WORDS):
+        return "type"
+    if sig.startswith("async "):
+        return "async-fn"
+    if sig.startswith(_FORM_FN_WORDS):
+        return "fn"
+    m = _FORM_ARROW.match(sig)
+    if m:
+        return "async-fn" if m.group(1) else "fn"
+    if _FORM_BINDING.match(sig):
+        return "binding"
+    return FORM_UNKNOWN
+
+
+def forms_agree(a: str, b: str) -> bool:
+    """Do two forms match well enough to assert a relationship?
+
+    BOTH must be known. "Unknown equals unknown" would make two shapes nobody
+    can read into a confident pair, which is the failure this whole change is
+    about — a guess dressed as a finding.
+    """
+    return bool(a) and bool(b) and a == b
+
+
+def forms_conflict(a: str, b: str) -> bool:
+    """Do two KNOWN forms rule each other out?
+
+    The weaker sibling of `forms_agree`, and the pair exists because the two
+    kinds of evidence this ledger acts on deserve different burdens.
+
+    An explicit by-name reference — the payload literally names the canon's
+    symbol — is strong, so it needs only the absence of a contradiction: stamp
+    unless the forms are both readable and different. A payload-level
+    RESEMBLANCE SCORE is weak, computed against the whole file, so one score
+    speaks for every symbol in it; that needs positive agreement before
+    asserting anything, which is `forms_agree`.
+
+    Demanding agreement everywhere was the first version of this and it was
+    wrong: it silenced the by-name path whenever a shape's definition was not
+    in the payload — an Edit rather than a Write — turning strong evidence
+    into no evidence for a reason that has nothing to do with the code.
+    """
+    return bool(a) and bool(b) and a != b
+
+
+def shape_family(form: str) -> str:
+    """The coarse category a form belongs to, or "" when the form is unknown.
+
+    WHY THIS IS SEPARATE FROM `shape_form`, and it is the correction to the
+    first version of this change. The two checks in this module ask opposite
+    questions, so they cannot share a burden:
+
+      STAMPING asserts "this IS that canon". Agreement in form is evidence
+      FOR the claim, so `forms_agree` / `forms_conflict` — the precise level,
+      where `fn` and `async-fn` are different — is right.
+
+      DIVERGENCE asserts "this is NOT the canon that dominates here — did you
+      mean to?" A form MISMATCH is the premise of that prompt, not an
+      objection to it. Gating it on `forms_agree` inverted the check: it went
+      silent on exactly the mismatches it exists to catch, and the acceptance
+      case of #2793 — a hand-rolled sync `confirmDanger` in a directory where
+      an async confirm helper is canon — stopped being flagged.
+
+    So divergence gates at FAMILY level and only on contradiction. A sync
+    function beside an async one is still a fair question. A frozen dataclass
+    told to build from an async service function is not a question at all,
+    and that is the #4204 prompt this removes.
+
+    This does NOT remove every false prompt #4204 recorded. Of the five, it
+    silences `Point` (a type, against a callable canon); `_p`, `get_point`,
+    `is_registered` and `sources_expected_to_emit` are callables like the
+    canon and still ask. At the signature level they are indistinguishable
+    from the #2793 case above, so nothing readable here can separate them —
+    only widening `kind` past `css | sym` (#4204 option 2) or comparing
+    meaning rather than form can. Stated here so the next reader does not
+    assume the gap is an oversight.
+    """
+    if form in ("fn", "async-fn"):
+        return "callable"
+    if form == "type":
+        return "type"
+    if form == "binding":
+        return "value"
+    if form == "css":
+        return "css"
+    return ""
+
+
+def families_conflict(a: str, b: str) -> bool:
+    """Do two forms belong to categorically different KNOWN families?
+
+    Like `forms_conflict`, both sides must be readable: an unknown form makes
+    this quieter, never more confident.
+    """
+    fa, fb = shape_family(a), shape_family(b)
+    return bool(fa) and bool(fb) and fa != fb
+
+
+def canon_form(rows: Iterable, snippet_id: int) -> str:
+    """The form a canon's own judged rows agree on, or FORM_UNKNOWN."""
+    forms: dict[str, int] = {}
+    for r in rows:
+        if r.snippet_id != snippet_id or r.status not in ("canonical", "instance"):
+            continue
+        f = shape_form(getattr(r, "signature", "") or "", r.kind)
+        if f:
+            forms[f] = forms.get(f, 0) + 1
+    if not forms:
+        return FORM_UNKNOWN
+    top, n = max(forms.items(), key=lambda kv: kv[1])
+    return top if n / sum(forms.values()) >= _FORM_SHARE else FORM_UNKNOWN
+
+
+def signature_in(code: str, symbol: str, kind: str) -> str:
+    """The line in ``code`` that DEFINES ``symbol``, or "" if none does.
+
+    The write-time checks are asked about a shape that may have no ledger row
+    yet — it is being written right now — so the payload is the only place its
+    signature exists. Finding nothing returns "", which reads downstream as
+    FORM_UNKNOWN and therefore as silence.
+    """
+    if not code or not symbol:
+        return ""
+    name = re.escape(symbol.lstrip("."))
+    if kind == "css":
+        pat = re.compile(r"^\s*\." + name + r"\b")
+    else:
+        pat = re.compile(
+            r"^\s*(?:(?:export|default|public|private|static|final)\s+)*"
+            r"(?:"
+            r"(?:(?:async\s+)?(?:def|function|func|fn|sub)|class|interface|type"
+            r"|struct|enum|const|let|var)\s+" + name + r"\b"
+            # A bare binding — `MAX = 10`, `Handler = ...` — which is how
+            # Python (and plain JS assignment) declares one.
+            r"|" + name + r"\s*(?::[^=\n]+)?=(?!=)"
+            r")"
+        )
+    for line in code.splitlines():
+        if pat.match(line):
+            return line.strip()
+    return ""
+
+
 def snippet_kind(symbol: str, language: str) -> str:
     """The ledger kind a snippet's reference belongs to — "css" when its
     symbol is a class selector (or it is a stylesheet with no symbol),
@@ -920,6 +1151,47 @@ async def recent_pulls(user_id: int, *, window: timedelta = PULL_WINDOW) -> dict
         return {}
 
 
+# THE HOOK'S EVIDENCE, WRITTEN ONCE AND READ BACK BY ONE PARSER.
+#
+# The stamp's score used to exist only inside a prose sentence, which meant
+# nothing could ask "how strong was the evidence for this row?" without
+# re-deriving it. That is how 32 rows on one project and 334 on another sat
+# unexamined: the number was printed, never stored, never queried. Format and
+# parser live beside each other so a change to one that breaks the other is a
+# visible edit rather than a silent drift (rule 33's reasoning, one module
+# down).
+_REFERENCE_REASON = "hook: pulled #{sid}; payload references `{symbol}`"
+_RESEMBLE_REASON = "hook: pulled #{sid}; payload resembles it ({score:.2f})"
+_RESEMBLE_READ = re.compile(
+    r"^hook: pulled #(\d+); payload resembles it \(([0-9]*\.?[0-9]+)\)$"
+)
+
+
+def stamp_score(reason: str | None) -> float | None:
+    """The resemblance a hook stamp recorded, or None if it was not one.
+
+    None means "this row was not written on a similarity score" — a by-name
+    reference, an agent judgment, an audit — NOT "the score was zero". Every
+    caller treats the two differently, because a row a person judged is not
+    weak evidence, it is a different kind of evidence entirely.
+    """
+    m = _RESEMBLE_READ.match((reason or "").strip())
+    return float(m.group(2)) if m else None
+
+
+def _stamp_allowed(rank: int, mine: str, canon: str) -> bool:
+    """May this evidence assert that a shape of form ``mine`` IS ``canon``?
+
+    Named rather than inlined because the asymmetry is the decision, not an
+    implementation detail: rank 2 (the payload names the canon's symbol) has
+    only to avoid contradicting, rank 1 (a whole-file similarity score) has to
+    positively agree. See `forms_conflict` for why both exist.
+    """
+    if rank >= 2:
+        return not forms_conflict(mine, canon)
+    return forms_agree(mine, canon)
+
+
 async def stamp_write_path_instances(
     user_id: int,
     project_id: int,
@@ -961,7 +1233,15 @@ async def stamp_write_path_instances(
         return []
 
     # Which pulled canons are in play for this payload, by kind, ranked.
-    in_play: dict[str, list[tuple[int, datetime, int, str]]] = {}
+    #
+    # THE FORM OF THE CANON TRAVELS WITH IT (#4204), because what gets written
+    # here is an ASSERTION — "this shape IS that canon" — and it is permanent
+    # until something re-judges it. The evidence below is payload-level: a
+    # resemblance score is computed against the whole file, so without a
+    # per-shape test every symbol in that file inherits one verdict. Measured
+    # on another project: a single write stamped `class SessionAbsent`,
+    # `async def attach` and `_run_control_client` as instances of one snippet.
+    in_play: dict[str, list[tuple[int, datetime, int, str, str]]] = {}
     for sid, pulled_at in pulled.items():
         note = await snippets_svc.get_snippet(user_id, sid)
         if note is None:
@@ -970,12 +1250,22 @@ async def stamp_write_path_instances(
         symbol = fields.get("symbol") or ""
         kind = snippet_kind(symbol, fields.get("language") or "")
         if references_symbol(code, symbol, kind):
-            rank, why = 2, f"hook: pulled #{sid}; payload references `{_norm_symbol(symbol)}`"
+            rank, why = 2, _REFERENCE_REASON.format(
-        elif sid in resembles:
+                sid=sid, symbol=_norm_symbol(symbol)
-            rank, why = 1, f"hook: pulled #{sid}; payload resembles it ({resembles[sid]:.2f})"
+            )
+        elif resembles.get(sid, 0.0) >= _RESEMBLE_MIN:
+            rank, why = 1, _RESEMBLE_REASON.format(sid=sid, score=resembles[sid])
         else:
+            # A SCORE BELOW THE FLOOR IS NOT WEAK EVIDENCE, IT IS NONE. This
+            # branch used to be `elif sid in resembles`, which took any score
+            # at all — 0.69 counted exactly as much as 0.95, and the number was
+            # printed into the reason line while never being tested against
+            # anything. A row written on that basis is indistinguishable
+            # afterwards from one written on real evidence.
             continue
-        in_play.setdefault(kind, []).append((rank, pulled_at, sid, why))
+        in_play.setdefault(kind, []).append(
+            (rank, pulled_at, sid, why, shape_form(fields.get("signature") or "", kind))
+        )
     if not in_play:
         return []
     for bucket in in_play.values():
@@ -998,8 +1288,31 @@ async def stamp_write_path_instances(
             bucket = in_play.get(kind)
             if not bucket:
                 continue
-            _rank, _at, sid, why = bucket[0]
             row = by_key.get((name, kind))
+            # PER SHAPE, NOT PER FILE (#4204). The best candidate whose FORM
+            # matches this shape's — not simply the best candidate. A class and
+            # a function in one file can no longer be handed the same canon
+            # because the file as a whole resembled it.
+            #
+            # The signature comes from the payload first: a shape being written
+            # right now may have no ledger row yet, and then the code is the
+            # only place it exists. No readable signature on either side means
+            # no stamp — `forms_agree` requires both to be known, so an
+            # unreadable shape falls silent instead of matching everything.
+            mine = shape_form(
+                signature_in(code, name, kind) or getattr(row, "signature", "") or "", kind
+            )
+            # THE BURDEN SCALES WITH THE EVIDENCE. Rank 2 is an explicit
+            # by-name reference to the canon in this very payload; it stands
+            # unless the forms actively contradict. Rank 1 is a similarity
+            # score over the whole file — one number that would otherwise
+            # speak for every symbol in it — so it must positively agree.
+            cand = next(
+                (t for t in bucket if _stamp_allowed(t[0], mine, t[4])), None
+            )
+            if cand is None:
+                continue
+            _rank, _at, sid, why, _cform = cand
             if row is None:
                 if not repo_key:
                     continue
@@ -1021,7 +1334,12 @@ async def stamp_write_path_instances(
             # call-site fact, independent of which one the row is judged to be.
             await record_uses(
                 session, row,
-                [s_id for rank, _at, s_id, _why in bucket if rank == 2],
+                # Indexed, not destructured: this list widened from 4 to 5 when
+                # the canon's form joined it (#4204) and a positional unpack
+                # here went on reading four. The unit lane never touches
+                # `record_uses`, so it stayed green and the integration lane
+                # was the only thing that said so.
+                [t[2] for t in bucket if t[0] == 2],
                 basis="hook", evidence="write path: pulled the snippet, payload names its symbol",
             )
         if stamped:
@@ -1671,6 +1989,24 @@ async def confirm_proposals(
 
 # 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.
+# How much a payload must resemble a canon before the hook may assert, with
+# nobody watching, that a shape IS an instance of it.
+#
+# 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
+# the reason line without ever being compared to anything. Those rows are
+# permanent — they feed `dominant_canon`, which then tells the next writer in
+# that directory what to build from — so a thin stamp does not stay thin, it
+# compounds.
+#
+# 0.80 rather than the retrieval floors near 0.70, deliberately. Those bars
+# decide whether to SHOW someone a record, where being wrong costs a glance.
+# This one decides whether to RECORD a claim about the codebase unattended,
+# where being wrong costs a wrong instruction to everyone who writes in that
+# directory afterwards. An unattended write should need more evidence than a
+# suggestion, not the same.
+_RESEMBLE_MIN = 0.80
+
 _DENSITY_MIN_JUDGED = 3
 _DENSITY_SHARE = 0.6
 
@@ -1713,11 +2049,20 @@ async def canon_density(project_id: int, path: str, kind: str) -> tuple[int, int
             )
         ).scalars().all()
     siblings = [r for r in rows if _dir_of(r.path) == directory]
-    return dominant_canon(siblings)
+    dom = dominant_canon(siblings)
+    if dom is None:
+        return None
+    # The canon's form travels with the count, read from its OWN judged rows.
+    # `dominant_canon` keeps its three-value shape: it answers "what dominates
+    # here", which is still a true and separately useful question, and its
+    # test pins that arithmetic. What changed is that nobody acts on the count
+    # alone any more.
+    return dom[0], dom[1], dom[2], canon_form(siblings, dom[0])
 
 
 async def write_time_divergence(
     project_id: int, path: str, shapes: list[tuple[str, str]], stamped: list[dict],
+    code: str = "",
 ) -> list[dict]:
     """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
@@ -1745,7 +2090,7 @@ async def write_time_divergence(
         dom = density.get(kind)
         if not dom:
             continue
-        sid, n, judged = dom
+        sid, n, judged, cform = dom
         if just_stamped.get((name, kind)) == sid:
             continue
         row = by_key.get((name, kind))
@@ -1753,6 +2098,23 @@ async def write_time_divergence(
             row.status != "unclassified" or row.proposed_snippet_id == sid
         ):
             continue
+        # IS THIS EVEN THE SAME CATEGORY OF THING AS THE CANON? (#4204)
+        # Without this the line was a pure base rate: "most things here are X,
+        # so be X", with the candidate never examined at all. It told a frozen
+        # dataclass to build from the `async_session` service canon.
+        #
+        # FAMILY, not form, and only on contradiction — see `shape_family`. A
+        # divergence prompt is ABOUT a mismatch, so requiring the candidate to
+        # match would silence the check precisely where it belongs.
+        #
+        # Signature from the payload first — a shape being written now may
+        # have no row yet — and an unreadable one produces a fair question
+        # rather than a guess, because `families_conflict` needs both sides.
+        mine = shape_form(
+            signature_in(code, name, kind) or getattr(row, "signature", "") or "", kind
+        )
+        if families_conflict(mine, cform):
+            continue
         out.append({"symbol": name, "kind": kind, "canon_snippet_id": sid,
                     "instances": n, "judged": judged})
     return out
@@ -1856,6 +2218,9 @@ async def flag_divergence(project_id: int, *, since: datetime | None) -> int:
         flagged = 0
         for siblings in by_dir.values():
             dom = dominant_canon(siblings)
+            # Computed once per directory rather than per row: the canon's
+            # form is a property of the canon, not of who is being judged.
+            cform = canon_form(siblings, dom[0]) if dom else FORM_UNKNOWN
             for r in siblings:
                 if r.status not in _MECHANICAL_TODO:
                     continue
@@ -1866,12 +2231,110 @@ async def flag_divergence(project_id: int, *, since: datetime | None) -> int:
                     continue
                 if r.proposed_snippet_id == dom[0]:
                     continue  # the proposer already says "instance of the canon"
+                # The same structural test the write-time check applies
+                # (#4204). The sweep and the hook must agree about what counts
+                # as divergence, or an audit contradicts the line the writer
+                # was shown at the keyboard.
+                if families_conflict(shape_form(r.signature or "", r.kind), cform):
+                    continue
                 r.diverges_from = dom[0]
                 flagged += 1
         await session.commit()
     return flagged
 
 
+# How many of a canon's rows a review listing shows before "…" — enough to
+# judge from, not the whole table.
+_REVIEW_ROWS_SHOWN = 12
+
+
+def stamps_to_review(rows: Iterable[CodeShape], *, top: int = 10) -> dict:
+    """Judged rows whose EVIDENCE no longer meets the bar the ledger now
+    holds, and canons whose own rows no longer agree what they are.
+
+    THIS CHANGES NOTHING, AND THAT IS THE POINT. Every row here was written
+    unattended by the write-path hook on a similarity score, and the damage
+    that caused was not the score being wrong once — it was a machine
+    asserting a permanent classification with nobody reading it. A sweep that
+    silently un-asserted them would be the identical mistake with a larger
+    blast radius and a cleaner conscience. So this reports; an agent reads the
+    evidence, decides, and records the decision through `classify_shapes`
+    under its own name. The ledger should be able to say "these look wrong"
+    without being able to act on it alone.
+
+    Two lists, because they are different questions:
+
+    `weak` — rows stamped on a resemblance below `_RESEMBLE_MIN`. The floor
+    arrived after they did (#4204), and a guard at the point of classification
+    does not undo the classifications already stored (lesson #4202). Rows a
+    person or an audit judged are never listed, however old: a human judgment
+    is not weak evidence, it is a different kind of evidence.
+
+    `incoherent` — canons whose judged rows do not agree on a form. A canon is
+    a claim that some set of shapes are the same sort of thing; when its own
+    members are a class, three getters and a dozen tests, that claim has
+    stopped being true, and every base-rate reading built on it is reading
+    noise. `canon_form` already makes such a canon fall silent — this is what
+    makes it VISIBLE, which is the half that was missing.
+    """
+    live = [r for r in rows if r.vanished_at is None]
+    weak = []
+    for r in live:
+        if r.status not in _NEEDS_TARGET or r.classified_by != "hook":
+            continue
+        score = stamp_score(r.reason)
+        if score is None or score >= _RESEMBLE_MIN:
+            continue
+        weak.append({
+            "path": r.path, "symbol": r.symbol, "kind": r.kind,
+            "status": r.status, "snippet_id": r.snippet_id,
+            "score": score, "signature": r.signature or "",
+            "form": shape_form(r.signature or "", r.kind),
+            "classified_at": r.classified_at.isoformat() if r.classified_at else None,
+        })
+    weak.sort(key=lambda d: (d["score"], d["path"], d["symbol"]))
+
+    by_canon: dict[int, list[CodeShape]] = {}
+    for r in live:
+        if r.status in _NEEDS_TARGET and r.snippet_id:
+            by_canon.setdefault(int(r.snippet_id), []).append(r)
+    incoherent = []
+    for sid, members in by_canon.items():
+        forms: dict[str, int] = {}
+        for r in members:
+            f = shape_form(r.signature or "", r.kind)
+            if f:
+                forms[f] = forms.get(f, 0) + 1
+        readable = sum(forms.values())
+        if readable < _DENSITY_MIN_JUDGED:
+            continue   # too few to say anything either way
+        if canon_form(members, sid):
+            continue   # a form holds the majority: coherent
+        incoherent.append({
+            "snippet_id": sid,
+            "judged": len(members),
+            "forms": dict(sorted(forms.items(), key=lambda kv: -kv[1])),
+            "weak_rows": sum(
+                1 for r in members
+                if r.classified_by == "hook"
+                and (stamp_score(r.reason) or 1.0) < _RESEMBLE_MIN
+            ),
+            "sample": [
+                {"path": r.path, "symbol": r.symbol,
+                 "signature": r.signature or "", "by": r.classified_by}
+                for r in members[:_REVIEW_ROWS_SHOWN]
+            ],
+        })
+    incoherent.sort(key=lambda d: (-d["weak_rows"], -d["judged"]))
+    return {
+        "weak_count": len(weak),
+        "weak": weak[:top],
+        "incoherent_count": len(incoherent),
+        "incoherent": incoherent[:top],
+        "floor": _RESEMBLE_MIN,
+    }
+
+
 def divergence_summary(rows: Iterable[CodeShape], *, top: int = 10) -> dict:
     """Readout view: flagged shapes (newest first) and the recheck count."""
     flagged = [r for r in rows if r.diverges_from is not None and r.status in _MECHANICAL_TODO]

tests/test_guidance_ownership.py

@@ -212,6 +212,19 @@ TOPICS: tuple[Topic, ...] = (
           "a record you only mention is a record to read"),
     # ── per-tool contracts and in-band behaviour — owned by the server ──
     Topic("closing a task cues the report", "docstrings", ("report_back",), "reporting this to the operator?"),
+    # The agent is the judge (#4208). Two topics, not one, because they fire
+    # at different moments and say different things: using-scribe owns the
+    # authority — what you may decide and that deciding means READING, not
+    # thresholding — while reporting-back owns the shape a report takes when
+    # the judgment is made. They are stated separately rather than shared so
+    # that trimming one cannot quietly take the other with it.
+    Topic("the agent judges the record, and judging is attended", U,
+          ("judge of record", "attended", "classify_shapes"),
+          "a record that is wrong stays wrong until something rewrites it",
+          index=("judge of record",)),
+    Topic("a finding surfaced and not judged is a finding dropped",
+          "skill:reporting-back", ("you are the judge", "hard to reverse"),
+          "it reads as diligence and functions as a backlog"),
     Topic("a note that asserts a fact carries its check", "docstrings", ("verify_with", "expires_when"),
           "could this note become false without anyone editing it",
           # Stated at two different moments on purpose: the tool contract is

tests/test_shape_form_gate.py

@@ -0,0 +1,325 @@
+"""A canon may only be urged on a shape that could plausibly BE it (#4204).
+
+WHAT WAS WRONG. `dominant_canon` is a base rate: it answers "what is most
+common in this directory" and never "is this that". With `kind` carrying only
+`css | sym`, a frozen dataclass, a module constant, a sync predicate, a class
+and an async service function were all siblings, so the prior was the entire
+argument — it told a registry module of pure helpers to build from the
+`async_session` service canon. Upstream, the auto-stamp had no resemblance
+floor at all (`elif sid in resembles`), so 0.69 asserted as confidently as
+0.95, and one payload-level score was applied to every symbol in the file.
+
+The two halves compound: loose stamping manufactures the density that the
+divergence check then reads as authority. Both are fixed by the same
+primitive, and it is tested here on REAL signatures taken from the ledger —
+Python, Go, TypeScript and Vue — rather than invented ones, because a
+classifier that only works on the examples its author imagined is the failure
+this is meant to end.
+
+The unknown case is tested hardest. `forms_agree` requires BOTH sides to be
+known, so an unreadable signature makes the checks quieter rather than more
+confident. A guard that treats "I cannot tell" as "match" is the shape of the
+bug, not the fix.
+"""
+from __future__ import annotations
+
+import pytest
+
+from scribe.services.shape_ledger import (
+    FORM_UNKNOWN, canon_form, families_conflict, forms_agree, shape_family,
+    shape_form, signature_in,
+)
+
+
+class _Row:
+    """A ledger row, as `canon_form` reads one."""
+
+    def __init__(self, signature, snippet_id=7, status="instance", kind="sym"):
+        self.signature, self.snippet_id = signature, snippet_id
+        self.status, self.kind = status, kind
+
+
+# ── shape_form, on signatures actually in the ledger ──────────────────────
+
+@pytest.mark.parametrize("signature,want", [
+    # Python
+    ("class SessionAbsent(RuntimeError):", "type"),
+    ("def build_channel() -> str:", "fn"),
+    ("async def attach(self) -> None:", "async-fn"),
+    ("async def _make_room(self, user_id: int, credential_id: int, opening) -> bool:",
+     "async-fn"),
+    ("def test_version_regex_rejects_bad_formats():", "fn"),
+    # Go
+    ("func NewService(pool *pgxpool.Pool, cfg *lidarrconfig.Service) *Service {", "fn"),
+    ("type Store struct {", "type"),
+    # TypeScript / Vue SFC
+    ("export function useBuild() {", "fn"),
+    ("function offsetWords(minutes: number): string {", "fn"),
+    ("export default class Thing {", "type"),
+    ("interface Lesson {", "type"),
+    ("const handler = async (req) => {", "async-fn"),
+    ("const offset = (n) => n + 1", "fn"),
+    ("const MAX_BUDGET = 10", "binding"),
+    # Unreadable
+    ("", FORM_UNKNOWN),
+    ("   ", FORM_UNKNOWN),
+    ("# just a comment", FORM_UNKNOWN),
+    ("}", FORM_UNKNOWN),
+])
+def test_shape_form_reads_real_signatures(signature, want) -> None:
+    assert shape_form(signature, "sym") == want
+
+
+def test_css_is_its_own_form_regardless_of_signature() -> None:
+    assert shape_form(".pin {", "css") == "css"
+    assert shape_form("", "css") == "css"
+
+
+def test_the_five_shapes_that_started_this_are_not_service_functions() -> None:
+    """The concrete case. #2860 is an `async def` service unit; none of these
+    is one, and all five were told to build from it."""
+    canon = shape_form("async def create_note(user_id: int, ...):", "sym")
+    assert canon == "async-fn"
+    for sig in (
+        "class Point:",
+        "def _p(source, kind, what, **kw) -> tuple[str, Point]:",
+        "def get_point(source: str) -> Point | None:",
+        "def is_registered(source: str) -> bool:",
+        "def sources_expected_to_emit() -> list[str]:",
+    ):
+        assert not forms_agree(shape_form(sig, "sym"), canon), sig
+
+
+def test_how_many_of_the_five_the_divergence_gate_actually_silences() -> None:
+    """HONEST ACCOUNTING, and it is deliberately an assertion rather than a
+    comment so it fails the day the answer changes.
+
+    Divergence gates at FAMILY level and only on contradiction, because a form
+    mismatch is the PREMISE of a divergence prompt, not an objection to it —
+    see `shape_family`. So of #4204's five false prompts this silences one:
+    `Point` is a type and the canon is a callable. The four `def` helpers are
+    callables like the canon and still ask.
+
+    That is not a shortcoming of the gate, it is the limit of the signature:
+    at this level those four are indistinguishable from #2793's acceptance
+    case, where a sync `confirmDanger` beside an async confirm helper SHOULD
+    be flagged. Separating them needs #4204 option 2 (widen `kind`) or a
+    comparison of meaning."""
+    canon = shape_form("async def create_note(user_id: int, ...):", "sym")
+    silenced = {
+        sig: families_conflict(shape_form(sig, "sym"), canon)
+        for sig in (
+            "class Point:",
+            "def _p(source, kind, what, **kw) -> tuple[str, Point]:",
+            "def get_point(source: str) -> Point | None:",
+            "def is_registered(source: str) -> bool:",
+            "def sources_expected_to_emit() -> list[str]:",
+        )
+    }
+    assert silenced["class Point:"] is True
+    assert sum(silenced.values()) == 1
+
+
+def test_a_sync_function_beside_an_async_canon_is_still_a_fair_question() -> None:
+    """#2793's acceptance case, which the first version of this change broke:
+    gating divergence on `forms_agree` silenced a hand-rolled sync confirm in
+    a directory where an async confirm helper is canon — the exact prompt the
+    milestone exists to produce."""
+    canon = shape_form("async function onTrash() {", "sym")
+    mine = shape_form("function confirmDanger() {", "sym")
+    assert canon == "async-fn" and mine == "fn"
+    assert not forms_agree(mine, canon)      # the STAMP gate would refuse
+    assert not families_conflict(mine, canon)  # the DIVERGENCE gate asks anyway
+
+
+@pytest.mark.parametrize(("form", "want"), [
+    ("fn", "callable"), ("async-fn", "callable"), ("type", "type"),
+    ("binding", "value"), ("css", "css"), (FORM_UNKNOWN, ""),
+])
+def test_every_form_has_a_family_and_unknown_has_none(form, want) -> None:
+    assert shape_family(form) == want
+
+
+@pytest.mark.parametrize(("a", "b"), [
+    (FORM_UNKNOWN, "fn"), ("fn", FORM_UNKNOWN), (FORM_UNKNOWN, FORM_UNKNOWN),
+])
+def test_an_unreadable_form_never_silences_a_divergence_prompt(a, b) -> None:
+    """Same direction as everywhere else in this module: not knowing makes the
+    check quieter about ASSERTING and never quieter about ASKING."""
+    assert not families_conflict(a, b)
+
+
+# ── forms_agree: unknown never matches ────────────────────────────────────
+
+def test_two_known_equal_forms_agree() -> None:
+    assert forms_agree("fn", "fn")
+
+
+def test_different_forms_do_not() -> None:
+    assert not forms_agree("fn", "async-fn")
+    assert not forms_agree("type", "fn")
+
+
+@pytest.mark.parametrize("a,b", [
+    (FORM_UNKNOWN, "fn"), ("fn", FORM_UNKNOWN), (FORM_UNKNOWN, FORM_UNKNOWN),
+])
+def test_unknown_never_agrees_with_anything(a, b) -> None:
+    """Including with itself — "I cannot tell" twice is not a match."""
+    assert not forms_agree(a, b)
+
+
+# ── canon_form: a canon whose own rows disagree has no opinion ────────────
+
+def test_a_canon_whose_instances_agree_asserts_that_form() -> None:
+    rows = [_Row("async def a():"), _Row("async def b():"), _Row("async def c():")]
+    assert canon_form(rows, 7) == "async-fn"
+
+
+def test_a_canon_whose_instances_disagree_asserts_nothing() -> None:
+    """Portal's #3283 exactly: one snippet recorded as canon for a class, a
+    plain def, an async def and a dozen tests. That spread is the SYMPTOM of
+    loose stamping, so reading it as "no opinion" makes the two fixes
+    cooperate — a poisoned canon falls silent instead of flagging others."""
+    rows = [_Row("class A:"), _Row("def b():"), _Row("async def c():")]
+    assert canon_form(rows, 7) == FORM_UNKNOWN
+
+
+def test_a_clear_majority_still_asserts() -> None:
+    rows = [_Row("def a():"), _Row("def b():"), _Row("def c():"), _Row("class D:")]
+    assert canon_form(rows, 7) == "fn"
+
+
+def test_rows_of_other_canons_and_unjudged_rows_do_not_vote() -> None:
+    rows = [
+        _Row("def a():"), _Row("def b():"), _Row("def c():"),
+        _Row("class X:", snippet_id=9),                    # another canon
+        _Row("class Y:", status="unclassified"),           # not judged
+        _Row("class Z:", status="variant"),                # a departure, not a vote
+    ]
+    assert canon_form(rows, 7) == "fn"
+
+
+def test_a_canon_with_no_readable_signatures_asserts_nothing() -> None:
+    assert canon_form([_Row(""), _Row(""), _Row("")], 7) == FORM_UNKNOWN
+
+
+def test_a_canon_with_no_rows_at_all_asserts_nothing() -> None:
+    assert canon_form([], 7) == FORM_UNKNOWN
+
+
+# ── signature_in: the payload is where a brand-new shape lives ────────────
+
+CODE = '''
+import re
+
+MAX = 10
+
+class Point:
+    """A point."""
+
+async def fetch(user_id: int) -> None:
+    ...
+
+def helper(x):
+    return x
+
+export function useBuild() {
+'''
+
+
+@pytest.mark.parametrize("symbol,want_form", [
+    ("Point", "type"),
+    ("fetch", "async-fn"),
+    ("helper", "fn"),
+    ("useBuild", "fn"),
+    ("MAX", "binding"),
+])
+def test_a_definition_in_the_payload_is_found(symbol, want_form) -> None:
+    assert shape_form(signature_in(CODE, symbol, "sym"), "sym") == want_form
+
+
+def test_a_symbol_that_is_only_CALLED_is_not_a_definition() -> None:
+    """The conflation worth keeping out: referencing a name is evidence of
+    USE, never of being that shape."""
+    assert signature_in("value = helper(3)\nreturn fetch(x)", "helper", "sym") == ""
+
+
+def test_a_missing_symbol_yields_nothing_rather_than_a_guess() -> None:
+    assert signature_in(CODE, "nowhere", "sym") == ""
+    assert shape_form(signature_in(CODE, "nowhere", "sym"), "sym") == FORM_UNKNOWN
+
+
+def test_an_empty_payload_yields_nothing() -> None:
+    assert signature_in("", "Point", "sym") == ""
+
+
+def test_a_css_class_is_found_by_its_selector() -> None:
+    assert signature_in(".pin { color: red; }", "pin", "css").startswith(".pin")
+    assert signature_in(".pin { }", ".pin", "css").startswith(".pin")
+
+
+def test_a_near_name_is_not_matched() -> None:
+    """Word-bounded, so `helper` never claims `helper_two`."""
+    assert signature_in("def helper_two(x):\n    ...", "helper", "sym") == ""
+
+
+# ── The burden scales with the evidence ───────────────────────────────────
+#
+# Two kinds of evidence reach the stamp, and treating them alike was the
+# first version of this fix and was wrong in both directions. An explicit
+# by-name reference — the payload literally names the canon's symbol — is
+# strong; demanding positive form agreement there silenced it whenever the
+# shape's definition was not in the payload (an Edit rather than a Write),
+# turning strong evidence into none for a reason unrelated to the code.
+# A whole-file resemblance SCORE is weak: one number speaks for every symbol
+# in the file, which is precisely how a class, an async method and a private
+# helper were all recorded as instances of one snippet.
+
+from scribe.services.shape_ledger import _stamp_allowed, forms_conflict  # noqa: E402
+
+
+@pytest.mark.parametrize("a,b,want", [
+    ("fn", "type", True),
+    ("async-fn", "fn", True),
+    ("fn", "fn", False),
+    (FORM_UNKNOWN, "fn", False),      # cannot contradict what you cannot read
+    ("fn", FORM_UNKNOWN, False),
+    (FORM_UNKNOWN, FORM_UNKNOWN, False),
+])
+def test_forms_conflict_needs_both_sides_readable(a, b, want) -> None:
+    assert forms_conflict(a, b) is want
+
+
+def test_a_named_reference_stands_unless_the_forms_contradict() -> None:
+    assert _stamp_allowed(2, "fn", "fn")
+    assert _stamp_allowed(2, FORM_UNKNOWN, "fn")      # unreadable shape, strong evidence
+    assert _stamp_allowed(2, "fn", FORM_UNKNOWN)
+    assert not _stamp_allowed(2, "type", "fn")        # a class is not that function
+
+
+def test_a_resemblance_score_must_positively_agree() -> None:
+    """The weak path, and the one that produced the mess being fixed."""
+    assert _stamp_allowed(1, "fn", "fn")
+    assert not _stamp_allowed(1, FORM_UNKNOWN, "fn")
+    assert not _stamp_allowed(1, "fn", FORM_UNKNOWN)
+    assert not _stamp_allowed(1, "type", "fn")
+
+
+def test_the_portal_case_no_longer_stamps_one_canon_onto_a_whole_file() -> None:
+    """A class, an async method and a sync helper in one file, against one
+    payload-level score. Previously all three were stamped; now the score
+    alone cannot carry any of them, and only a matching form could."""
+    canon = "async-fn"
+    written = [("class SessionAbsent(RuntimeError):", False),
+               ("async def attach(self) -> None:", True),
+               ("def _run_control_client(self, command: str) -> None:", False)]
+    for signature, should_stamp in written:
+        assert _stamp_allowed(1, shape_form(signature, "sym"), canon) is should_stamp
+
+
+def test_the_resemblance_floor_is_above_the_retrieval_floors() -> None:
+    """An unattended WRITE about the codebase should need more evidence than
+    a suggestion shown to a reader — the retrieval bars sit near 0.70."""
+    from scribe.services.shape_ledger import _RESEMBLE_MIN
+
+    assert _RESEMBLE_MIN >= 0.80

tests/test_stamps_to_review.py

@@ -0,0 +1,199 @@
+"""The review surface: rows whose evidence no longer meets the bar, and
+canons whose own rows no longer agree what they are (#4204, #4208).
+
+WHAT THESE GUARD, and it is a property rather than an output shape. The
+ledger was poisoned by a machine writing permanent classifications that
+nobody read — 32 rows on one project, 334 on another. The correction is NOT
+a machine that un-writes them; that is the same mistake pointed the other
+way. It is a reader that puts the evidence in front of a judge.
+
+So the tests below assert two things hardest:
+
+  * a judgment is NEVER listed as weak, however old and whatever its score
+    would have been — an agent's decision is a different kind of evidence,
+    not a worse one;
+  * the listing changes nothing, which is checked by giving it rows and
+    asserting the rows come back untouched.
+
+The round-trip test exists because the score used to live only inside a
+prose sentence. Writer and parser are now one constant apart, and a change
+to one that the other does not follow is exactly the kind of silent rot the
+hook's own history is made of.
+"""
+
+
+import pytest
+
+from scribe.services.shape_ledger import (
+    _RESEMBLE_MIN, _RESEMBLE_REASON, stamp_score, stamps_to_review,
+)
+
+
+class _Row:
+    """A ledger row, as the review reader touches one."""
+
+    def __init__(self, *, path="src/a.py", symbol="f", kind="sym",
+                 status="instance", snippet_id=1, reason=None,
+                 classified_by="hook", signature="async def f():",
+                 vanished_at=None, classified_at=None):
+        self.path, self.symbol, self.kind = path, symbol, kind
+        self.status, self.snippet_id, self.reason = status, snippet_id, reason
+        self.classified_by, self.signature = classified_by, signature
+        self.vanished_at, self.classified_at = vanished_at, classified_at
+
+
+def _stamped(score, **kw):
+    return _Row(reason=_RESEMBLE_REASON.format(sid=1, score=score), **kw)
+
+
+# ── the writer and the parser are one constant apart ──────────────────────
+
+@pytest.mark.parametrize("score", [0.0, 0.5, 0.69, 0.77, 0.8, 0.95, 1.0])
+def test_every_score_the_hook_can_write_reads_back(score) -> None:
+    written = _RESEMBLE_REASON.format(sid=4204, score=score)
+    assert stamp_score(written) == pytest.approx(round(score, 2))
+
+
+def test_the_real_reasons_from_the_poisoned_ledgers_parse() -> None:
+    """Verbatim from Portal and Scribe rows, so the parser is pinned to text
+    that actually exists in the database rather than to text it generates."""
+    assert stamp_score("hook: pulled #3283; payload resembles it (0.69)") == 0.69
+    assert stamp_score("hook: pulled #3283; payload resembles it (0.77)") == 0.77
+    assert stamp_score("hook: pulled #2860; payload resembles it (0.68)") == 0.68
+
+
+@pytest.mark.parametrize("reason", [
+    None, "", "   ",
+    "hook: pulled #3461; payload references `icon-btn`",   # the OTHER evidence
+    "an agent wrote prose here",
+    "hook: pulled #1; payload resembles it",               # no score
+])
+def test_anything_that_is_not_a_score_reads_as_no_score(reason) -> None:
+    """None means "not written on a score", never "scored zero" — the two
+    must not collapse, because one is a judgment and the other is weak."""
+    assert stamp_score(reason) is None
+
+
+# ── weak: only hook stamps, only below the floor ──────────────────────────
+
+def test_a_stamp_below_the_floor_is_listed_with_its_evidence() -> None:
+    out = stamps_to_review([_stamped(0.69)])
+    assert out["weak_count"] == 1
+    row = out["weak"][0]
+    assert row["score"] == 0.69
+    assert row["form"] == "async-fn"
+    assert row["signature"] == "async def f():"
+    assert out["floor"] == _RESEMBLE_MIN
+
+
+def test_a_stamp_at_or_above_the_floor_is_not_listed() -> None:
+    assert stamps_to_review([_stamped(_RESEMBLE_MIN)])["weak_count"] == 0
+    assert stamps_to_review([_stamped(0.95)])["weak_count"] == 0
+
+
+@pytest.mark.parametrize("by", ["agent", "audit", "import", "mechanical"])
+def test_a_judgment_is_never_weak_however_it_would_have_scored(by) -> None:
+    """The measured case: 302 of Scribe's 334 rows under one canon were
+    `audit`-classified with no score at all. Listing those as weak would
+    invite an agent to withdraw the only judgments in the ledger that were
+    ever actually made by one."""
+    rows = [_Row(classified_by=by, reason=None),
+            _Row(classified_by=by,
+                 reason=_RESEMBLE_REASON.format(sid=1, score=0.10))]
+    assert stamps_to_review(rows)["weak_count"] == 0
+
+
+def test_an_unclassified_or_vanished_row_is_not_listed() -> None:
+    assert stamps_to_review([_stamped(0.1, status="unclassified")])["weak_count"] == 0
+    assert stamps_to_review([_stamped(0.1, vanished_at="gone")])["weak_count"] == 0
+
+
+def test_the_weakest_evidence_is_listed_first() -> None:
+    out = stamps_to_review([
+        _stamped(0.77, symbol="c"), _stamped(0.68, symbol="a"),
+        _stamped(0.71, symbol="b"),
+    ])
+    assert [r["symbol"] for r in out["weak"]] == ["a", "b", "c"]
+
+
+def test_top_bounds_the_listing_but_not_the_count() -> None:
+    out = stamps_to_review([_stamped(0.7, symbol=f"s{i}") for i in range(9)], top=3)
+    assert out["weak_count"] == 9 and len(out["weak"]) == 3
+
+
+# ── incoherent: a canon whose own members disagree ────────────────────────
+
+def _member(sig, sid=7, **kw):
+    return _Row(snippet_id=sid, signature=sig, classified_by="audit", **kw)
+
+
+def test_a_canon_whose_members_agree_is_not_listed() -> None:
+    rows = [_member("async def a():"), _member("async def b():"),
+            _member("async def c():"), _member("async def d():")]
+    assert stamps_to_review(rows)["incoherent_count"] == 0
+
+
+def test_a_canon_whose_members_are_all_different_things_is_listed() -> None:
+    """Portal's #3283, in miniature: a class, a getter, a test and a binding
+    recorded as instances of one pattern."""
+    rows = [_member("class SessionAbsent(RuntimeError):"),
+            _member("def build_channel() -> str:"),
+            _member("async def attach(self) -> None:"),
+            _member("MAX = 10")]
+    out = stamps_to_review(rows)
+    assert out["incoherent_count"] == 1
+    entry = out["incoherent"][0]
+    assert entry["snippet_id"] == 7 and entry["judged"] == 4
+    assert set(entry["forms"]) == {"type", "fn", "async-fn", "binding"}
+    assert len(entry["sample"]) == 4
+
+
+def test_too_few_readable_members_says_nothing_either_way() -> None:
+    """Two rows that disagree is not an incoherent canon, it is two rows.
+    The floor is the same one the density check uses."""
+    rows = [_member("class A:"), _member("def b():")]
+    assert stamps_to_review(rows)["incoherent_count"] == 0
+
+
+def test_unreadable_signatures_do_not_manufacture_incoherence() -> None:
+    rows = [_member(""), _member("   "), _member("???"), _member("")]
+    assert stamps_to_review(rows)["incoherent_count"] == 0
+
+
+def test_a_canon_carrying_weak_stamps_is_ranked_above_one_that_is_not() -> None:
+    clean = [_member("class A:", sid=1), _member("def b():", sid=1),
+             _member("MAX = 1", sid=1), _member("async def c():", sid=1)]
+    dirty = [_Row(snippet_id=2, signature=s, classified_by="hook",
+                  reason=_RESEMBLE_REASON.format(sid=2, score=0.70))
+             for s in ("class D:", "def e():", "MAX = 2", "async def f():")]
+    out = stamps_to_review(clean + dirty)
+    assert out["incoherent_count"] == 2
+    assert out["incoherent"][0]["snippet_id"] == 2
+    assert out["incoherent"][0]["weak_rows"] == 4
+    assert out["incoherent"][1]["weak_rows"] == 0
+
+
+# ── the property that matters most: it is a reader ────────────────────────
+
+def test_the_review_changes_nothing_about_the_rows_it_reads() -> None:
+    """Asserted rather than assumed. The whole design claim of this surface
+    is that it reports and never acts; a future refactor that "helpfully"
+    resets a status here would pass every other test in this file."""
+    rows = [_stamped(0.69), _member("class A:"), _member("def b():"),
+            _member("MAX = 1"), _member("async def c():")]
+    before = [(r.status, r.snippet_id, r.classified_by, r.reason) for r in rows]
+    stamps_to_review(rows)
+    after = [(r.status, r.snippet_id, r.classified_by, r.reason) for r in rows]
+    assert before == after
+
+
+def test_the_service_carries_no_machinery_for_bulk_withdrawal() -> None:
+    """A structural guard (rule 167) on the design decision, not on output.
+    If someone adds an auto-retire path to the ledger, this fails and the
+    conversation about whether a machine may un-judge in bulk happens again
+    — deliberately, rather than in a diff nobody read."""
+    import scribe.services.shape_ledger as ledger
+
+    body = open(ledger.__file__).read()
+    for banned in ("def retire_weak", "def auto_unclassify", "def bulk_withdraw"):
+        assert banned not in body, banned

tests/test_write_path_trigger.py

@@ -1638,8 +1638,11 @@ async def test_the_write_time_divergence_check_is_named_in_band():
             1, "frontend/src/components/Danger.vue", code=REAL_CODE, project_id=24,
             stamp_shapes=[("sym", "confirmDanger")],
         )
+    # The payload travels with the call now (#4204): the check compares the
+    # shape being written against the canon's form, and a shape written right
+    # now may exist nowhere but in this code.
     check.assert_awaited_once_with(24, "frontend/src/components/Danger.vue",
-                                   [("sym", "confirmDanger")], [])
+                                   [("sym", "confirmDanger")], [], REAL_CODE)
     assert out["divergence"] == div
     assert "Divergence check at `frontend/src/components/Danger.vue`" in out["context"]
     assert "`confirmDanger` → #2761 (20 of 21 judged siblings are its instances)" in out["context"]