diff --git a/frontend/src/components/rules/RuleListPane.vue b/frontend/src/components/rules/RuleListPane.vue
index 87a8311d..82f63f64 100644
--- a/frontend/src/components/rules/RuleListPane.vue
+++ b/frontend/src/components/rules/RuleListPane.vue
@@ -1,16 +1,7 @@
@@ -431,7 +424,7 @@ const SNIPPET_DEAD_WEIGHT =
{{ driftBadge(s) }}
-
+
by {{ s.owner ?? "another user" }}
diff --git a/src/scribe/mcp/tools/lessons.py b/src/scribe/mcp/tools/lessons.py
index 0e507aca..24d46c91 100644
--- a/src/scribe/mcp/tools/lessons.py
+++ b/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(
- int(note.id), empty_usage()
- )
+ await attach_usage([out])
record_pulled(
user_id=uid, note_id=int(note.id),
source="mcp_get_lesson", project_id=project_id,
diff --git a/src/scribe/mcp/tools/rulebooks.py b/src/scribe/mcp/tools/rulebooks.py
index 35a61bc0..7d44473f 100644
--- a/src/scribe/mcp/tools/rulebooks.py
+++ b/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(
diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py
index 5824d6ba..f9a0b47d 100644
--- a/src/scribe/mcp/tools/snippets.py
+++ b/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])
- for it in labeled:
- it["usage"] = usage.get(int(it["id"]), empty_usage())
+ await attach_usage(labeled)
return {"snippets": labeled, "total": total}
diff --git a/src/scribe/routes/knowledge.py b/src/scribe/routes/knowledge.py
index b30ce936..c7c79d3f 100644
--- a/src/scribe/routes/knowledge.py
+++ b/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,
diff --git a/src/scribe/routes/lessons.py b/src/scribe/routes/lessons.py
index e7e19d6c..91377964 100644
--- a/src/scribe/routes/lessons.py
+++ b/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])
- for it in items:
- it["usage"] = usage.get(int(it["id"]), empty_usage())
+ await attach_usage(items)
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(
- lesson_id, empty_usage()
- )
+ await attach_usage([out])
# 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).
diff --git a/src/scribe/routes/snippets.py b/src/scribe/routes/snippets.py
index ff286995..585d803a 100644
--- a/src/scribe/routes/snippets.py
+++ b/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])
- for it in items:
- it["usage"] = usage.get(int(it["id"]), empty_usage())
+ await attach_usage(items)
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(
- snippet_id, empty_usage()
- )
+ await attach_usage([data])
# 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).
diff --git a/src/scribe/services/dedup.py b/src/scribe/services/dedup.py
index 24bb1849..90101e63 100644
--- a/src/scribe/services/dedup.py
+++ b/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
- rule) or the same project (a project rule). Rules aren't a semantic-retrieval
- surface, so a normalized-title match is the right (and only) signal. Fail-open
- like find_duplicate_note."""
+ """Title-identical rule in the same topic (a rulebook rule) or the same
+ project (a project rule) — the one signal certain enough to BLOCK on.
+ Fail-open 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
diff --git a/src/scribe/services/note_usage.py b/src/scribe/services/note_usage.py
index 5b0400ed..cdada2b2 100644
--- a/src/scribe/services/note_usage.py
+++ b/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())
diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py
index 43bc94cd..7b6665dd 100644
--- a/src/scribe/services/plugin_context.py
+++ b/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:`). 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."""
diff --git a/src/scribe/services/shape_ledger.py b/src/scribe/services/shape_ledger.py
index 47845416..bff44ae5 100644
--- a/src/scribe/services/shape_ledger.py
+++ b/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
diff --git a/tests/conftest.py b/tests/conftest.py
index 24db63d6..3058aa3b 100644
--- a/tests/conftest.py
+++ b/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
diff --git a/tests/helpers.py b/tests/helpers.py
index 8bc80f1c..f3b60903 100644
--- a/tests/helpers.py
+++ b/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)
diff --git a/tests/test_contract_around_the_change.py b/tests/test_contract_around_the_change.py
index 839ae695..919b3b1d 100644
--- a/tests/test_contract_around_the_change.py
+++ b/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)
diff --git a/tests/test_divergence_meaning_gate.py b/tests/test_divergence_meaning_gate.py
new file mode 100644
index 00000000..0e18e28a
--- /dev/null
+++ b/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
diff --git a/tests/test_hook_json_reader.py b/tests/test_hook_json_reader.py
index fcdc371c..c94b16c7 100644
--- a/tests/test_hook_json_reader.py
+++ b/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)
diff --git a/tests/test_integration_rule_move.py b/tests/test_integration_rule_move.py
index d37c4f75..93ef55b9 100644
--- a/tests/test_integration_rule_move.py
+++ b/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"])
diff --git a/tests/test_integration_rule_verification.py b/tests/test_integration_rule_verification.py
index 9e94f523..2c3e01ea 100644
--- a/tests/test_integration_rule_verification.py
+++ b/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
diff --git a/tests/test_integration_shape_classify.py b/tests/test_integration_shape_classify.py
index 034cfdc5..d7943c62 100644
--- a/tests/test_integration_shape_classify.py
+++ b/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 {\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
diff --git a/tests/test_lesson_usage_surface.py b/tests/test_lesson_usage_surface.py
index 27894429..0b4c3a97 100644
--- a/tests/test_lesson_usage_surface.py
+++ b/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
- # The one call is not inside the loop that assigns the rows.
- call = src.index("usage_for_notes(")
- assign = src.index('["usage"]')
- assert call < assign
+ assert src.count("attach_usage(") == 1
+ assert "attach_usage(items)" in src, "the seam must get the page, not a row"
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, (
- "a row with no recorded usage would come back without the key, and a "
- "reader cannot tell that from a reporting failure"
+ assert "attach_usage(" in src
+ assert 'it["usage"] =' not in src, (
+ "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="'
- start = view.index(key) + len(key)
- advice = view[start:view.index('"', start)]
+ # The sentence moved out of this template into the shared table (#4230,
+ # recorded on #3460). Assert BOTH halves, so the guard cannot pass by the
+ # 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}"
)
diff --git a/tests/test_rule_overlap_gate.py b/tests/test_rule_overlap_gate.py
new file mode 100644
index 00000000..5ca8ffc3
--- /dev/null
+++ b/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
diff --git a/tests/test_session_slippage_readout.py b/tests/test_session_slippage_readout.py
index 73b656d1..29f0a151 100644
--- a/tests/test_session_slippage_readout.py
+++ b/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)
diff --git a/tests/test_usage_attach_seam.py b/tests/test_usage_attach_seam.py
new file mode 100644
index 00000000..c5d936d9
--- /dev/null
+++ b/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()))
diff --git a/tests/test_write_path_design_arm.py b/tests/test_write_path_design_arm.py
new file mode 100644
index 00000000..85a8faf8
--- /dev/null
+++ b/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"]