Merge pull request 'dev → main: rule overlap check, design-guidance write arm, usage chip seam, divergence meaning gate' (#180) from dev into main

frontend/src/components/rules/RuleListPane.vue

@@ -1,16 +1,7 @@
 <script setup lang="ts">
 import type { RuleHeader } from "@/api/rulebooks";
 import UsageBadge from "@/components/UsageBadge.vue";
-
+import { DEAD_WEIGHT_ADVICE } from "@/utils/deadWeight";
-/** The dead-weight nudge for a RULE — two remedies, not one, which is the
- *  whole reason this advice is per-kind. A snippet nobody opens should
- *  probably go. A rule nobody opens may be perfectly good and simply firing on
- *  the wrong thing, so "delete it" would be the wrong nudge half the time and
- *  the operator has to be the one who picks. */
-const RULE_DEAD_WEIGHT =
-  "Kept arriving without being read. Either its trigger fires on the wrong " +
-  "work — reword “when to apply” so it says when — or it is not wanted here. " +
-  "Until one or the other, it takes a slot in every write it matches.";
 
 defineProps<{ topicId: number; rules: RuleHeader[] }>();
 const emit = defineEmits<{
@@ -54,7 +45,7 @@ const emit = defineEmits<{
               ? 'Asserts a fact nobody has confirmed yet'
               : `Check last passed ${r.last_verified}`"
           >{{ r.last_verified === "never" ? "unverified" : `checked ${r.last_verified}` }}</span>
-          <UsageBadge :usage="r.usage" :dead-weight-advice="RULE_DEAD_WEIGHT" />
+          <UsageBadge :usage="r.usage" :dead-weight-advice="DEAD_WEIGHT_ADVICE.rule" />
         </div>
         <div class="statement">{{ r.statement }}</div>
         <div v-if="r.when_to_apply || r.updated_at" class="meta">

frontend/src/utils/deadWeight.ts

@@ -0,0 +1,68 @@
+/**
+ * What to suggest when a record looks like dead weight — one sentence per kind.
+ *
+ * `UsageBadge` deliberately takes the advice as a prop rather than deriving it,
+ * because the COUNTS read identically for every kind and the REMEDY does not: a
+ * snippet nobody opens should probably go, while a rule or a lesson in the same
+ * position more often has a trigger that fires on the wrong work. Telling an
+ * operator to delete one of those would be the wrong nudge about half the time.
+ *
+ * The table lives here because the copy was about to exist in five places.
+ * Three already had their own: `SnippetListView`, `RuleListPane`, and
+ * `LessonDetailView` with the sentence inline in its template. The unified
+ * Knowledge browse renders notes, tasks, processes, snippets and lessons in one
+ * mixed feed (#4230), so it needs all of them at once — and a per-view constant
+ * is how three surfaces end up giving three different answers to the same
+ * question.
+ *
+ * Every sentence names the COST of leaving it, not just the fact. "Never
+ * opened" is an observation; "takes a slot in every future menu" is why the
+ * reader should care, and it is the half that makes the chip actionable.
+ */
+
+/** The record kinds the Knowledge feed can show, plus the ones only their own
+ *  views show. Keyed by `note_type`, with `rule` alongside — rules live in a
+ *  separate table but answer the same question (see `RecordUsage`). */
+export type DeadWeightKind =
+  | "note"
+  | "task"
+  | "process"
+  | "snippet"
+  | "lesson"
+  | "rule";
+
+export const DEAD_WEIGHT_ADVICE: Record<DeadWeightKind, string> = {
+  snippet:
+    "Offered repeatedly without ever being opened — consider rewriting its " +
+    "“when to reach for it” so it says when, or deleting it. It takes a slot " +
+    "in every future auto-inject menu.",
+  rule:
+    "Kept arriving without being read. Either its trigger fires on the wrong " +
+    "work — reword “when to apply” so it says when — or it is not wanted here. " +
+    "Until one or the other, it takes a slot in every write it matches.",
+  lesson:
+    "Repeatedly offered and never opened usually means the trigger fires on " +
+    "the wrong situation — re-key `when_to_apply` rather than deleting the " +
+    "claim.",
+  note:
+    "Surfaced again and again and never opened. Usually the title is the " +
+    "problem: recall matches on it first, so a note titled for its author " +
+    "rather than for the situation keeps winning slots it cannot pay for.",
+  process:
+    "Offered without ever being run. Either the steps no longer match how " +
+    "the work is actually done, or it is being matched on the wrong trigger " +
+    "— check which before retiring it.",
+  task:
+    "Surfaced repeatedly and never opened. On a task this more often says " +
+    "the work has gone stale than that the record is wrong — decide whether " +
+    "it is still wanted before re-titling it.",
+};
+
+/** The advice for a kind, falling back to the note wording.
+ *
+ * The fallback is deliberate rather than an empty string: a kind added to the
+ * feed later should still get a usable sentence, and "the title is doing the
+ * matching" is the reading that holds for any record recall can choose. */
+export function deadWeightAdvice(kind: string | null | undefined): string {
+  return DEAD_WEIGHT_ADVICE[kind as DeadWeightKind] ?? DEAD_WEIGHT_ADVICE.note;
+}

frontend/src/views/KnowledgeView.vue

@@ -3,7 +3,10 @@ import { ref, computed, watch, onMounted, onUnmounted, nextTick } from "vue";
 import { useRouter } from "vue-router";
 import { apiGet } from "@/api/client";
 import type { TaskKind, TaskStatus, TaskPriority } from "@/types/note";
+import type { RecordUsage } from "@/types/usage";
+import { deadWeightAdvice } from "@/utils/deadWeight";
 import KindBadge from "@/components/KindBadge.vue";
+import UsageBadge from "@/components/UsageBadge.vue";
 import NoteSweepPane from "@/components/NoteSweepPane.vue";
 import StatusBadge from "@/components/StatusBadge.vue";
 import PriorityBadge from "@/components/PriorityBadge.vue";
@@ -44,6 +47,10 @@ interface KnowledgeItem {
   project_id: number | null;
   created_at: string;
   updated_at: string;
+  /** Surfaced-vs-opened counts, zero-filled by the route for EVERY row, so
+   *  "never surfaced" is a value here rather than a missing field (#4230).
+   *  `UsageBadge` renders nothing at all below one surfacing. */
+  usage?: RecordUsage | null;
   // Set only when another user owns this record — their suggestion, not one of
   // yours. Absent means it's yours.
   shared?: boolean;
@@ -659,6 +666,16 @@ onUnmounted(() => {
                 class="shared-tag"
                 :title="`Shared by ${item.owner ?? 'another user'} — their record, not yours`"
               >by {{ item.owner ?? "another user" }}</span>
+              <!-- Surfaced-vs-opened (#4230). This is the only list in the UI
+                   that browses notes and lessons, so it is the only place
+                   those two kinds can show the counter at all. The advice is
+                   looked up per row because this feed is mixed — the remedy
+                   for a snippet nobody opens is not the remedy for a lesson. -->
+              <UsageBadge
+                :usage="item.usage"
+                :noun="item.note_type"
+                :dead-weight-advice="deadWeightAdvice(item.note_type)"
+              />
               <span class="k-card-date">{{ formatDate(item.updated_at) }}</span>
             </div>
           </div>

frontend/src/views/LessonDetailView.vue

@@ -25,6 +25,7 @@ import { deleteLesson, getLesson, type Lesson } from "@/api/lessons";
 import ConfirmDialog from "@/components/ConfirmDialog.vue";
 import TagPill from "@/components/TagPill.vue";
 import UsageBadge from "@/components/UsageBadge.vue";
+import { DEAD_WEIGHT_ADVICE } from "@/utils/deadWeight";
 import { useToastStore } from "@/stores/toast";
 import { renderMarkdown } from "@/utils/markdown";
 
@@ -110,7 +111,7 @@ onMounted(load);
         <UsageBadge
           :usage="lesson.usage"
           noun="lesson"
-          dead-weight-advice="Repeatedly offered and never opened usually means the trigger fires on the wrong situation — re-key `when_to_apply` rather than deleting the claim."
+          :dead-weight-advice="DEAD_WEIGHT_ADVICE.lesson"
         />
       </div>
 

frontend/src/views/SnippetListView.vue

@@ -10,6 +10,7 @@ import {
 } from "@/api/snippets";
 import { useToastStore } from "@/stores/toast";
 import UsageBadge from "@/components/UsageBadge.vue";
+import { DEAD_WEIGHT_ADVICE } from "@/utils/deadWeight";
 
 const router = useRouter();
 const toast = useToastStore();
@@ -237,14 +238,6 @@ function driftTitle(s: SnippetListItem): string {
   const what = reasons[v.status] ?? "";
   return v.detail ? `${when}: ${what}. ${v.detail}` : `${when}: ${what}.`;
 }
-
-/** The dead-weight nudge for a SNIPPET, passed to the shared badge. Kept here
- *  rather than inside the component because the remedy is kind-specific — a
- *  rule in the same position gets different advice (milestone 333 step 5). */
-const SNIPPET_DEAD_WEIGHT =
-  "Offered repeatedly without ever being opened — consider rewriting its " +
-  "“when to reach for it” so it says when, or deleting it. It takes a slot " +
-  "in every future auto-inject menu.";
 </script>
 
 <template>
@@ -431,7 +424,7 @@ const SNIPPET_DEAD_WEIGHT =
           <span v-if="driftBadge(s)" class="drift-tag" :title="driftTitle(s)">
             {{ driftBadge(s) }}
           </span>
-          <UsageBadge :usage="s.usage" :dead-weight-advice="SNIPPET_DEAD_WEIGHT" />
+          <UsageBadge :usage="s.usage" :dead-weight-advice="DEAD_WEIGHT_ADVICE.snippet" />
           <span v-if="s.shared" class="shared-tag" :title="`Shared by ${s.owner ?? 'another user'} — a suggestion, not your own record`">
             by {{ s.owner ?? "another user" }}
           </span>

src/scribe/mcp/tools/lessons.py

@@ -17,7 +17,7 @@ from scribe.services import lessons as lessons_svc
 from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
 from scribe.mcp.tools import systems as systems_tools
-from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes
+from scribe.services.note_usage import attach_usage, record_pulled
 
 
 # The payload shape lives in the service (`lesson_to_dict`), shared with the
@@ -65,7 +65,7 @@ async def list_lessons(
     # per lesson (#4196). An agent listing lessons can see which of its own
     # triggers are firing and which are not, which is the reading that leads to
     # `update_lesson` rather than to a second lesson about the same failure.
-    usage = await usage_for_notes([int(it["id"]) for it in labelled])
+    await attach_usage(labelled)
     rows = [
         {
             "id": it["id"], "title": it["title"], "tags": it.get("tags", []),
@@ -73,7 +73,7 @@ async def list_lessons(
             # Projected by `_note_to_item` straight off the `data` mirror —
             # absent when the row carries none, rather than an empty string.
             "when_to_apply": it.get("when_to_apply", ""),
-            "usage": usage.get(int(it["id"]), empty_usage()),
+            "usage": it["usage"],
             **({"shared": True, "owner": it.get("owner")} if it.get("shared") else {}),
         }
         for it in labelled
@@ -212,9 +212,7 @@ async def get_lesson(lesson_id: int, project_id: int = 0) -> dict:
     # Read BEFORE the pull is recorded, so the number an agent is shown is the
     # one that was true when it asked — otherwise every first read of a lesson
     # reports a pull that is its own.
-    out["usage"] = (await usage_for_notes([int(note.id)])).get(
+    await attach_usage([out])
-        int(note.id), empty_usage()
-    )
     record_pulled(
         user_id=uid, note_id=int(note.id),
         source="mcp_get_lesson", project_id=project_id,

src/scribe/mcp/tools/rulebooks.py

@@ -489,13 +489,20 @@ async def create_rule(
         order_index: Display order within the topic (default 0).
         force: Bypass the near-duplicate gate. By default, a title-identical rule
             already in this topic BLOCKS creation and returns its id so you update
-            it instead. Set true only for a genuinely distinct rule.
+            it instead. Set true only for a genuinely distinct rule. A rule or
+            preference that answers the same MOMENT under another title does
+            not block: the create goes through and the reply carries
+            `overlaps` and `overlap_note` — read the top one and decide.
     """
     uid = current_user_id()
     if not force:
         dup = await dedup_svc.find_duplicate_rule(title, topic_id=topic_id)
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "rule")
+    # Before the create, so the new rule cannot find itself (#4134).
+    overlaps = await dedup_svc.find_overlapping_rules(
+        uid, title, statement, when_to_apply,
+    )
     rule = await rulebooks_svc.create_rule(
         topic_id=topic_id, user_id=uid,
         title=title, statement=statement, when_to_apply=when_to_apply,
@@ -503,7 +510,9 @@ async def create_rule(
         why=why, how_to_apply=how_to_apply, order_index=order_index,
         verify_with=verify_with, expires_when=expires_when,
     )
-    return await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data = await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data.update(dedup_svc.overlap_response(overlaps, "rule"))
+    return data
 
 
 async def create_project_rule(
@@ -582,7 +591,9 @@ async def create_project_rule(
         order_index: Display order within the project's rule list (default 0).
         force: Bypass the near-duplicate gate. By default, a title-identical rule
             already on this project BLOCKS creation and returns its id so you
-            update it instead. Set true only for a genuinely distinct rule.
+            update it instead. Set true only for a genuinely distinct rule. An
+            overlap by meaning never blocks; it arrives as `overlaps` and
+            `overlap_note` on the reply — see create_rule.
     """
     uid = current_user_id()
     derived_title = title.strip() or statement.strip().split(".")[0][:50]
@@ -590,6 +601,9 @@ async def create_project_rule(
         dup = await dedup_svc.find_duplicate_rule(derived_title, project_id=project_id)
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "rule")
+    overlaps = await dedup_svc.find_overlapping_rules(
+        uid, derived_title, statement, when_to_apply, project_id=project_id,
+    )
     rule = await rulebooks_svc.create_project_rule(
         project_id=project_id, user_id=uid,
         title=derived_title, statement=statement, when_to_apply=when_to_apply,
@@ -597,7 +611,9 @@ async def create_project_rule(
         why=why, how_to_apply=how_to_apply, order_index=order_index,
         verify_with=verify_with, expires_when=expires_when,
     )
-    return await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data = await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data.update(dedup_svc.overlap_response(overlaps, "rule"))
+    return data
 
 
 async def update_rule(
@@ -784,7 +800,10 @@ async def create_preference(
             a preference could only be filed after the fact (#4249).
         force: Bypass the near-duplicate gate. For a genuinely distinct
             preference, not for one that is "mostly" different — a mostly
-            different preference is an update.
+            different preference is an update. A RULE that already answers
+            this moment comes back as `overlaps` / `overlap_note` on the reply
+            rather than blocking; if it says the same thing, the preference is
+            the weaker copy of it and should go.
     """
     uid = current_user_id()
     if not when_to_apply.strip():
@@ -803,13 +822,18 @@ async def create_preference(
         dup = await dedup_svc.find_duplicate_rule(title, topic_id=topic_id)
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "rule")
+    overlaps = await dedup_svc.find_overlapping_rules(
+        uid, title, statement, when_to_apply,
+    )
     rule = await rulebooks_svc.create_rule(
         topic_id=topic_id, user_id=uid,
         title=title, statement=statement, when_to_apply=when_to_apply,
         kind="preference", arose_from_id=arose_from_id,
         why=why, how_to_apply=how_to_apply, order_index=order_index,
     )
-    return await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data = await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data.update(dedup_svc.overlap_response(overlaps, "preference"))
+    return data
 
 
 async def update_preference(

src/scribe/mcp/tools/snippets.py

@@ -15,7 +15,7 @@ from scribe.mcp.tools import systems as systems_tools
 from scribe.services import access as access_svc
 from scribe.services import dedup as dedup_svc
 from scribe.services import snippets as snippets_svc
-from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes
+from scribe.services.note_usage import attach_usage, record_pulled
 from scribe.services import systems as systems_svc
 
 
@@ -90,9 +90,7 @@ async def list_snippets(
         repo=repo, path=path, symbol=symbol, verification=verification,
     )
     labeled = await access_svc.label_shared_items(uid, items)
-    usage = await usage_for_notes([int(it["id"]) for it in labeled])
+    await attach_usage(labeled)
-    for it in labeled:
-        it["usage"] = usage.get(int(it["id"]), empty_usage())
     return {"snippets": labeled, "total": total}
 
 

src/scribe/routes/knowledge.py

@@ -7,6 +7,7 @@ from scribe.auth import get_current_user_id, login_required
 from scribe.routes.utils import parse_pagination
 from scribe.services.access import label_shared_items
 from scribe.services.knowledge import FACET_TYPES
+from scribe.services.note_usage import attach_usage
 
 logger = logging.getLogger(__name__)
 
@@ -62,10 +63,23 @@ async def list_knowledge():
         offset=offset,
     )
 
+    items = await label_shared_items(uid, items)
+    # The surfaced-vs-opened counts, on the list a person actually browses
+    # (#4230). `usage_for_notes` always worked on every note row, but only the
+    # snippet and rule lists ever attached it — and this is the ONLY lesson and
+    # note list in the UI, so those two kinds had the counter collected and
+    # shown nowhere. Attaching here rather than teaching `/api/lessons` a
+    # second time is what closes both holes at once: `/knowledge` is how notes,
+    # lessons and processes are all browsed.
+    #
+    # Mixed kinds is not a problem for this: usage keys on the note row, which
+    # every facet of this feed is.
+    await attach_usage(items)
+
     return jsonify({
         # Mark rows another user owns: this feed can be mixed-ownership, and an
         # unmarked card reads as one the viewer wrote.
-        "items": await label_shared_items(uid, items),
+        "items": items,
         "total": total,
         "page": page,
         "per_page": limit,

src/scribe/routes/lessons.py

@@ -36,7 +36,7 @@ from scribe.services.access import (
     describe_provenance,
     label_shared_items,
 )
-from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes
+from scribe.services.note_usage import attach_usage, record_pulled
 
 logger = logging.getLogger(__name__)
 
@@ -83,9 +83,7 @@ async def list_lessons_route():
     # surfaced AND repeatedly opened, and the far commoner reading of the same
     # row is that the trigger fires on the wrong situation, which `update_lesson`
     # exists to fix.
-    usage = await usage_for_notes([int(it["id"]) for it in items])
+    await attach_usage(items)
-    for it in items:
-        it["usage"] = usage.get(int(it["id"]), empty_usage())
     return jsonify({"lessons": items, "total": total})
 
 
@@ -194,9 +192,7 @@ async def get_lesson_route(lesson_id: int):
         uid, out["learned_from"]
     )
     out.update(await describe_provenance(uid, note))
-    out["usage"] = (await usage_for_notes([lesson_id])).get(
+    await attach_usage([out])
-        lesson_id, empty_usage()
-    )
     # Opening the detail view IS a pull — the operator chose to look. Tagged
     # apart from the MCP sources so "an agent was handed it" and "a human read
     # it" stay distinguishable; they mean different things for pruning (#2085).

src/scribe/routes/snippets.py

@@ -19,7 +19,7 @@ from scribe.routes.utils import not_found, parse_pagination
 from scribe.services import dedup as dedup_svc
 from scribe.services import snippets as snippets_svc
 from scribe.services import systems as systems_svc
-from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes
+from scribe.services.note_usage import attach_usage, record_pulled
 from scribe.services.access import (
     can_write_note,
     describe_provenance,
@@ -75,9 +75,7 @@ async def list_snippets_route():
     # One aggregate for the whole page — a per-row lookup here would be N+1 by
     # construction. Every row gets the key, zero-filled, so the UI renders
     # "never pulled" rather than having to treat a missing field as a state.
-    usage = await usage_for_notes([int(it["id"]) for it in items])
+    await attach_usage(items)
-    for it in items:
-        it["usage"] = usage.get(int(it["id"]), empty_usage())
     return jsonify({"snippets": items, "total": total})
 
 
@@ -168,9 +166,7 @@ async def get_snippet_route(snippet_id: int):
         for s in await systems_svc.list_record_systems(note.user_id, snippet_id)
     ]
     data.update(await describe_provenance(uid, note))
-    data["usage"] = (await usage_for_notes([snippet_id])).get(
+    await attach_usage([data])
-        snippet_id, empty_usage()
-    )
     # Opening the detail view IS a pull — the operator chose to look. Tagged
     # apart from the MCP sources so "the agent reused it" and "a human read it"
     # stay distinguishable; they mean different things for pruning (#2085).

src/scribe/services/dedup.py

@@ -772,10 +772,17 @@ async def find_duplicate_rule(
     topic_id: int | None = None,
     project_id: int | None = None,
 ) -> DuplicateMatch | None:
-    """Title-based near-duplicate of a rule, scoped to the same topic (a rulebook
+    """Title-identical rule in the same topic (a rulebook rule) or the same
-    rule) or the same project (a project rule). Rules aren't a semantic-retrieval
+    project (a project rule) — the one signal certain enough to BLOCK on.
-    surface, so a normalized-title match is the right (and only) signal. Fail-open
+    Fail-open like find_duplicate_note.
-    like find_duplicate_note."""
+
+    This is not the only duplicate signal for rules. It said so until #4134 —
+    "rules aren't a semantic-retrieval surface" — which stopped being true
+    when rules were embedded (rule_document, semantic_search_rules), and a
+    title is the field LEAST likely to collide when someone is deliberately
+    writing a second record about the same moment. find_overlapping_rules is
+    the meaning half; it surfaces rather than blocks, for the reason recorded
+    above _RULE_OVERLAP_FLOOR."""
     norm = " ".join((title or "").split()).lower()
     if not norm or (topic_id is None and project_id is None):
         return None
@@ -797,6 +804,128 @@ async def find_duplicate_rule(
     return None
 
 
+
+# --- rule / preference overlap (#4134) ----------------------------------------
+# Rules and preferences are one table and one ranking: every hook arm searches
+# them with no `kind` filter. So a preference that restates a rule is not a
+# harmless near-copy — when only the preference places, a session receives
+# binding guidance labelled "preference" and treats it as optional. The title
+# gate above cannot see it: a second record about the same moment is exactly
+# the case where someone chose a different title.
+#
+# WHY THIS SURFACES INSTEAD OF BLOCKING. Measured 2026-09-21 on bge-small-en-
+# v1.5, querying with the gate's own rule_document shape across 16 sampled
+# records (10 preferences, 6 rules) and reading the nearest OTHER record (#4134
+# has the ids):
+#
+#   a preference rewording an existing rule                          0.850
+#   nearest distinct neighbours, 16 samples                0.672 – 0.853
+#     "when to delegate" beside "never delegate writing"             0.853
+#     "work lands on the working branch" beside "nothing reaches
+#      the release branch unasked"                                   0.850
+#     "let each action finish" beside "poll CI yourself"             0.847
+#
+# The two bands overlap: records that are deliberately distinct about one
+# moment — a rule for what must happen beside a rule for what must not — sit
+# exactly where a true restatement does. No threshold separates them, so a
+# block would refuse legitimate records and teach force=true on every create.
+# What the embedding CAN say reliably is "these answer the same moment", and
+# whether they say the same THING is a reading, which is the author's. So the
+# create goes through and carries the records it overlaps, with what to do if
+# they are the same.
+#
+# 0.80 is the floor because the one measured true duplicate sat at 0.850 and a
+# floor at the edge of it would miss the next, slightly looser rewording;
+# 6 of the 16 distinct neighbours also clear it, which is the cost, paid in
+# one line on the create's reply rather than in a refused write. Retune with
+# the embedder, not the corpus.
+_RULE_OVERLAP_FLOOR = 0.80
+_RULE_OVERLAP_LIMIT = 3
+
+
+@dataclass
+class RuleOverlap:
+    """An existing rule or preference that answers the same moment."""
+    id: int
+    title: str
+    kind: str          # "rule" | "preference"
+    project_id: int | None
+    similarity: float
+
+
+async def find_overlapping_rules(
+    user_id: int,
+    title: str,
+    statement: str,
+    when_to_apply: str,
+    *,
+    project_id: int | None = None,
+) -> list[RuleOverlap]:
+    """Existing rules AND preferences whose trigger reads as this one's.
+
+    Queried with rule_document — the exact shape the corpus is embedded as —
+    so the score compares like with like. Both kinds, because the harm is
+    across them (#4134).
+
+    Scope follows the new record's home. A project rule is compared with
+    global rules plus that project's own, the set it will rank against. A
+    global record (project_id None) applies everywhere, so it is compared with
+    every rule the caller owns: a global rule restating one project's rule is
+    the same overlap, arriving in that project.
+
+    Run BEFORE the create, so the new record cannot match itself. Never
+    raises: an overlap is advice, and a create must not depend on it.
+    """
+    doc_title, doc_body = embeddings_svc.rule_document(title, statement, when_to_apply)
+    query = "\n\n".join(p for p in (doc_title, doc_body) if p)
+    # The note gate's floor, for the same reason: a short document sits in a
+    # tight neighbourhood and resembles everything.
+    if len(query.strip()) < _MIN_BODY_FOR_SEMANTIC:
+        return []
+    try:
+        hits = await embeddings_svc.semantic_search_rules(
+            user_id, query, limit=_RULE_OVERLAP_LIMIT,
+            threshold=_RULE_OVERLAP_FLOOR,
+            project_id=project_id, everywhere=project_id is None,
+        )
+    except Exception:
+        logger.debug("rule overlap check skipped", exc_info=True)
+        return []
+    return [
+        RuleOverlap(
+            id=rule.id, title=rule.title, kind=rule.kind or "rule",
+            project_id=rule.project_id, similarity=round(score, 3),
+        )
+        for score, rule in hits
+    ]
+
+
+def overlap_response(overlaps: list[RuleOverlap], new_kind: str) -> dict:
+    """The keys a rule/preference create adds to its reply when the record it
+    just wrote answers the same moment as an existing one. Empty when none."""
+    if not overlaps:
+        return {}
+    top = overlaps[0]
+    named = "; ".join(
+        f'{o.kind} {o.id} "{o.title}" ({o.similarity})' for o in overlaps
+    )
+    return {
+        "overlaps": [
+            {"id": o.id, "title": o.title, "kind": o.kind,
+             "project_id": o.project_id, "similarity": o.similarity}
+            for o in overlaps
+        ],
+        "overlap_note": (
+            f"Created — and it answers the same moment as: {named}. Read "
+            f"{top.kind} {top.id} now. If it says the same thing, fold what is "
+            f"new into it (update_{top.kind}) and delete this {new_kind}: two "
+            f"records ranked together split one instruction, and the weaker "
+            f"one can arrive alone. If they say different things about one "
+            f"moment, keep both — that is common and fine."
+        ),
+    }
+
+
 # --- the plan gate (milestone 415) -------------------------------------------
 # A session asked "what work is open?" that cannot see an existing plan makes a
 # second one: a new milestone beside the one that already covers the work, or

src/scribe/services/note_usage.py

@@ -30,6 +30,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from collections.abc import Sequence
 
 from sqlalchemy import case, func, select
 
@@ -271,3 +272,64 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
             if latest and (slot["last_pulled_at"] or "") < latest:
                 slot["last_pulled_at"] = latest
     return out
+
+
+def _row_id(row: dict, key: str) -> int | None:
+    """The note id on a payload row, or None when there is not one to read.
+
+    Skipping is deliberate: an id this cannot parse is not a reason to fail a
+    whole list, and GUESSING one would credit another record's counts to this
+    row — a wrong chip is worse than no chip, because it reads as a
+    measurement. `bool` is excluded explicitly because `int(True)` is 1, which
+    would quietly attach note #1's usage to a row carrying a flag.
+    """
+    raw = row.get(key)
+    if raw is None or isinstance(raw, bool):
+        return None
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        return None
+
+
+async def attach_usage(rows: Sequence[dict], *, key: str = "id") -> None:
+    """Add `usage` to every row of a payload a door is about to return (#4230).
+
+    The one seam both doors and every record kind share. Before this, four call
+    sites carried their own copy of the same lines — two list routes and two
+    detail routes — and `/api/knowledge`, which is the list a person ACTUALLY
+    browses notes and lessons in, was about to become a fifth. That is how the
+    chip came to reach two record kinds out of four while a service named
+    `usage_for_notes` worked on all of them: each door read fine on its own,
+    and nobody was comparing them.
+
+    ONE AGGREGATE FOR THE WHOLE PAGE. `usage_for_notes` is a single GROUP BY
+    over the id set; calling it per row would be N+1 by construction, which is
+    the one shape a list route must not have.
+
+    EVERY ROW GETS THE KEY, zero-filled, so a record predating the table reads
+    as "never surfaced, never pulled" rather than making the UI treat a missing
+    field as a state. `UsageBadge` then renders nothing at all below one
+    surfacing, because "0/0" would look like a verdict where there is only an
+    absence of evidence.
+
+    NO try/except HERE, deliberately — it is not an oversight. The fail-open
+    already lives one layer down: `usage_for_notes` catches its own failure,
+    reports it through `_report_failure("readout")` and returns the zero-filled
+    map, so a broken readout degrades without breaking the list it decorates.
+    Wrapping it again would swallow the REPORT along with the error, and a
+    silently-swallowed readout failure is exactly #2663 — every counter reading
+    zero in production for weeks while the writes landed fine.
+
+    Mutates in place and returns None, matching how the call sites already used
+    it: these rows are the payload, not a copy of it.
+
+    A detail payload is just a one-row list — `await attach_usage([data])` —
+    so the single-record doors share this seam rather than keeping a second
+    shape that could drift from it.
+    """
+    pairs = [(row, _row_id(row, key)) for row in rows]
+    usage = await usage_for_notes([nid for _, nid in pairs if nid is not None])
+    for row, nid in pairs:
+        if nid is not None:
+            row["usage"] = usage.get(nid, empty_usage())

src/scribe/services/plugin_context.py

@@ -2298,7 +2298,17 @@ async def build_write_path_hint(
     # of this guard: that one runs a SEMANTIC search, and moving it here would
     # run an embedding query on every write in the session. Its gating is a
     # separate question from this one (see the note on #3244).
+    # The design arm (#4256) is decided HERE, above the guard, so a UI write
+    # that matched no prior art still carries it — and it returns on its own
+    # rather than joining the guard's condition, because joining it would let
+    # a design line switch the standing-rule arm below on for writes where it
+    # has never run, moving that arm's call distribution under its floor.
+    design_text, design_dedup = await _design_arm(
+        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 design_text:
+            return {**empty, "context": design_text, "derive_keys": [design_dedup]}
         return empty
 
     owners = await owner_names_for({
@@ -2320,6 +2330,9 @@ async def build_write_path_hint(
     # Seeded with the staleness line, which is decided above the early
     # return and so cannot wait for this list to exist.
     lines: list[str] = list(staleness)
+    # First after staleness: it BINDS, where everything below is prior art.
+    if design_text:
+        lines.append(design_text)
     sync_note_ids: list[int] = []
     if synced:
         # The sync framing (#2708). Deliberately imperative about the record —
@@ -2552,7 +2565,9 @@ async def build_write_path_hint(
         "stamped": stamped,
         "divergence": divergence,
         "derive": derive,
-        "derive_keys": [d["key"] for d in derive],
+        "derive_keys": [d["key"] for d in derive] + (
+            [design_dedup] if design_dedup else []
+        ),
         "rule_ids": rule_ids,
         "checkpoint": checkpoint,
     }
@@ -2703,6 +2718,106 @@ async def build_tool_rule_hint(
     return out
 
 
+# --- the design-guidance arm (#4256) ----------------------------------------
+# A design system BINDS like a rule, and until this it had one channel: the
+# session-start block, which names it and the call that reads its prose. That
+# is complete for a session that knows to ask and silent for one that is
+# writing a component — the same gap every unasked arm exists to close.
+#
+# A TRIGGER, NOT A SEARCH. A project has exactly one design system
+# (projects.design_system_id is a single FK), so there is nothing to rank and
+# no vector to compute: the question "does this guidance apply here" is
+# answered by the file being UI. Deterministic and cheap, and it takes no
+# slot from the ranked menu — the band, floor and budget the other arms were
+# tuned against are untouched by construction, which is why this adds no
+# retrieval_logs row: there is no score distribution for it to join.
+#
+# AN INDEX, NOT THE PROSE. Resolved guidance runs to thousands of characters
+# (a house style is long by nature), which would take most of the hook's
+# additionalContext cap on its own. So the line names the SECTIONS of each
+# inherited layer — the headings are self-describing ("Where the accent must
+# NOT appear", "Voice and tone") the way rule titles are — and inlines only a
+# layer short enough to be a line: in practice the leaf, since a child system
+# holds just its departure from the house style. Choosing a paragraph by
+# meaning would need the guidance embedded per section; that is justified
+# only if this index turns out not to be read.
+#
+# ONCE PER SESSION PER SYSTEM, on the hook's token-keyed channel
+# (`exclude_derive`, keyed `design:<id>`). That channel already dedups opaque
+# keys on its own file, so the arm needs no new plugin state.
+_DESIGN_UI_EXTENSIONS = frozenset({
+    ".vue", ".svelte", ".css", ".scss", ".sass", ".less",
+    ".tsx", ".jsx", ".html",
+})
+# A guidance layer this short is shown whole; anything longer is indexed.
+_DESIGN_INLINE_CHARS = 500
+_DESIGN_HEADING = re.compile(r"^##\s+(.+?)\s*$", re.M)
+
+
+def design_key(design_system_id: int) -> str:
+    """The dedup token for the design arm on the hook's keyed channel."""
+    return f"design:{int(design_system_id)}"
+
+
+def is_ui_path(path: str) -> bool:
+    """Whether writing `path` is writing UI — the design arm's trigger."""
+    name = (path or "").rsplit("/", 1)[-1].lower()
+    return any(name.endswith(ext) for ext in _DESIGN_UI_EXTENSIONS)
+
+
+def _design_line(path: str, design: dict) -> str:
+    """Name the design system that binds this file, and what its prose covers."""
+    ds_id = design["id"]
+    layers: list[str] = []
+    for layer in design.get("guidance") or []:
+        text = (layer.get("guidance") or "").strip()
+        if not text:
+            continue
+        flat = " ".join(text.split())
+        headings = _DESIGN_HEADING.findall(text)
+        if len(flat) <= _DESIGN_INLINE_CHARS:
+            layers.append(f"{layer['title']}: \"{flat}\"")
+        elif headings:
+            layers.append(f"{layer['title']} covers " + " · ".join(headings))
+        else:
+            short, _cut = elide(flat, _DESIGN_INLINE_CHARS)
+            layers.append(f"{layer['title']}: \"{short}\"")
+    inherits = (
+        " (inherits " + " › ".join(design["inherits_from"]) + ")"
+        if design.get("inherits_from") else ""
+    )
+    out = (
+        f"> Design system binds `{path}`: {design['title']} (id {ds_id}){inherits}. "
+        f"Read `get_design_system({ds_id})` → `resolved_guidance` before writing "
+        f"UI here, and take values from `resolve_design_system({ds_id})` rather "
+        f"than hand-writing them."
+    )
+    if layers:
+        out += " " + "; ".join(layers) + "."
+    return out + " (Shown once per session.)"
+
+
+async def _design_arm(
+    user_id: int, project_id: int, path: str, skip: set[str],
+) -> tuple[str, str]:
+    """(line, dedup key) for a UI write in a project with a design system,
+    or ("", "") — never raises: a design hint must never break a write."""
+    if not project_id or not is_ui_path(path):
+        return "", ""
+    try:
+        project = await projects_svc.get_project(user_id, project_id)
+        ds_id = getattr(project, "design_system_id", None) if project else None
+        if not ds_id or design_key(ds_id) in skip:
+            return "", ""
+        design = await design_systems_svc.design_context(user_id, ds_id)
+        if not design:
+            return "", ""
+        return _design_line(path, design), design_key(ds_id)
+    except Exception:
+        logger.debug("write-path design arm failed", exc_info=True)
+        return "", ""
+
+
 def _derive_line(path: str, derive: list[dict]) -> str:
     """The ledger's word on the names being written (#2900): a duplicate
     family to derive, or a canon to reuse — said at the write."""

src/scribe/services/shape_ledger.py

@@ -1521,11 +1521,37 @@ _SEMANTIC_CAP = 150
 # "both are about migrations"; first live run paired every alembic
 # upgrade()/downgrade() with an unrelated canon at exactly that band.
 _SEMANTIC_FLOOR = 0.8
+# How many above-floor hits the semantic arm asks for. Named because the
+# NUMBER is load-bearing twice over: it caps the work, and a result set that
+# came back short of it is a complete picture of what cleared the floor —
+# which is what lets a miss be read as evidence rather than as a cut-off
+# (`BASIS_NO_SEMANTIC_MATCH`).
+_SEMANTIC_LIMIT = 3
+# The proposer looked at this body, compared it against every canon in its
+# language family, and matched none of them above `_SEMANTIC_FLOOR` (#4208).
+#
+# This is a NEGATIVE RESULT, and it is stored because it is the only evidence
+# in the ledger that speaks to what a shape MEANS rather than what it looks
+# like. `proposal_basis` otherwise names how a proposal was arrived at; here
+# it records that the arm ran and came back empty, with `proposed_snippet_id`
+# left NULL. Every reader keys "is there a proposal" on `proposed_snippet_id`
+# or `proposal_group`, never on the basis, so this cannot be mistaken for one:
+# `list_shapes(proposal=...)` and `confirm_shape_proposals` both filter on the
+# id, and the latter requires it non-NULL before it will confirm anything.
+#
+# It is deliberately NOT written for the two cases that merely look the same:
+# a body too thin to compare (`_substance` below the write-path minimum), and
+# a row the per-refresh cap never reached. Those are "I cannot tell", and the
+# ledger's standing discipline — the one `FORM_UNKNOWN` enforces everywhere
+# else — is that not knowing must make a check quieter, never more confident.
+BASIS_NO_SEMANTIC_MATCH = "no-semantic-match"
 # Bump when a basis's rule changes: rows remember the (body, ruleset) they
 # were examined under, so a tightened rule re-examines everything once.
 # v3: language-family gate on the sym bases, reference stoplist, semantic
 # restricted to the shape's own project (#2871).
-_PROPOSER_VERSION = 3
+# v4: the semantic arm records its misses as well as its hits (#4208), so
+# every already-examined row must be looked at once more to acquire one.
+_PROPOSER_VERSION = 4
 # Signature resemblance floor, name blanked (difflib ratio) — and a length
 # floor, because `def NAME():` resembles `def NAME(x):` at 0.95 while saying
 # nothing; a family shape has parameters to resemble.
@@ -1760,8 +1786,29 @@ def _substance(text: str) -> int:
 
 
 async def _semantic_canon(
-    user_id: int, body: str, allowed: set[int]
+    user_id: int, body: str, allowed: set[int], *, report: dict | None = None
 ) -> tuple[int, float] | None:
+    """The canon this body MEANS, or None.
+
+    `report` is an out-param in the style `semantic_search_notes` already
+    uses, and it carries the one thing the return value cannot: whether a
+    None is EVIDENCE. `report["conclusive"] = True` says the arm really
+    compared this body against the allowed canons and none cleared the floor.
+    It is left unset whenever the arm could not form an opinion — a body with
+    too little substance to embed, no allowed canon to compare against, or a
+    result set that came back full and may therefore have been truncated.
+
+    The truncation case is why `_SEMANTIC_LIMIT` is named. The search returns
+    the top N above the floor; if it returns fewer than N, N was not binding
+    and we have seen everything that cleared the floor, so "no allowed canon
+    among them" is a fact about the corpus. If it returns exactly N, an
+    allowed canon could be sitting at N+1 and the same silence means nothing.
+    Reading the second case as the first is how a cut-off becomes a finding.
+
+    Callers must treat a missing key as "cannot tell", never as "no match" —
+    which is also what makes the existing test double, an `AsyncMock` that
+    returns None and touches no report, stay correct by default.
+    """
     from scribe.services.embeddings import semantic_search_notes
     from scribe.services.plugin_context import (
         WRITEPATH_DEFAULT_THRESHOLD, WRITEPATH_MIN_CODE_CHARS, concept_query,
@@ -1771,13 +1818,15 @@ async def _semantic_canon(
         return None
     query = concept_query(body) or body
     hits = await semantic_search_notes(
-        user_id, query, limit=3,
+        user_id, query, limit=_SEMANTIC_LIMIT,
         threshold=max(WRITEPATH_DEFAULT_THRESHOLD, _SEMANTIC_FLOOR),
         note_type="snippet", scope="browse",
     )
     for score, note in hits:
         if int(note.id) in allowed:
             return int(note.id), round(float(score), 3)
+    if report is not None and len(hits) < _SEMANTIC_LIMIT:
+        report["conclusive"] = True
     return None
 
 
@@ -1870,16 +1919,29 @@ async def propose_for_repo(
                 row.proposed_sha = ""
                 continue
             checked += 1
+            verdict: dict = {}
             try:
-                found = await _semantic_canon(user_id, d[5], semantic_allowed(row.path))
+                found = await _semantic_canon(
+                    user_id, d[5], semantic_allowed(row.path), report=verdict,
+                )
             except Exception:
                 logger.warning("semantic proposal failed", exc_info=True)
                 found = None
+                # An arm that threw formed no opinion. Clearing this is not
+                # belt-and-braces: a partially-filled report would record a
+                # failure as a finding about the code.
+                verdict = {}
             if found:
                 row.proposed_snippet_id, row.proposal_score = found
                 row.proposal_basis = "semantic"
                 row.proposal_group = None
                 proposed += 1
+            elif verdict.get("conclusive"):
+                # No canon, and the arm is sure of it. Kept as the row's basis
+                # with `proposed_snippet_id` still NULL, so it reads as "asked
+                # and answered" rather than "not asked" — the distinction
+                # `flag_divergence` needs and could not previously make.
+                row.proposal_basis = BASIS_NO_SEMANTIC_MATCH
         await session.commit()
     return {"examined": examined, "proposed": proposed, "semantic_checked": checked}
 
@@ -2435,6 +2497,26 @@ 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"
+                # ...and the converse, which is the only evidence here that
+                # is about MEANING rather than shape (#4208). The four false
+                # prompts #4204 left standing are callables in a directory of
+                # callables: at the signature level they are indistinguishable
+                # from #2793's acceptance case, a sync `confirmDanger` beside
+                # an async confirm canon, and no refinement of `shape_form`
+                # ever separates them — a registry accessor and a service unit
+                # differ by the JOB they do, which a signature does not carry.
+                #
+                # The proposer does read bodies, and when its semantic arm
+                # compared this one against every canon in its language family
+                # and matched none of them, that is a positive finding that
+                # this shape is not the canon's work. Urging the canon anyway
+                # would be asserting over a measurement we already hold.
+                #
+                # Only the conclusive miss is stored, so an unexamined row and
+                # a body too thin to embed still ask the question rather than
+                # being quietly excused.
+                if r.proposal_basis == BASIS_NO_SEMANTIC_MATCH:
+                    continue
                 # 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

tests/conftest.py

@@ -147,3 +147,20 @@ def _no_rule_arm():
     with patch("scribe.services.plugin_context.semantic_search_rules",
                AsyncMock(return_value=[])):
         yield
+
+
+@pytest.fixture(autouse=True)
+def _no_rule_overlap():
+    """Stub the rule/preference create path's overlap check (#4134).
+
+    The same reason as _no_rule_arm, one door over: every create_rule /
+    create_project_rule / create_preference now asks semantic_search_rules
+    whether an existing record answers the same moment, so each existing
+    rule-tool unit test would load the embedding model through a call it never
+    meant to make. The check's own behaviour is tested in
+    tests/test_rule_overlap_gate.py, which binds the real function at import
+    time — before this patch runs — and stubs the search beneath it instead.
+    """
+    with patch("scribe.services.dedup.find_overlapping_rules",
+               AsyncMock(return_value=[])):
+        yield

tests/helpers.py

@@ -418,3 +418,55 @@ def writepath_cfg(**over):
     }
     cfg.update(over)
     return cfg
+
+
+def need_tools(*tools):
+    """Skip unless every named executable is on PATH.
+
+    For the hook tests, which drive `plugin/hooks/*.sh` through a real shell
+    and therefore depend on whatever that shell reaches for — `jq`, `awk`,
+    `git`, `curl`. Those are present on the CI image and routinely absent from
+    a developer's box, and the honest answer there is "not exercised", not a
+    failure: a red test would say the hook is broken when nothing about the
+    hook was ever run.
+
+    A skip rather than a stub on purpose. Stubbing `jq` would test the stub —
+    these tests exist precisely because the shell's behaviour is the thing in
+    question (#2932), so anything short of the real tool proves nothing.
+
+    Consolidated 2026-09-21 from three byte-identical copies in
+    test_contract_around_the_change, test_hook_json_reader and
+    test_session_slippage_readout.
+    """
+    import shutil
+
+    import pytest
+
+    for t in tools:
+        if shutil.which(t) is None:
+            pytest.skip(f"hook runtime tool {t!r} not installed")
+
+
+async def rule_row(rule_id: int):
+    """Read a Rule straight from Postgres, outside whatever session the code
+    under test used.
+
+    THE POINT IS THE SEPARATE SESSION. An integration test that asserts on the
+    object the service just returned is asserting on that session's identity
+    map, which can hold a value the database never accepted — a column the
+    write never reached, a default the ORM supplied rather than the schema.
+    Opening a new session forces a real read and is the only way `verify_with`
+    being NULL rather than "" is distinguishable at all (milestone 312).
+
+    Import is lazy because this module is imported by unit tests that have no
+    database and must not pay for one — the same reason `plugin_config` defers
+    its service imports.
+
+    Consolidated 2026-09-21 from two byte-identical copies in
+    test_integration_rule_move and test_integration_rule_verification.
+    """
+    from scribe.models import async_session
+    from scribe.models.rulebook import Rule
+
+    async with async_session() as s:
+        return await s.get(Rule, rule_id)

tests/test_contract_around_the_change.py

@@ -34,24 +34,19 @@ A FIXTURE REPO, NEVER THIS ONE. Asserting against Scribe's own files would
 make the test a description of today's tree, failing the next time someone
 renames something (rule 115's reasoning, one floor down).
 """
-import shutil
 import subprocess
 from pathlib import Path
 
 import pytest
 
+from tests.helpers import need_tools
+
 DEFS = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "scribe_defs.sh"
 HOOK = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "scribe_prior_art.sh"
 
 
-def _need(*tools):
-    for t in tools:
-        if shutil.which(t) is None:
-            pytest.skip(f"hook runtime tool {t!r} not installed")
-
-
 def run(script: str) -> str:
-    _need("bash", "awk", "git", "grep", "sed")
+    need_tools("bash", "awk", "git", "grep", "sed")
     r = subprocess.run(
         ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
         capture_output=True, text=True,
@@ -63,7 +58,7 @@ def run(script: str) -> str:
 @pytest.fixture()
 def repo(tmp_path):
     """A small git repo: a definition with a reader, and one without."""
-    _need("git")
+    need_tools("git")
     # `other` lives in lib.py BESIDE widget, and nothing references it. That
     # placement is the point of the no-readers case: putting it in its own
     # file would leave that file as its reader, since only the file being
@@ -248,6 +243,6 @@ def test_the_hook_asks_the_contract_question_first():
 
 
 def test_the_hook_is_still_shell_valid():
-    _need("bash")
+    need_tools("bash")
     subprocess.run(["bash", "-n", str(HOOK)], check=True)
     subprocess.run(["bash", "-n", str(DEFS)], check=True)

tests/test_divergence_meaning_gate.py

@@ -0,0 +1,182 @@
+"""A divergence prompt may be silenced by MEANING, never by silence (#4208).
+
+WHAT THIS IS ABOUT. #4204 gave the divergence check a structural gate: a canon
+is only urged on a shape whose form could plausibly BE it. That silenced one of
+the five false prompts it was filed for. The other four are `def` helpers in a
+directory whose canon is an `async def` service unit — callables beside a
+callable — and no refinement of `shape_form` ever separates them, because they
+differ from #2793's acceptance case (a hand-rolled sync `confirmDanger` where
+an async confirm helper is canon) only by the JOB they do. A signature does not
+carry a job.
+
+So the lever has to be meaning, and the ledger already holds one reading of it:
+the proposer's semantic arm embeds each definition's own BODY against canon.
+What it did not do was record its misses. A hit became `proposal_basis =
+"semantic"`; a miss left the row indistinguishable from a row nobody had looked
+at yet. `flag_divergence` could therefore ask the proposer "do you agree this is
+the canon?" but never "did you check, and did you find it is not?".
+
+THE WHOLE RISK IS IN THE NEGATIVE. A miss is only evidence if the arm actually
+formed an opinion, and there are three ways for it to come back empty that look
+identical from the outside:
+
+    body too thin to embed   -> no opinion
+    no allowed canon to test -> no opinion
+    result set was truncated -> no opinion (the canon may be at N+1)
+    compared, nothing above the floor -> EVIDENCE
+
+Only the last may silence a prompt. Reading any of the others as a negative is
+how "I cannot tell" turns into "I checked" — the exact failure #4204 was opened
+on, and the one `FORM_UNKNOWN` already guards against everywhere else in this
+module: not knowing must make a check QUIETER, never more confident.
+
+These tests pin the report contract that carries that distinction. The
+end-to-end behaviour — a conclusive miss silencing a real prompt while #2793's
+acceptance case still raises — is in
+tests/test_integration_shape_classify.py, because it needs real rows.
+"""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.services.shape_ledger import (
+    _SEMANTIC_LIMIT, BASIS_NO_SEMANTIC_MATCH, _semantic_canon,
+)
+
+# Comfortably over WRITEPATH_MIN_CODE_CHARS (48 non-whitespace characters), so
+# these tests exercise the comparison rather than the substance guard. One of
+# #4204's four survivors, quoted rather than invented.
+BODY = (
+    "def is_registered(source: str) -> bool:\n"
+    "    return source in _REGISTRY and _REGISTRY[source].enabled\n"
+)
+TOO_THIN = "def f():\n    pass\n"
+
+CANON = 2860          # the allowed canon, as a caller would pass it
+OTHER = 9999          # a snippet that is not in the allowed set
+
+
+class _FakeNote:
+    """Only `.id` is read off a hit."""
+
+    def __init__(self, note_id: int) -> None:
+        self.id = note_id
+
+
+def _hits(*hits: tuple[float, int]) -> AsyncMock:
+    return AsyncMock(return_value=[(score, _FakeNote(nid)) for score, nid in hits])
+
+
+def _patch(mock: AsyncMock):
+    return patch("scribe.services.embeddings.semantic_search_notes", mock)
+
+
+# ── the miss that IS evidence ────────────────────────────────────────────
+
+
+async def test_a_short_result_set_is_a_conclusive_miss() -> None:
+    """Fewer hits than asked for means the limit was not binding: everything
+    above the floor came back, and the canon was not among it. That is a fact
+    about the corpus, not an artefact of where the list was cut."""
+    mock = _hits((0.91, OTHER))
+    report: dict = {}
+    with _patch(mock):
+        found = await _semantic_canon(1, BODY, {CANON}, report=report)
+    assert found is None
+    assert report.get("conclusive") is True
+
+
+async def test_an_empty_result_set_is_also_conclusive() -> None:
+    """Nothing cleared the floor at all — the strongest form of the miss."""
+    report: dict = {}
+    with _patch(_hits()):
+        assert await _semantic_canon(1, BODY, {CANON}, report=report) is None
+    assert report.get("conclusive") is True
+
+
+# ── the three misses that are NOT ────────────────────────────────────────
+
+
+async def test_a_full_result_set_may_have_been_truncated() -> None:
+    """The case that makes `_SEMANTIC_LIMIT` load-bearing rather than a tuning
+    knob. The search returns the top N above the floor; when it returns
+    exactly N, an allowed canon can be sitting at N+1 and this same silence
+    would mean nothing. Reading it as a negative would silence real
+    divergences in direct proportion to how many snippets the operator has —
+    a check that quietly weakens as the corpus grows, which is the worst
+    possible failure mode for a guard nobody is watching."""
+    mock = _hits(*[(0.9, OTHER + i) for i in range(_SEMANTIC_LIMIT)])
+    report: dict = {}
+    with _patch(mock):
+        assert await _semantic_canon(1, BODY, {CANON}, report=report) is None
+    assert "conclusive" not in report
+
+
+async def test_a_body_too_thin_to_embed_forms_no_opinion() -> None:
+    """And does not spend an embedding finding that out."""
+    mock = _hits()
+    report: dict = {}
+    with _patch(mock):
+        assert await _semantic_canon(1, TOO_THIN, {CANON}, report=report) is None
+    assert "conclusive" not in report
+    mock.assert_not_awaited()
+
+
+async def test_no_allowed_canon_means_nothing_was_compared() -> None:
+    """An empty allowed set is not "the canons all missed" — there were none
+    to miss. Distinct because the language-family gate (#2871) empties this
+    set routinely: a Vue body simply has no Python canon to be compared to."""
+    mock = _hits()
+    report: dict = {}
+    with _patch(mock):
+        assert await _semantic_canon(1, BODY, set(), report=report) is None
+    assert "conclusive" not in report
+    mock.assert_not_awaited()
+
+
+# ── a hit is a proposal, not a miss ──────────────────────────────────────
+
+
+async def test_a_hit_returns_the_canon_and_claims_no_miss() -> None:
+    mock = _hits((0.88, CANON))
+    report: dict = {}
+    with _patch(mock):
+        found = await _semantic_canon(1, BODY, {CANON}, report=report)
+    assert found == (CANON, 0.88)
+    assert "conclusive" not in report
+
+
+async def test_an_allowed_canon_below_the_top_hit_still_wins() -> None:
+    """The scan is over the whole result set, so a disallowed snippet ranking
+    first does not hide an allowed one behind it. Pinned because if it did,
+    the short-list case above would start reporting conclusive misses for
+    bodies that DO have a canon."""
+    mock = _hits((0.95, OTHER), (0.83, CANON))
+    report: dict = {}
+    with _patch(mock):
+        found = await _semantic_canon(1, BODY, {CANON}, report=report)
+    assert found == (CANON, 0.83)
+    assert "conclusive" not in report
+
+
+# ── the contract callers depend on ───────────────────────────────────────
+
+
+async def test_a_caller_that_passes_no_report_still_gets_an_answer() -> None:
+    """The existing test double is an `AsyncMock(return_value=None)` that
+    never touches a report. Absence of the key must therefore mean "cannot
+    tell" at every call site — so a stub, an older caller, or an arm that
+    threw all default to asking the question rather than excusing it."""
+    with _patch(_hits()):
+        assert await _semantic_canon(1, BODY, {CANON}) is None
+
+
+@pytest.mark.parametrize("value", ["semantic", "symbol", "reference", "derive"])
+def test_the_miss_basis_is_not_one_of_the_proposal_bases(value: str) -> None:
+    """It shares a column with them and must not collide: every reader keys
+    "is there a proposal" on `proposed_snippet_id`, but `confirm_shape_proposals`
+    filters BY basis, and a collision there would mean confirming a miss as
+    though it were a match."""
+    assert BASIS_NO_SEMANTIC_MATCH != value

tests/test_hook_json_reader.py

@@ -22,25 +22,20 @@ from __future__ import annotations
 
 import json
 import re
-import shutil
 import subprocess
 from pathlib import Path
 from urllib.parse import quote
 
 import pytest
 
+from tests.helpers import need_tools
+
 HOOKS = Path(__file__).resolve().parents[1] / "plugin" / "hooks"
 DEFS = HOOKS / "scribe_defs.sh"
 PARSER = HOOKS / "scribe_json.awk"
 TURN = HOOKS / "scribe_turn.awk"
 
 
-def _need(*tools):
-    for t in tools:
-        if shutil.which(t) is None:
-            pytest.skip(f"hook runtime tool {t!r} not installed")
-
-
 def sh(script: str, stdin: str = "") -> str:
     """Run a snippet with scribe_defs.sh sourced, under the hooks' own flags.
 
@@ -50,7 +45,7 @@ def sh(script: str, stdin: str = "") -> str:
     quietly normalise the characters it exists to check: the first version of
     this file did, and reported a round-trip failure that was entirely its own.
     """
-    _need("bash", "awk")
+    need_tools("bash", "awk")
     r = subprocess.run(
         ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
         input=stdin.encode(), capture_output=True,
@@ -60,7 +55,7 @@ def sh(script: str, stdin: str = "") -> str:
 
 
 def flat(doc: str, mode: str = "whole") -> list[tuple[str, str, str]]:
-    _need("awk")
+    need_tools("awk")
     r = subprocess.run(["awk", "-v", f"mode={mode}", "-f", str(PARSER)],
                        input=doc.encode(), capture_output=True)
     assert r.returncode == 0, r.stderr.decode()
@@ -260,7 +255,7 @@ def test_urlenc_and_the_envelope_can_fail():
 # The transcript turn, which is the largest thing jq was doing here.
 
 def _turn(records: list[dict]) -> dict:
-    _need("awk")
+    need_tools("awk")
     doc = "\n".join(json.dumps(r) for r in records) + "\n"
     p1 = subprocess.run(["awk", "-v", "mode=lines", "-f", str(PARSER)],
                         input=doc, capture_output=True, text=True)

tests/test_integration_rule_move.py

@@ -20,7 +20,7 @@ from scribe.models.rulebook import Rule
 from scribe.services import canonical_systems as canonical_svc
 from scribe.services import rule_versions as rv_svc
 from scribe.services import rulebooks as rulebooks_svc
-from tests.helpers import ensure_user
+from tests.helpers import ensure_user, rule_row
 
 pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
@@ -69,17 +69,12 @@ async def homes():
     return ids
 
 
-async def _row(rule_id: int) -> Rule:
-    async with async_session() as s:
-        return await s.get(Rule, rule_id)
-
-
 async def test_a_project_rule_becomes_global_and_keeps_everything(homes):
     owner, rule_id = homes["owner"], homes["rule"]
     moved = await rulebooks_svc.move_rule(rule_id, owner, topic_id=homes["topic"])
 
     assert moved.id == rule_id
-    row = await _row(rule_id)
+    row = await rule_row(rule_id)
     assert (row.topic_id, row.project_id) == (homes["topic"], None)
     assert len(await rv_svc.list_versions(rule_id)) == 1, "the move must not drop history"
     areas = await rulebooks_svc.list_rule_systems([rule_id])
@@ -100,7 +95,7 @@ async def test_a_global_rule_can_move_onto_a_project(homes):
     owner, rule_id = homes["owner"], homes["rule"]
     await rulebooks_svc.move_rule(rule_id, owner, topic_id=homes["topic"])
     await rulebooks_svc.move_rule(rule_id, owner, project_id=homes["other"])
-    row = await _row(rule_id)
+    row = await rule_row(rule_id)
     assert (row.topic_id, row.project_id) == (None, homes["other"])
 
 
@@ -126,7 +121,7 @@ async def test_refusals_happen_before_anything_is_written(homes):
     with pytest.raises(ValueError, match=f"rule {clash.id}"):
         await rulebooks_svc.move_rule(rule_id, owner, topic_id=homes["topic"])
 
-    row = await _row(rule_id)
+    row = await rule_row(rule_id)
     assert (row.topic_id, row.project_id) == (None, homes["home"])
 
 

tests/test_integration_rule_verification.py

@@ -23,7 +23,7 @@ import pytest_asyncio
 from scribe.models import async_session
 from scribe.models.rulebook import Rule
 from scribe.services import rulebooks as rulebooks_svc
-from tests.helpers import ensure_user
+from tests.helpers import ensure_user, rule_row
 
 pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
@@ -57,13 +57,8 @@ async def constraint():
     return {"uid": uid, "rule": rule.id}
 
 
-async def _row(rule_id: int) -> Rule:
-    async with async_session() as s:
-        return await s.get(Rule, rule_id)
-
-
 async def test_the_check_and_its_expiry_persist(constraint):
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verify_with == "read the workflow's shell setting"
     assert row.expires_when == "the runner can be given a bash shell"
     assert row.verified_at is not None
@@ -79,7 +74,7 @@ async def test_an_empty_string_becomes_null_not_an_empty_column(constraint):
     await rulebooks_svc.update_rule(
         constraint["rule"], constraint["uid"], verify_with="", expires_when="",
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verify_with is None
     assert row.expires_when is None
 
@@ -89,7 +84,7 @@ async def test_naming_a_field_in_clear_empties_it(constraint):
     await rulebooks_svc.update_rule(
         constraint["rule"], constraint["uid"], clear=["verify_with"],
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verify_with is None
     # expires_when was NOT named, so it survives — clearing is per-field, and
     # a caller retiring one field must not lose the others.
@@ -107,7 +102,7 @@ async def test_rewording_the_check_drops_the_stamp(constraint):
         constraint["rule"], constraint["uid"],
         verify_with="read the runner's container shell, not the image's",
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verified_at is None
 
 
@@ -115,7 +110,7 @@ async def test_clearing_the_check_drops_the_stamp(constraint):
     await rulebooks_svc.update_rule(
         constraint["rule"], constraint["uid"], clear=["verify_with"],
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verified_at is None
 
 
@@ -132,7 +127,7 @@ async def test_editing_anything_else_leaves_the_stamp_alone(constraint):
             "applies to the build, not to `run:`.",
         expires_when="the runner grows a shell setting",
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verified_at is not None
     assert row.why.startswith("act_runner picks the shell")
 
@@ -221,11 +216,11 @@ async def test_a_failed_check_writes_nothing(rulebook_of_three):
     not in a special condition — it is WRONG. Recording the failure would let
     it sit there being false with the sweep satisfied that someone looked.
     """
-    before = await _row(rulebook_of_three["stale"])
+    before = await rule_row(rulebook_of_three["stale"])
     await rulebooks_svc.mark_rule_verified(
         rulebook_of_three["stale"], rulebook_of_three["uid"], still_true=False,
     )
-    after = await _row(rulebook_of_three["stale"])
+    after = await rule_row(rulebook_of_three["stale"])
     assert after.verified_at == before.verified_at
 
 

tests/test_integration_shape_classify.py

@@ -889,6 +889,91 @@ async def test_a_second_confirm_dialog_is_detected_and_named(seeded):
     assert total == 0
 
 
+@pytest.mark.integration
+async def test_a_conclusive_meaning_miss_silences_what_the_signature_cannot(seeded):
+    """#4208: the four false prompts #4204's form gate provably cannot reach.
+
+    THE FIXTURE IS THE ACCEPTANCE CASE ABOVE, DELIBERATELY. That is the whole
+    difficulty of this issue: a hand-rolled `confirmDanger` beside an async
+    confirm canon is structurally IDENTICAL to a registry helper beside an
+    async service canon — same family, same form contradiction, same directory
+    density. The form gate has to keep asking about both, so nothing derived
+    from a signature can separate them. The only difference is whether the
+    shape does the canon's JOB, and the only reading of that the ledger holds
+    is the proposer's per-symbol body comparison.
+
+    So the two runs differ in exactly one thing. In the test above the semantic
+    arm is quiet — it answers "nothing" without claiming to have looked — and
+    the prompt is RAISED, which is what milestone #2793 exists to produce. Here
+    it answers "I compared this body against the canons in its family and it is
+    none of them", and the prompt is WITHHELD. Holding the fixture identical is
+    what makes this a test of the meaning gate rather than of the setup.
+
+    Asserted on the stored basis as well as the outcome, so that a future
+    change which silences the prompt for some other reason fails here instead
+    of reading as a pass.
+    """
+    from datetime import datetime, timedelta, timezone
+    from unittest.mock import AsyncMock, patch
+
+    from scribe.services import shape_ledger
+    from scribe.services import snippets as snippets_svc
+    from scribe.services.shape_ledger import (
+        BASIS_NO_SEMANTIC_MATCH, flag_divergence, live_rows, propose_for_repo,
+    )
+
+    owner, pid = seeded["owner"], seeded["pid"]
+    canon = await snippets_svc.create_snippet(
+        owner, name="cls_confirm_factory_meaning",
+        code="export async function factory(): Promise<boolean> {\n  return true;\n}\n",
+        language="typescript", repo="Widget",
+        path="frontend/src/composables/useConfirm.ts", symbol="factory",
+        project_id=pid,
+    )
+    sid = int(canon.id)
+    comp = "frontend/src/components"
+    base = _defs(
+        *[(f"{comp}/{n}.vue", "sym", f"on{n}", f"async function on{n}() {{",
+           f"async function on{n}() {{\n  const ok = await factory();\n  if (!ok) return;\n}}")
+          for n in ("Trash", "Delete", "Remove", "Restore")],
+    )
+    await sync_repo_shapes(pid, REPO, base, seen_marker="aaa111")
+    await classify_shapes(owner, pid, [
+        {"path": f"{comp}/{n}.vue", "symbol": f"on{n}", "status": "instance", "snippet_id": sid}
+        for n in ("Trash", "Delete", "Remove", "Restore")
+    ], via="audit")
+    previous = datetime.now(timezone.utc)
+
+    later = base + _defs(
+        (f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
+         "function confirmDanger() {\n  return window.confirm('Really?');\n}"),
+    )
+    await sync_repo_shapes(pid, REPO, later, seen_marker="bbb222")
+
+    def _conclusive_miss(*_args, report=None, **_kw):
+        """The arm ran, compared, and found no canon — the one empty answer
+        that is evidence. `_semantic_canon` itself decides when it may say
+        this (a result set shorter than the limit); the unit tests for that
+        judgment are in tests/test_divergence_meaning_gate.py."""
+        if report is not None:
+            report["conclusive"] = True
+        return None
+
+    with patch.object(shape_ledger, "_semantic_canon",
+                      AsyncMock(side_effect=_conclusive_miss)):
+        await propose_for_repo(owner, pid, REPO, later)
+
+    rows = await live_rows(pid)
+    danger = next(r for r in rows if r.symbol == "confirmDanger")
+    assert danger.proposal_basis == BASIS_NO_SEMANTIC_MATCH
+    # The miss is not a proposal: nothing may read it as one.
+    assert danger.proposed_snippet_id is None
+
+    assert await flag_divergence(pid, since=previous - timedelta(seconds=1)) == 0
+    _, total = await list_project_shapes(owner, pid, flag="divergence")
+    assert total == 0, "a shape the proposer measured as unrelated must not be urged"
+
+
 @pytest.mark.integration
 async def test_history_records_what_was_used_when_and_drift_asks_for_a_recheck(seeded):
     from scribe.services.shape_ledger import shape_history

tests/test_lesson_usage_surface.py

@@ -25,6 +25,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from scribe.mcp.tools import lessons as lesson_tools
+from scribe.services import note_usage
 from scribe.services.note_usage import empty_usage
 
 
@@ -47,7 +48,7 @@ async def test_the_mcp_listing_carries_usage_for_every_row():
                      new=AsyncMock(return_value=(_rows(), 2))),
         patch.object(lesson_tools.access_svc, "label_shared_items",
                      new=AsyncMock(side_effect=lambda _uid, items: items)),
-        patch.object(lesson_tools, "usage_for_notes",
+        patch.object(note_usage, "usage_for_notes",
                      new=AsyncMock(return_value=used)),
     ):
         out = await lesson_tools.list_lessons()
@@ -71,7 +72,7 @@ async def test_the_listing_asks_for_usage_once_for_the_whole_page():
                      new=AsyncMock(return_value=(_rows(), 2))),
         patch.object(lesson_tools.access_svc, "label_shared_items",
                      new=AsyncMock(side_effect=lambda _uid, items: items)),
-        patch.object(lesson_tools, "usage_for_notes", new=reader),
+        patch.object(note_usage, "usage_for_notes", new=reader),
     ):
         await lesson_tools.list_lessons()
 
@@ -101,7 +102,7 @@ async def test_get_lesson_reads_the_count_before_recording_its_own_pull():
         patch.object(lesson_tools, "_to_dict", return_value={"id": 7}),
         patch.object(lesson_tools.access_svc, "describe_provenance",
                      new=AsyncMock(return_value={})),
-        patch.object(lesson_tools, "usage_for_notes", new=_usage_read),
+        patch.object(note_usage, "usage_for_notes", new=_usage_read),
         patch.object(lesson_tools, "record_pulled",
                      side_effect=lambda **_kw: order.append("pull")),
     ):
@@ -116,6 +117,13 @@ async def test_get_lesson_reads_the_count_before_recording_its_own_pull():
 
 # ── the REST door, on structure (rule 167) ───────────────────────────────────
 #
+# MOVED 2026-09-21 (#4230), not weakened. These used to look for
+# `usage_for_notes(` and `empty_usage()` in the route body. Both now live in
+# ONE seam, `note_usage.attach_usage`, which seven doors share — so the
+# zero-fill and the single aggregate are pinned once, against the seam, in
+# tests/test_usage_attach_seam.py. What stays HERE is what only this route can
+# get wrong: that it calls the seam at all, once, and before it records a pull.
+#
 # Its siblings in test_lesson_rest_door.py are source guards for the same
 # reason: the route is decorated and returns a Quart response, so driving it
 # means standing up the app. What matters here is reachable from the source
@@ -136,19 +144,17 @@ def test_the_rest_listing_reads_usage_once_for_the_page():
     """A per-row lookup would be N+1 by construction — the listing's own
     comment says so, and this is what makes that comment checkable."""
     src = _route_source("list_lessons_route")
-    assert src.count("usage_for_notes(") == 1
+    assert src.count("attach_usage(") == 1
-    # The one call is not inside the loop that assigns the rows.
+    assert "attach_usage(items)" in src, "the seam must get the page, not a row"
-    call = src.index("usage_for_notes(")
-    assign = src.index('["usage"]')
-    assert call < assign
 
 
 def test_every_rest_row_carries_the_key_even_at_zero():
     """"Never surfaced" is a state the UI renders; a missing field is not."""
     src = _route_source("list_lessons_route")
-    assert "empty_usage()" in src, (
+    assert "attach_usage(" in src
-        "a row with no recorded usage would come back without the key, and a "
+    assert 'it["usage"] =' not in src, (
-        "reader cannot tell that from a reporting failure"
+        "the route re-spells the attach by hand, so a row with no recorded "
+        "usage can come back without the key again"
     )
 
 
@@ -156,7 +162,7 @@ def test_the_rest_detail_door_also_reads_before_it_records():
     """Two doors that disagree about what the number counts are worse than
     one door that is wrong, because only one of them looks wrong."""
     src = _route_source("get_lesson_route")
-    assert src.index("usage_for_notes(") < src.index("record_pulled(")
+    assert src.index("attach_usage(") < src.index("record_pulled(")
 
 
 def test_these_guards_can_fail():
@@ -181,9 +187,14 @@ def test_the_detail_view_renders_the_badge_rather_than_respelling_it():
     assert "UsageBadge" in view
     assert "usage-tag" not in view, "re-spelled the chip instead of reusing it"
 
-    key = 'dead-weight-advice="'
+    # The sentence moved out of this template into the shared table (#4230,
-    start = view.index(key) + len(key)
+    # recorded on #3460). Assert BOTH halves, so the guard cannot pass by the
-    advice = view[start:view.index('"', start)]
+    # view pointing at an entry that no longer says the right thing.
+    assert "DEAD_WEIGHT_ADVICE.lesson" in view, "the view no longer reads the lesson advice"
+    table = (Path(__file__).resolve().parents[1]
+             / "frontend/src/utils/deadWeight.ts").read_text()
+    start = table.index("  lesson:")
+    advice = table[start:table.index(",\n", start)]
     assert "when_to_apply" in advice, (
         f"the dead-weight advice does not point at the trigger: {advice!r}"
     )

tests/test_rule_overlap_gate.py

@@ -0,0 +1,276 @@
+"""Rule and preference creates surface the records they overlap (#4134).
+
+Rules and preferences share one table and one ranking, so a preference that
+restates a rule splits one instruction in two — and when only the preference
+places, binding guidance arrives labelled optional. The title gate could not
+see it: a second record about the same moment is exactly the one written
+under a different title.
+
+These tests pin three things:
+
+- WHAT is compared: the new record's rule_document (the shape the corpus is
+  embedded as), against BOTH kinds, in the scope the new record will rank in.
+- THAT it advises rather than blocks. The measurement above
+  dedup._RULE_OVERLAP_FLOOR found distinct neighbours scoring as high as a
+  true restatement, so a block would refuse legitimate records.
+- WHERE it runs: all three create doors, before the create, so the new record
+  cannot match itself.
+
+`find_overlapping_rules` is bound here at import time, before conftest's
+autouse `_no_rule_overlap` replaces the module attribute, so the service tests
+exercise the real function and stub the search beneath it.
+"""
+from __future__ import annotations
+
+import ast
+import inspect
+import pathlib
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.services import dedup
+from scribe.services.dedup import (
+    RuleOverlap,
+    find_overlapping_rules,
+    overlap_response,
+)
+from tests.helpers import fake_rule
+from tests.helpers import plain_rule_detail as _plain_detail
+
+pytestmark = pytest.mark.usefixtures("_bind_user")
+
+SEARCH = "scribe.services.dedup.embeddings_svc.semantic_search_rules"
+TOOLS = "scribe.mcp.tools.rulebooks"
+
+# Long enough to clear _MIN_BODY_FOR_SEMANTIC in the rule_document shape.
+TRIGGER = (
+    "About to push a commit to dev and stop to ask whether pushing is allowed, "
+    "or ending a turn with a question about landing work that was already "
+    "committed on the branch the operator treats as home."
+)
+STATEMENT = "Push to dev after committing without asking first."
+
+
+# ── what is compared ──────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_the_query_is_the_embedded_document_shape():
+    search = AsyncMock(return_value=[])
+    with patch(SEARCH, search):
+        await find_overlapping_rules(7, "Push without asking", STATEMENT, TRIGGER)
+    query = search.call_args.args[1]
+    title, body = dedup.embeddings_svc.rule_document(
+        "Push without asking", STATEMENT, TRIGGER,
+    )
+    assert query == f"{title}\n\n{body}"
+    assert "When to apply:" in query
+
+
+@pytest.mark.asyncio
+async def test_both_kinds_are_searched():
+    """The harm is ACROSS kinds; a kind filter would hide exactly it."""
+    search = AsyncMock(return_value=[])
+    with patch(SEARCH, search):
+        await find_overlapping_rules(7, "t", STATEMENT, TRIGGER)
+    assert search.call_args.kwargs.get("kind") is None
+
+
+@pytest.mark.asyncio
+async def test_a_global_record_is_compared_with_every_rule_the_caller_owns():
+    search = AsyncMock(return_value=[])
+    with patch(SEARCH, search):
+        await find_overlapping_rules(7, "t", STATEMENT, TRIGGER)
+    kw = search.call_args.kwargs
+    assert kw["everywhere"] is True
+    assert kw["project_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_a_project_rule_is_compared_with_what_it_will_rank_against():
+    search = AsyncMock(return_value=[])
+    with patch(SEARCH, search):
+        await find_overlapping_rules(7, "t", STATEMENT, TRIGGER, project_id=5)
+    kw = search.call_args.kwargs
+    assert kw["project_id"] == 5
+    assert kw["everywhere"] is False
+
+
+@pytest.mark.asyncio
+async def test_the_floor_is_the_measured_one():
+    search = AsyncMock(return_value=[])
+    with patch(SEARCH, search):
+        await find_overlapping_rules(7, "t", STATEMENT, TRIGGER)
+    assert search.call_args.kwargs["threshold"] == dedup._RULE_OVERLAP_FLOOR
+
+
+def test_the_floor_sits_below_the_measured_restatement():
+    """A reworded duplicate of a real rule scored 0.850 (the measurement above
+    _RULE_OVERLAP_FLOOR). A floor at or above that misses the case the check
+    exists for; this fails if someone raises it there by analogy with the
+    blocking gates' 0.90+."""
+    assert dedup._RULE_OVERLAP_FLOOR < 0.85
+
+
+@pytest.mark.asyncio
+async def test_a_short_document_is_not_searched():
+    search = AsyncMock(return_value=[])
+    with patch(SEARCH, search):
+        out = await find_overlapping_rules(7, "t", "s", "when")
+    assert out == []
+    search.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_failed_search_lets_the_create_through():
+    with patch(SEARCH, AsyncMock(side_effect=RuntimeError("no embedder"))):
+        out = await find_overlapping_rules(7, "t", STATEMENT, TRIGGER)
+    assert out == []
+
+
+@pytest.mark.asyncio
+async def test_hits_carry_their_kind():
+    hits = [
+        (0.8504, fake_rule(id=1, title="`dev` is home", kind="rule")),
+        (0.81, fake_rule(id=9, title="Report pushes", kind="preference", project_id=3)),
+    ]
+    with patch(SEARCH, AsyncMock(return_value=hits)):
+        out = await find_overlapping_rules(7, "t", STATEMENT, TRIGGER)
+    assert out == [
+        RuleOverlap(1, "`dev` is home", "rule", None, 0.85),
+        RuleOverlap(9, "Report pushes", "preference", 3, 0.81),
+    ]
+
+
+# ── what it says ──────────────────────────────────────────────────────────
+
+
+def test_no_overlap_adds_nothing():
+    assert overlap_response([], "rule") == {}
+
+
+def test_the_note_names_the_record_to_read_and_both_outcomes():
+    out = overlap_response(
+        [RuleOverlap(1, "`dev` is home", "rule", None, 0.85)], "preference",
+    )
+    assert out["overlaps"] == [{
+        "id": 1, "title": "`dev` is home", "kind": "rule",
+        "project_id": None, "similarity": 0.85,
+    }]
+    note = out["overlap_note"]
+    assert "Created" in note                 # it did not block
+    assert "update_rule" in note             # the top match's own door
+    assert "delete this preference" in note  # the new record's kind
+    assert "keep both" in note               # distinct records are legitimate
+
+
+# ── where it runs ─────────────────────────────────────────────────────────
+
+
+def _rule_one_overlaps():
+    return [RuleOverlap(1, "`dev` is home", "rule", None, 0.85)]
+
+
+@pytest.mark.asyncio
+async def test_create_rule_checks_before_creating_and_reports():
+    order: list[str] = []
+    find = AsyncMock(side_effect=lambda *a, **k: order.append("find") or _rule_one_overlaps())
+    create = AsyncMock(side_effect=lambda **k: order.append("create") or fake_rule(id=50))
+    with patch(f"{TOOLS}.dedup_svc.find_duplicate_rule", AsyncMock(return_value=None)), \
+         patch(f"{TOOLS}.dedup_svc.find_overlapping_rules", find), \
+         patch(f"{TOOLS}.rulebooks_svc.create_rule", create), _plain_detail():
+        from scribe.mcp.tools.rulebooks import create_rule
+        out = await create_rule(
+            topic_id=10, title="Push without asking", statement=STATEMENT,
+            when_to_apply=TRIGGER,
+        )
+    assert order == ["find", "create"]
+    assert find.call_args.args == (7, "Push without asking", STATEMENT, TRIGGER)
+    assert out["id"] == 50
+    assert out["overlaps"][0]["id"] == 1
+
+
+@pytest.mark.asyncio
+async def test_create_project_rule_checks_in_its_project():
+    find = AsyncMock(return_value=_rule_one_overlaps())
+    with patch(f"{TOOLS}.dedup_svc.find_duplicate_rule", AsyncMock(return_value=None)), \
+         patch(f"{TOOLS}.dedup_svc.find_overlapping_rules", find), \
+         patch(f"{TOOLS}.rulebooks_svc.create_project_rule",
+               AsyncMock(return_value=fake_rule(id=51, project_id=5, topic_id=None))), \
+         _plain_detail():
+        from scribe.mcp.tools.rulebooks import create_project_rule
+        out = await create_project_rule(
+            project_id=5, title="Push without asking", statement=STATEMENT,
+            when_to_apply=TRIGGER,
+        )
+    assert find.call_args.kwargs == {"project_id": 5}
+    assert "overlap_note" in out
+
+
+@pytest.mark.asyncio
+async def test_create_preference_checks_and_names_itself_a_preference():
+    find = AsyncMock(return_value=_rule_one_overlaps())
+    with patch(f"{TOOLS}.dedup_svc.find_duplicate_rule", AsyncMock(return_value=None)), \
+         patch(f"{TOOLS}.dedup_svc.find_overlapping_rules", find), \
+         patch(f"{TOOLS}.rulebooks_svc.create_rule",
+               AsyncMock(return_value=fake_rule(id=52, kind="preference"))), \
+         _plain_detail():
+        from scribe.mcp.tools.rulebooks import create_preference
+        out = await create_preference(
+            topic_id=10, title="Push without asking", statement=STATEMENT,
+            when_to_apply=TRIGGER, arose_from_id=42,
+        )
+    find.assert_awaited_once()
+    assert "delete this preference" in out["overlap_note"]
+
+
+@pytest.mark.asyncio
+async def test_a_create_with_no_overlap_reads_as_before():
+    with patch(f"{TOOLS}.dedup_svc.find_duplicate_rule", AsyncMock(return_value=None)), \
+         patch(f"{TOOLS}.dedup_svc.find_overlapping_rules", AsyncMock(return_value=[])), \
+         patch(f"{TOOLS}.rulebooks_svc.create_rule",
+               AsyncMock(return_value=fake_rule(id=53))), _plain_detail():
+        from scribe.mcp.tools.rulebooks import create_rule
+        out = await create_rule(
+            topic_id=10, title="t", statement=STATEMENT, when_to_apply=TRIGGER,
+        )
+    assert "overlaps" not in out and "overlap_note" not in out
+
+
+CREATE_DOORS = ("create_rule", "create_project_rule", "create_preference")
+
+
+def test_every_rule_create_door_asks():
+    """Structural, so a fourth create door fails here rather than shipping
+    with the title gate alone. Keyed on the doors that call
+    find_duplicate_rule: any door gated by title must also be checked by
+    meaning, because the title gate is the one that cannot see this."""
+    root = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
+    tree = ast.parse((root / "mcp" / "tools" / "rulebooks.py").read_text())
+    title_gated, overlap_checked = set(), set()
+    for fn in tree.body:
+        if not isinstance(fn, ast.AsyncFunctionDef):
+            continue
+        for node in ast.walk(fn):
+            if isinstance(node, ast.Attribute):
+                if node.attr == "find_duplicate_rule":
+                    title_gated.add(fn.name)
+                elif node.attr == "find_overlapping_rules":
+                    overlap_checked.add(fn.name)
+    assert title_gated >= set(CREATE_DOORS), "registry drifted from the module"
+    assert title_gated <= overlap_checked, (
+        f"title-gated but never checked by meaning: "
+        f"{sorted(title_gated - overlap_checked)}"
+    )
+
+
+def test_the_stale_premise_is_gone():
+    """find_duplicate_rule claimed rules were not a semantic-retrieval surface
+    — false since rules were embedded, and the reason this gap survived."""
+    # Flattened: the docstring is hard-wrapped, and the claim straddled a
+    # line. Keyed on the CONCLUSION it drew, not the premise — the corrected
+    # docstring quotes the premise in order to retire it.
+    doc = " ".join((inspect.getdoc(dedup.find_duplicate_rule) or "").split())
+    assert "the right (and only) signal" not in doc
+    assert "find_overlapping_rules" in doc

tests/test_session_slippage_readout.py

@@ -32,13 +32,14 @@ from __future__ import annotations
 
 import json
 import os
-import shutil
 import subprocess
 import time
 from pathlib import Path
 
 import pytest
 
+from tests.helpers import need_tools
+
 ROOT = Path(__file__).resolve().parents[1]
 HOOKS = ROOT / "plugin" / "hooks"
 DEFS = HOOKS / "scribe_defs.sh"
@@ -47,14 +48,8 @@ RECORDER = HOOKS / "scribe_record_outcome.sh"
 HOOKS_JSON = HOOKS / "hooks.json"
 
 
-def _need(*tools):
-    for t in tools:
-        if shutil.which(t) is None:
-            pytest.skip(f"hook runtime tool {t!r} not installed")
-
-
 def sh(script: str) -> str:
-    _need("bash", "awk")
+    need_tools("bash", "awk")
     r = subprocess.run(
         ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
         capture_output=True, text=True, timeout=30,
@@ -173,7 +168,7 @@ def test_a_session_no_rule_touched_says_nothing_at_all(tmp_path):
 # ── The hook that carries it ──────────────────────────────────────────────
 
 def run_precompact(event: dict, tmpdir: Path) -> subprocess.CompletedProcess:
-    _need("bash")
+    need_tools("bash")
     env = dict(os.environ)
     env["TMPDIR"] = str(tmpdir)
     return subprocess.run(["bash", str(PRECOMPACT)], input=json.dumps(event),
@@ -210,7 +205,7 @@ def test_the_compaction_hook_never_emits_a_json_envelope(tmp_path):
 # ── The recorder that makes `acted` mean anything ─────────────────────────
 
 def run_recorder(event: dict, tmpdir: Path) -> subprocess.CompletedProcess:
-    _need("bash")
+    need_tools("bash")
     env = {"PATH": os.environ["PATH"], "HOME": str(tmpdir), "TMPDIR": str(tmpdir)}
     return subprocess.run(["bash", str(RECORDER)], input=json.dumps(event),
                           capture_output=True, text=True, timeout=30, env=env)
@@ -257,7 +252,7 @@ def test_the_recorder_is_registered_on_the_rule_outcome_tool():
 
 
 def test_the_recorder_is_shell_valid():
-    _need("bash")
+    need_tools("bash")
     subprocess.run(["bash", "-n", str(RECORDER)], check=True)
 
 

tests/test_usage_attach_seam.py

@@ -0,0 +1,205 @@
+"""One seam attaches `usage`, and every door that shows it uses that seam (#4230).
+
+WHAT WENT WRONG. `usage_for_notes` is named for notes and works on every note
+row. Yet the surfaced-vs-opened chip reached snippets and rules only: notes had
+it nowhere, and lessons had it collected but shown nowhere a person could
+reach, because the only lesson LIST in the UI is the Knowledge browse and that
+route never attached it.
+
+The cause was not any one missing line. Seven call sites carried their own copy
+of the same few lines — two REST lists, two REST details, two MCP lists, one
+MCP detail — and each read perfectly well on its own. Nobody was comparing
+them, so "which doors attach usage?" had no answer anywhere in the code. That
+is the same failure `test_system_tagging_door_parity.py` records for System
+tagging (#4249): whichever door nobody exercised for a kind is the one that
+never grew the feature, and a human reviewer does not reliably catch it because
+each door is only ever read alone.
+
+So this file asserts the PROPERTY, not the behaviour of one route: the attach
+logic exists once, and no door re-implements it. A kind added next month either
+goes through the seam or fails here.
+"""
+from __future__ import annotations
+
+import ast
+import pathlib
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.services.note_usage import attach_usage, empty_usage
+
+ROOT = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
+
+# The aggregate the seam is built around. Calling it from a door is the shape
+# this file exists to prevent — not because the call is wrong, but because
+# seven of them drift.
+AGGREGATE = "usage_for_notes"
+
+
+def _counts(surfaced: int = 5, pulled: int = 0) -> dict:
+    u = empty_usage()
+    u["surfaced_count"] = surfaced
+    u["pull_count"] = pulled
+    return u
+
+
+def _aggregate_returns(mapping: dict[int, dict]) -> AsyncMock:
+    return patch(
+        "scribe.services.note_usage.usage_for_notes",
+        AsyncMock(return_value=mapping),
+    )
+
+
+# ── the seam itself ───────────────────────────────────────────────────────
+
+
+async def test_every_row_gets_the_key_even_with_no_events() -> None:
+    """Zero-filled, never absent. The UI must not have to tell "no events"
+    from "no field" — and `UsageBadge` renders nothing below one surfacing, so
+    an un-surfaced record is quiet without the caller doing anything."""
+    rows = [{"id": 1}, {"id": 2}]
+    with _aggregate_returns({1: _counts(surfaced=3)}):
+        await attach_usage(rows)
+    assert rows[0]["usage"]["surfaced_count"] == 3
+    assert rows[1]["usage"] == empty_usage()
+
+
+async def test_one_aggregate_for_the_whole_page() -> None:
+    """The N+1 guard. A per-row lookup here would be N+1 by construction, which
+    is the one shape a list route must not have — and it is invisible in
+    review, because the per-row version reads more naturally."""
+    rows = [{"id": n} for n in range(25)]
+    mock = AsyncMock(return_value={})
+    with patch("scribe.services.note_usage.usage_for_notes", mock):
+        await attach_usage(rows)
+    assert mock.await_count == 1, "usage must be read once per page, not per row"
+    assert sorted(mock.await_args.args[0]) == list(range(25))
+
+
+async def test_a_detail_payload_is_just_a_one_row_list() -> None:
+    """The single-record doors share the seam rather than keeping a second
+    shape beside it. Two shapes for one job is how the seven copies started."""
+    data = {"id": 7, "title": "x"}
+    with _aggregate_returns({7: _counts(surfaced=9, pulled=2)}):
+        await attach_usage([data])
+    assert data["usage"]["pull_count"] == 2
+
+
+async def test_a_row_with_no_id_is_skipped_rather_than_failing_the_list() -> None:
+    """An unusable id is not a reason to 500 a page of otherwise fine rows."""
+    rows = [{"id": 1}, {"title": "no id here"}]
+    with _aggregate_returns({1: _counts()}):
+        await attach_usage(rows)
+    assert "usage" in rows[0]
+    assert "usage" not in rows[1]
+
+
+async def test_a_boolean_is_not_an_id() -> None:
+    """`int(True)` is 1, so a row carrying a flag under the key would silently
+    be credited with note #1's counts. A wrong chip is worse than no chip: it
+    reads as a measurement."""
+    rows = [{"id": True}]
+    with _aggregate_returns({1: _counts(surfaced=40)}):
+        await attach_usage(rows)
+    assert "usage" not in rows[0]
+
+
+async def test_a_string_id_still_resolves() -> None:
+    """Payload rows come from several serialisers; one of them handing back a
+    stringified id should not silently drop the chip."""
+    rows = [{"id": "12"}]
+    with _aggregate_returns({12: _counts(surfaced=4)}):
+        await attach_usage(rows)
+    assert rows[0]["usage"]["surfaced_count"] == 4
+
+
+@pytest.mark.parametrize("key", ["note_id", "record_id"])
+async def test_the_key_can_be_named(key: str) -> None:
+    rows = [{key: 3}]
+    with _aggregate_returns({3: _counts()}):
+        await attach_usage(rows, key=key)
+    assert "usage" in rows[0]
+
+
+async def test_an_empty_page_asks_nothing_and_breaks_nothing() -> None:
+    mock = AsyncMock(return_value={})
+    with patch("scribe.services.note_usage.usage_for_notes", mock):
+        await attach_usage([])
+    assert mock.await_args.args[0] == []
+
+
+# ── the property: one seam, and every door uses it ────────────────────────
+
+
+def _calls(tree: ast.Module) -> set[str]:
+    out = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Call):
+            fn = node.func
+            name = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", None)
+            if name:
+                out.add(name)
+    return out
+
+
+def _door_modules() -> list[pathlib.Path]:
+    return sorted(
+        [*(ROOT / "routes").glob("*.py"), *(ROOT / "mcp" / "tools").glob("*.py")]
+    )
+
+
+def test_no_door_calls_the_aggregate_directly() -> None:
+    """THE GUARD. Seven doors each called `usage_for_notes` and zero-filled by
+    hand; the eighth would have been `/api/knowledge`, and the chip would have
+    kept reaching some kinds and not others.
+
+    Keyed on the CALL, not on the text, so a module that merely names the
+    function in a comment explaining the seam is not a false positive — and a
+    hand-kept skip list, which would itself go stale, is not needed (rule 167).
+    """
+    offenders = []
+    for path in _door_modules():
+        if AGGREGATE in _calls(ast.parse(path.read_text())):
+            offenders.append(str(path.relative_to(ROOT.parent.parent)))
+    assert not offenders, (
+        f"these doors call {AGGREGATE}() themselves instead of attach_usage(); "
+        f"that is how the chip came to reach two record kinds out of four: "
+        f"{offenders}"
+    )
+
+
+# (module, the functions that return note-bearing payloads)
+#
+# Not a list of everything that COULD attach usage — a list of the doors that
+# demonstrably show it today. A door dropping its call silently is the exact
+# regression this pins.
+DOORS = [
+    ("routes/lessons.py", "list_lessons_route or get_lesson_route"),
+    ("routes/snippets.py", "list/get snippet routes"),
+    ("routes/knowledge.py", "list_knowledge — the only note & lesson list in the UI"),
+    ("mcp/tools/lessons.py", "list_lessons / get_lesson"),
+    ("mcp/tools/snippets.py", "list_snippets"),
+]
+
+
+@pytest.mark.parametrize(("module", "why"), DOORS)
+def test_every_door_that_shows_usage_goes_through_the_seam(module: str, why: str) -> None:
+    assert "attach_usage" in _calls(ast.parse((ROOT / module).read_text())), (
+        f"{module} no longer attaches usage ({why}). If that is deliberate, "
+        f"remove it from DOORS and say why; a door that silently stops "
+        f"attaching looks exactly like a corpus nobody uses."
+    )
+
+
+def test_the_knowledge_browse_is_covered_because_it_is_the_only_note_list() -> None:
+    """Pinned on its own, with the reason, because it is the non-obvious one.
+
+    `/api/lessons` already attached usage and it did not help: no view calls
+    it. `KnowledgeView` is the only list in the UI that renders notes and
+    lessons, so `/api/knowledge` is the only route through which those two
+    kinds can show the counter at all. Deleting this line would restore the
+    original bug while every other test here still passed.
+    """
+    assert any(m == "routes/knowledge.py" for m, _ in DOORS)
+    assert "attach_usage" in _calls(ast.parse((ROOT / "routes" / "knowledge.py").read_text()))

tests/test_write_path_design_arm.py

@@ -0,0 +1,211 @@
+"""The write-path design arm: a UI write is told which design system binds it (#4256).
+
+A design system binds like a rule, and before this it reached a session only
+through the session-start block — complete for a session that knows to ask,
+silent for one writing a component. These tests pin:
+
+- THE TRIGGER is the file, not a search: a UI path in a project that has a
+  design system. No vectors, no score, no slot taken from the ranked menu.
+- WHAT IT SAYS is an index — each inherited layer's section headings, with a
+  layer short enough to be a line (the leaf's departure) shown whole.
+- IT DOES NOT MOVE THE OTHER ARMS. A design-only write returns without
+  running the standing-rule arm, which is gated on there being prior art;
+  letting the design line into that gate would change the rule arm's call
+  distribution under the floor it was tuned against.
+- ONCE PER SESSION PER SYSTEM, on the hook's token-keyed channel.
+"""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services import plugin_context as pc
+from tests.helpers import writepath_cfg
+
+HOUSE = (
+    "## Aesthetic\n\nModern-mythic with restraint. " + "Long prose. " * 60
+    + "\n\n## Where the accent must NOT appear\n\nNot on buttons.\n\n"
+    "## Voice and tone\n\nPlain language for anything functional."
+)
+LEAF = "The accent appears on the wordmark and active navigation."
+
+
+def _design(ds_id=9):
+    return {
+        "id": ds_id, "title": "App", "description": "",
+        "inherits_from": ["House"],
+        "guidance": [
+            {"design_system_id": 1, "title": "House", "guidance": HOUSE},
+            {"design_system_id": ds_id, "title": "App", "guidance": LEAF},
+        ],
+        "token_count": 3, "token_groups": ["accent"],
+    }
+
+
+def _project(ds_id=9):
+    return MagicMock(id=2, title="App", design_system_id=ds_id)
+
+
+# ── the trigger ───────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("path,ui", [
+    ("frontend/src/views/NoteView.vue", True),
+    ("frontend/src/styles/components.css", True),
+    ("web/App.TSX", True),
+    ("templates/index.html", True),
+    ("src/scribe/services/dedup.py", False),
+    ("frontend/src/utils/deadWeight.ts", False),
+    ("README.md", False),
+    ("", False),
+])
+def test_ui_paths(path, ui):
+    """`.ts` is deliberately not UI: a utility module is logic, and firing on
+    it would put the design line in front of writes it says nothing about."""
+    assert pc.is_ui_path(path) is ui
+
+
+@pytest.mark.asyncio
+async def test_no_project_asks_nothing():
+    get = AsyncMock()
+    with patch.object(pc.projects_svc, "get_project", get):
+        assert await pc._design_arm(1, 0, "a/B.vue", set()) == ("", "")
+    get.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_non_ui_write_asks_nothing():
+    get = AsyncMock()
+    with patch.object(pc.projects_svc, "get_project", get):
+        assert await pc._design_arm(1, 2, "src/x.py", set()) == ("", "")
+    get.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_project_without_a_design_system_says_nothing():
+    ctx = AsyncMock()
+    with patch.object(pc.projects_svc, "get_project", AsyncMock(return_value=_project(None))), \
+         patch.object(pc.design_systems_svc, "design_context", ctx):
+        assert await pc._design_arm(1, 2, "a/B.vue", set()) == ("", "")
+    ctx.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_already_shown_this_session_is_not_fetched_again():
+    ctx = AsyncMock()
+    with patch.object(pc.projects_svc, "get_project", AsyncMock(return_value=_project())), \
+         patch.object(pc.design_systems_svc, "design_context", ctx):
+        out = await pc._design_arm(1, 2, "a/B.vue", {pc.design_key(9)})
+    assert out == ("", "")
+    ctx.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_an_unreadable_design_system_says_nothing():
+    with patch.object(pc.projects_svc, "get_project", AsyncMock(return_value=_project())), \
+         patch.object(pc.design_systems_svc, "design_context", AsyncMock(return_value=None)):
+        assert await pc._design_arm(1, 2, "a/B.vue", set()) == ("", "")
+
+
+@pytest.mark.asyncio
+async def test_a_failure_never_breaks_the_write():
+    with patch.object(pc.projects_svc, "get_project", AsyncMock(side_effect=RuntimeError)):
+        assert await pc._design_arm(1, 2, "a/B.vue", set()) == ("", "")
+
+
+# ── what it says ──────────────────────────────────────────────────────────
+
+
+def test_the_line_indexes_the_house_style_and_inlines_the_departure():
+    line = pc._design_line("a/B.vue", _design())
+    assert "App (id 9) (inherits House)" in line
+    assert "`get_design_system(9)` → `resolved_guidance`" in line
+    assert "`resolve_design_system(9)`" in line
+    # The long layer is named by its headings, not pasted.
+    assert "House covers Aesthetic · Where the accent must NOT appear · Voice and tone" in line
+    assert "Long prose." not in line
+    # The short layer is the app's own departure, and is shown whole.
+    assert f'App: "{LEAF}"' in line
+
+
+def test_the_line_stays_a_line():
+    """The index exists because the prose does not fit: resolved guidance
+    runs to thousands of characters. A line that grew back to that size
+    would be the prose again under another name."""
+    assert len(pc._design_line("a/B.vue", _design())) < 800
+
+
+def test_a_long_layer_with_no_headings_is_elided_not_pasted():
+    design = _design()
+    design["guidance"][0]["guidance"] = "Unheaded prose. " * 200
+    line = pc._design_line("a/B.vue", design)
+    assert len(line) < 1200
+    assert "House:" in line
+
+
+# ── in the hint ───────────────────────────────────────────────────────────
+
+
+def _quiet():
+    """Every other arm silent: nothing recorded, nothing similar."""
+    return [
+        patch.object(pc, "get_writepath_config", AsyncMock(return_value=writepath_cfg())),
+        patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))),
+        patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])),
+        patch.object(pc, "record_retrieval", MagicMock()),
+        patch.object(pc.projects_svc, "get_project", AsyncMock(return_value=_project())),
+        patch.object(pc.design_systems_svc, "design_context", AsyncMock(return_value=_design())),
+        patch.object(pc, "owner_names_for", AsyncMock(return_value={})),
+    ]
+
+
+async def _hint(patches, **kw):
+    import contextlib
+    with contextlib.ExitStack() as stack:
+        for p in patches:
+            stack.enter_context(p)
+        rules = stack.enter_context(
+            patch.object(pc, "semantic_search_rules", AsyncMock(return_value=[]))
+        )
+        out = await pc.build_write_path_hint(1, "frontend/src/B.vue", project_id=2, **kw)
+    return out, rules
+
+
+@pytest.mark.asyncio
+async def test_a_ui_write_with_no_prior_art_still_hears_the_design_system():
+    out, _ = await _hint(_quiet())
+    assert out["context"].startswith("> Design system binds `frontend/src/B.vue`")
+    assert out["derive_keys"] == [pc.design_key(9)]
+    assert out["note_ids"] == [] and out["rule_ids"] == []
+
+
+@pytest.mark.asyncio
+async def test_a_design_only_write_does_not_switch_the_rule_arm_on():
+    """The rule arm runs only where there is prior art. A design line that
+    joined that gate would start a semantic rule search on every UI write —
+    a new population of calls under a floor tuned without them."""
+    _, rules = await _hint(_quiet())
+    rules.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_shown_once_per_session():
+    out, _ = await _hint(_quiet(), exclude_derive=[pc.design_key(9)])
+    assert out["context"] == ""
+    assert out["derive_keys"] == []
+
+
+@pytest.mark.asyncio
+async def test_beside_prior_art_it_leads_and_rides_the_keyed_channel():
+    patches = _quiet()
+    patches[1] = patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=(
+        [{"id": 5, "title": "fs-button", "user_id": 1, "note_type": "snippet"}], 1,
+    )))
+    out, _ = await _hint(patches)
+    lines = out["context"].splitlines()
+    assert lines[0].startswith("> Design system binds")
+    assert any("fs-button" in ln for ln in lines[1:])
+    assert pc.design_key(9) in out["derive_keys"]
+    # It takes no menu slot: the snippet is still reported as surfaced.
+    assert 5 in out["note_ids"]