The prompt-boundary rule arm, and a slot a preference cannot lose (#3852, #3894) #151

Merged
bvandeusen merged 2 commits from dev into main 2026-09-11 11:12:00 -04:00
7 changed files with 735 additions and 10 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
{ {
"name": "scribe", "name": "scribe",
"description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.", "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
"version": "2026.09.11.0319", "version": "2026.09.11.1154",
"author": { "author": {
"name": "Bryan Van Deusen" "name": "Bryan Van Deusen"
}, },
+36
View File
@@ -7,6 +7,18 @@
# only — never bodies; the agent calls get_note(id) to pull anything it judges # only — never bodies; the agent calls get_note(id) to pull anything it judges
# relevant. Most turns inject nothing. # relevant. Most turns inject nothing.
# #
# TWO ARMS SINCE #3852, on one request. Rules and preferences are retrieved
# against the same prompt and returned in the same payload, ahead of the notes
# menu. That arm exists because the two act arms are keyed on a file write or
# a command, so a rule governing what to SAY — extract intent from loose
# phrasing, raise a conflict before acting, end a finding with an offer — had
# no moment to fire at. The operator's message is the only query that exists
# before a response is composed.
#
# The two arms are gated separately server-side: turning the notes menu off
# leaves rules arriving, because they are different claims with different
# costs of being missed.
#
# Best-effort enrichment ONLY: unlike the SessionStart channel there is no # Best-effort enrichment ONLY: unlike the SessionStart channel there is no
# static floor here. If the instance is unconfigured/unreachable, or anything # static floor here. If the instance is unconfigured/unreachable, or anything
# fails, the hook stays SILENT and exits 0 — it must never block a prompt. # fails, the hook stays SILENT and exits 0 — it must never block a prompt.
@@ -65,16 +77,32 @@ fi
# Per-session dedup: ids already injected this session are skipped. # Per-session dedup: ids already injected this session are skipped.
state_dir="${TMPDIR:-/tmp}/scribe-autoinject" state_dir="${TMPDIR:-/tmp}/scribe-autoinject"
mkdir -p "$state_dir" 2>/dev/null || true mkdir -p "$state_dir" 2>/dev/null || true
# RULES DEDUP IN A DIFFERENT DIRECTORY, and it has to be this one. The rule
# ledger is SHARED by every arm that can name a rule — the two PreToolUse
# hooks already keep it under scribe-priorart — so that one session keeps ONE
# list and a rule named here is not re-announced before the next Bash call.
# A private copy here would make each arm's "already seen" mean something
# different, which is the state #3749/#3750 exist to keep coherent. The
# directory name is the prior-art hook's history, not a scope claim.
rule_state_dir="${TMPDIR:-/tmp}/scribe-priorart"
mkdir -p "$rule_state_dir" 2>/dev/null || true
idfile="" idfile=""
rulefile=""
exclude_q="" exclude_q=""
if [ -n "$session_id" ]; then if [ -n "$session_id" ]; then
# session_id is an opaque token from Claude Code; keep only filename-safe chars. # session_id is an opaque token from Claude Code; keep only filename-safe chars.
safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_') safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
idfile="$state_dir/${safe_sid}.ids" idfile="$state_dir/${safe_sid}.ids"
rulefile="$rule_state_dir/${safe_sid}.rules.ids"
if [ -f "$idfile" ]; then if [ -f "$idfile" ]; then
seen=$(tr '\n' ',' < "$idfile" 2>/dev/null | sed 's/,$//') seen=$(tr '\n' ',' < "$idfile" 2>/dev/null | sed 's/,$//')
[ -n "$seen" ] && exclude_q="&exclude_ids=${seen}" [ -n "$seen" ] && exclude_q="&exclude_ids=${seen}"
fi fi
# AGED, not read flat: an exclusion that never expires means a rule surfaced
# once in a long session is silenced for the rest of it, even as the session
# stops holding what it was told. scribe_rules_live carries the reasoning.
rule_seen=$(scribe_rules_live "$rulefile")
[ -n "$rule_seen" ] && exclude_q="${exclude_q}&exclude_rule_ids=${rule_seen}"
fi fi
body=$(curl -fsS --max-time 5 \ body=$(curl -fsS --max-time 5 \
@@ -89,6 +117,14 @@ context=$(printf '%s' "$body" | jq -r '.context // empty' 2>/dev/null) || exit 0
if [ -n "$idfile" ]; then if [ -n "$idfile" ]; then
printf '%s' "$body" | jq -r '.note_ids[]? // empty' 2>/dev/null >> "$idfile" || true printf '%s' "$body" | jq -r '.note_ids[]? // empty' 2>/dev/null >> "$idfile" || true
fi fi
# Rules onto the SHARED ledger, stamped so they can age out. Only FRESH ids
# come back in rule_ids (#3752) — a rule rendered as a repeat is already on
# the ledger, and re-appending it would keep pushing its stamp forward so it
# never aged at all.
if [ -n "$rulefile" ]; then
printf '%s' "$body" | jq -r '.rule_ids[]? // empty' 2>/dev/null \
| scribe_rules_append "$rulefile"
fi
jq -n --arg c "$context" \ jq -n --arg c "$context" \
'{hookSpecificOutput: {hookEventName: "UserPromptSubmit", additionalContext: $c}}' '{hookSpecificOutput: {hookEventName: "UserPromptSubmit", additionalContext: $c}}'
+24
View File
@@ -91,13 +91,37 @@ async def autoinject_retrieve():
project_id (opt) — explicit project scope override (ad-hoc/testing). project_id (opt) — explicit project scope override (ad-hoc/testing).
exclude_ids (opt) — comma-separated note ids already injected this exclude_ids (opt) — comma-separated note ids already injected this
session; skipped so each note injects at most once. session; skipped so each note injects at most once.
exclude_rule_ids — comma-separated rule ids already surfaced this
(opt) session. SHARED with /prior-art and /tool-rules on
purpose: one session keeps ONE rule ledger, so a
rule named by any arm is not re-announced by
another. Ages out (#3751), so salience decays.
TWO ARMS, TWO SETS OF GATES. Rules ride the same hook and the same query
but nothing else: the notes menu can be disabled, thresholded and top-k'd
by the operator without touching whether a rule reaches them. Composed
here rather than inside either builder so neither one's early return can
silently suppress the other.
Rules come FIRST in the payload. A rule or preference governing the answer
is more consequential than a menu of things that might be worth reading,
and a reader who stops after the first block should have stopped after
the right one.
""" """
q = (request.args.get("q") or "").strip() q = (request.args.get("q") or "").strip()
project_id, _repo, _unbound = await _project_scope() project_id, _repo, _unbound = await _project_scope()
exclude_ids = _int_list(request.args.get("exclude_ids")) exclude_ids = _int_list(request.args.get("exclude_ids"))
exclude_rule_ids = _int_list(request.args.get("exclude_rule_ids"))
rules = await plugin_ctx_svc.build_prompt_rule_hint(
g.user.id, q, project_id=project_id, exclude_rule_ids=exclude_rule_ids
)
result = await plugin_ctx_svc.build_autoinject_hint( result = await plugin_ctx_svc.build_autoinject_hint(
g.user.id, q, project_id=project_id, exclude_ids=exclude_ids g.user.id, q, project_id=project_id, exclude_ids=exclude_ids
) )
blocks = [b for b in (rules["context"], result["context"]) if b]
result["context"] = "\n\n".join(blocks)
result["rule_ids"] = rules["rule_ids"]
return jsonify(result) return jsonify(result)
+12
View File
@@ -818,6 +818,7 @@ async def semantic_search_rules(
limit: int = 5, limit: int = 5,
threshold: float = _SIMILARITY_THRESHOLD, threshold: float = _SIMILARITY_THRESHOLD,
tier: str | None = None, tier: str | None = None,
kind: str | None = None,
report: dict | None = None, report: dict | None = None,
) -> list[tuple[float, "Rule"]]: ) -> list[tuple[float, "Rule"]]:
"""Return up to *limit* (score, rule) pairs most relevant to *query*. """Return up to *limit* (score, rule) pairs most relevant to *query*.
@@ -855,6 +856,16 @@ async def semantic_search_rules(
Pass a tier when a caller genuinely wants one class — a listing, an audit, Pass a tier when a caller genuinely wants one class — a listing, an audit,
a UI that renders the tiers apart. Not to approximate relevance. a UI that renders the tiers apart. Not to approximate relevance.
`kind` narrows to `rule` or `preference`, and NONE is likewise the ordinary
case: a caller asking "what governs this" wants both, because the reader
needs to know what binds AND how the operator wants it done. The one place
it is passed is a RESERVED SLOT — a query that may only return a
preference, so the slot cannot be spent on something else. That is the
same reason `note_type` exists on the sibling search, and the same failure
it prevents: a slot silently filled by the wrong kind is worse than no
slot, because the line is indistinguishable from one that earned its place
on score.
Collapses to best-chunk-per-rule like the note search, so a long rule split Collapses to best-chunk-per-rule like the note search, so a long rule split
across chunks competes once rather than crowding the results with itself. across chunks competes once rather than crowding the results with itself.
@@ -895,6 +906,7 @@ async def semantic_search_rules(
Project.user_id == user_id, Project.user_id == user_id,
), ),
*( [Rule.tier == tier] if tier else [] ), *( [Rule.tier == tier] if tier else [] ),
*( [Rule.kind == kind] if kind else [] ),
) )
# Overfetch so collapsing chunks to their best row still fills # Overfetch so collapsing chunks to their best row still fills
# the page — the same reason the note search overfetches. # the page — the same reason the note search overfetches.
+265
View File
@@ -314,6 +314,42 @@ _AUTOINJECT_BAND = 0.10
# awareness menu (titles only), never a content dump. # awareness menu (titles only), never a content dump.
_AUTOINJECT_MAX_TOP_K = 10 _AUTOINJECT_MAX_TOP_K = 10
# --- the prompt-boundary rule arm (#3852) ------------------------------------
#
# Both existing rule arms are keyed on something the session is about to DO —
# a file write, a command. A rule that governs what to SAY has no such moment.
# Extract intent from loose phrasing, raise a conflict before acting, hand off
# an action with its reason, end a finding with an offer: every one binds on a
# RESPONSE, and no tool call precedes a response.
#
# The operator's message is the only query that exists before one is composed,
# and this arm is what runs against it. Until now that hook searched notes
# alone, so no rule had ever been retrieved against a thing the operator said.
PROMPTRULE_THRESHOLD_KEY = "kb_promptrule_threshold"
# INHERITED FROM THE ACT ARMS, AND NOT YET EARNED HERE. 0.72 was tuned against
# code and shell commands. An operator's prose is a different query shape
# against the same documents, and nothing yet says the two distributions line
# up — triggers are written in the vocabulary of the MOMENT, which for most
# rules is act vocabulary, so prose may well score lower across the board.
#
# Starting at the act arms' number anyway is deliberate: it is the only value
# with evidence behind it, and guessing lower would put an unmeasured bar in
# front of a corpus that binds. Every call is logged under `prompt_rule` from
# the first deploy, so a few days of real traffic settles it — read
# `near_miss_samples` (#3807) before moving this, not the percentile alone.
PROMPTRULE_DEFAULT_THRESHOLD = 0.72
# MORE THAN THE ACT ARMS' SINGLE SLOT, anchored on this hook's budget rather
# than theirs. RULEHINT_LIMIT is 1 because that arm fires before EVERY Bash
# call, where a second line is a second interruption per command. This arm
# fires once per TURN, on the same hook whose notes menu already spends
# AUTOINJECT_DEFAULT_TOP_K slots — so that is the comparable budget.
#
# And a prompt genuinely contains more than one act. "Merge to main and then
# start on X" is two, governed by different rules; k=1 cannot serve that case
# at all, where the act arms never face it because a command is one thing.
PROMPTRULE_LIMIT = 3
def _slugify(text: str) -> str: def _slugify(text: str) -> str:
"""kebab-case slug for a skill directory name (a-z0-9 + single hyphens).""" """kebab-case slug for a skill directory name (a-z0-9 + single hyphens)."""
@@ -642,6 +678,235 @@ async def build_autoinject_hint(
return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg} return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
async def _reserve_slot_for_preference(
user_id: int,
query: str,
hits: list,
*,
threshold: float,
project_id: int,
already: set[int],
) -> tuple[list, int | None]:
"""Guarantee a preference one slot, if one clears the bar (#3894).
THE ASYMMETRY THIS EXISTS FOR. A rule and a preference are not equally
served by a shared score contest, because their losses are not equal:
- a RULE crowded out here can still fire at the act arm. A `git push`
reaches `pre_tool_rule`, a file write reaches `write_path_rule`. The
prompt hit is a preview of a second chance.
- a PREFERENCE about how to answer has no second chance. There is no
later act — the response IS the act — so crowded out here it is never
delivered at all.
A straight ranking therefore favours the record whose loss is recoverable
over the one whose loss is total, and it does so INVISIBLY: the rule that
won is a legitimate hit, the telemetry looks healthy, and the only symptom
is a preference that quietly never arrives. `reuse_slot` exists for the
same shape one corpus over (#2463), where snippets kept losing to project
records that merely resembled the query.
THE SLOT BUYS POSITION, NOT A LOWER BAR — same as `reuse_slot`, which also
reserves at `cfg["threshold"]`. A weak preference cannot buy the slot, so
silence stays the default and the reserved line is never worse than the
ones it sits beside. If `preference_slot` later shows a stream of
near-misses, `best_available_id` (#3807) names which preference was
refused and a separate bar becomes an argument with evidence behind it
rather than a knob added on a guess.
LEDGER REPEATS STILL COUNT AS REPRESENTED. A preference already on the
session's ledger occupies the slot rather than being skipped for a fresh
one: it is still rendered (#3750), just with the tail that says so, and a
preference is the kind of record where being reminded is the point.
Returns the possibly-extended hit list, and the id the slot spent — the
caller needs that to keep each source's surfaced set matching its own log
row (#3668), since the slot logs under its own name.
"""
if any(rule.kind == "preference" for _s, rule in hits):
return hits, None
_t0 = time.perf_counter()
_rep: dict = {}
# KIND-FILTERED, so the query can only answer with what the slot is for.
# Verifying the kind afterwards would be weaker: an unfiltered search that
# happened to return a rule would spend the slot on it, and the line would
# be indistinguishable from one that earned its place.
found = await semantic_search_rules(
user_id, query, limit=1, threshold=threshold,
kind="preference", report=_rep,
)
fresh = [(s, r) for s, r in found if r.id not in already]
# ITS OWN SOURCE, and both sides of the trade logged. #2463's own finding
# is the warning rather than the precedent here: the hit that slot pushed
# OUT was in retrieval_logs while the query that pushed it out was not, so
# the slot could never be judged against what it displaced. `results` is
# fresh-only, matching what gets recorded as surfaced below (#3752/#3668).
record_retrieval(
user_id=user_id, source="preference_slot", query=query,
threshold=threshold, limit=1, project_id=project_id,
is_task=None, results=fresh,
best_available=_rep.get("best_available_score"),
best_available_id=_rep.get("best_available_id"),
searched=bool(_rep.get("searched", True)),
suppressed=len(found) - len(fresh),
duration_ms=(time.perf_counter() - _t0) * 1000.0,
)
seen = {rule.id for _s, rule in hits}
slot = [(s, r) for s, r in found
if r.kind == "preference" and r.id not in seen][:1]
if not slot:
return hits, None
slot_id = int(slot[0][1].id)
if slot_id not in already:
record_rule_surfaced(
user_id=user_id, rule_ids=[slot_id], source="preference_slot",
)
# IT EXTENDS, IT NEVER DISPLACES — and here it parts company with
# `reuse_slot`, which evicts its menu's weakest hit. The reason is the
# ledger rather than taste. A displaced hit was RETURNED by the general
# search and is sitting in that call's `retrieval_logs` row, but would not
# have been shown — so `prompt_rule`'s surfaced set would stop matching
# its own log row, and #3668's identity would break for a reason nothing
# in the data explains. That identity is the cheapest true statement
# available about this pair of tables, and milestone #379 is what it costs
# to lose it: five steps planned against a gap that was two counters
# disagreeing, not a write path dropping rows.
#
# The price is one extra line, only when the general search already filled
# the limit AND a preference cleared the bar without placing. Cheap, and
# it buys a surface whose two tables can always be checked against each
# other.
return hits + slot, slot_id
async def build_prompt_rule_hint(
user_id: int,
query: str,
*,
project_id: int = 0,
exclude_rule_ids: list[int] | None = None,
) -> dict:
"""Rules and preferences that may apply to what the operator just asked.
The third rule arm, and the one that closes a gap the other two cannot
reach. `write_path_rule` is keyed on code, `pre_tool_rule` on a command —
both are things the session is about to DO. A rule that governs what to
SAY has no such trigger, and residency was the only surface it ever had.
Removing residency (milestone 394) without this would drop that half of
the corpus on the floor.
A SEPARATE FUNCTION, not a branch inside build_autoinject_hint, and the
reason is its early returns. That arm bails when auto-inject is disabled,
when the query is blank, when nothing clears the note bar — and every one
of those is a statement about NOTES. Folded in, a user who turned the
notes menu off would silently lose their rules too, which is the kind of
coupling nothing downstream could see. Two functions, two sets of gates,
composed by the caller.
THE OUTPUT IS DELIBERATELY NOT QUOTED, where the notes menu is. The task
asked whether the two share a header; the answer is that neither needs
one. A note line is a bare title and needs the menu's header to say what
it is doing there, while a rule line names itself in its opening words
("Standing rule that may apply…" / "Preference that may apply…"). Leaving
rules unquoted separates the two claims visually with no extra prose, and
matches how a rule line already renders on both act arms.
Fails open and returns empty context on any error, like its siblings: a
recall aid may never break the operator's prompt.
"""
out: dict = {"context": "", "rule_ids": []}
q = (query or "").strip()
if not q:
return out
try:
try:
threshold = float(await get_setting(
user_id, PROMPTRULE_THRESHOLD_KEY,
str(PROMPTRULE_DEFAULT_THRESHOLD)))
except (TypeError, ValueError):
threshold = PROMPTRULE_DEFAULT_THRESHOLD
threshold = min(1.0, max(0.0, threshold))
t0 = time.perf_counter()
_rep: dict = {}
# NOT scoped to the project, and that is the corpus's own decision
# rather than an omission here — semantic_search_rules is scoped by
# OWNERSHIP on purpose, because "is there a rule about this" is asked
# across a whole rulebook. `project_id` below reaches the log row and
# nothing else.
hits = await semantic_search_rules(
user_id, q, limit=PROMPTRULE_LIMIT, threshold=threshold,
report=_rep,
)
duration_ms = (time.perf_counter() - t0) * 1000.0
already = set(exclude_rule_ids or [])
fresh = [(score, rule) for score, rule in hits if rule.id not in already]
# BEFORE the early return, for the reason both sibling arms spell out
# at length: a call that found nothing is the only evidence a bar is
# too high, and an arm that logs only the calls it liked reports a
# flawless clear-rate however badly it is tuned. This bar is inherited
# and unverified for this corpus, so the zero rows are the point.
record_retrieval(
user_id=user_id, source="prompt_rule", query=q,
threshold=threshold, limit=PROMPTRULE_LIMIT,
project_id=project_id,
is_task=None, results=fresh, duration_ms=duration_ms,
best_available=_rep.get("best_available_score"),
best_available_id=_rep.get("best_available_id"),
searched=bool(_rep.get("searched", True)),
suppressed=len(hits) - len(fresh),
)
# THE RESERVED SLOT RUNS BEFORE THE BAIL-OUT, and that ordering is
# load-bearing rather than tidy. An empty general result is not proof
# that no preference qualifies: the general search overfetches by
# distance and then collapses, so a preference ranked below that
# window is invisible to it while a kind-filtered query finds it at
# once. Bailing first would make the slot dead in exactly the corpus
# it exists for — one where rules outnumber preferences.
hits, slot_id = await _reserve_slot_for_preference(
user_id, q, hits, threshold=threshold,
project_id=project_id, already=already,
)
# `hits`, not `fresh` (#3750): a call whose only hit is a repeat still
# has something to say, it just says it differently.
if not hits:
return out
lines = [
_rule_hint_line(rule, where="to this request", seen=rule.id in already)
for _score, rule in hits
]
# FRESH-ONLY (#3752). A reference is a rendering decision, not a
# retrieval outcome, and counting one here would inflate the
# denominator pull_through is read from.
#
# `fresh` is the PRE-SLOT list on purpose: it is exactly what this
# call's own `retrieval_logs` row recorded, so the two stay equal
# (#3668). The slot's hit is surfaced under `preference_slot` by the
# helper, against that source's own row.
rule_ids = [rule.id for _score, rule in fresh]
# RANKED, not ambient: this arm chose what it showed. The name is also
# in `rule_usage.RANKED_SOURCES`, and it has to be — a ranked source
# missing from that tuple is counted as a bulk delivery nobody decided
# on, which silently moves it out of the pull-through denominator.
if rule_ids:
record_rule_surfaced(
user_id=user_id, rule_ids=rule_ids, source="prompt_rule",
)
out["context"] = "\n".join(lines)
out["rule_ids"] = rule_ids
except Exception:
logger.debug("prompt rule arm failed", exc_info=True)
return out
# --- Write-path trigger (#2082): prior art at the moment code is written ------ # --- Write-path trigger (#2082): prior art at the moment code is written ------
# Auto-inject above fires on the operator's prompt. The moment reuse is actually # Auto-inject above fires on the operator's prompt. The moment reuse is actually
# lost is later — when the AGENT decides mid-task to write a helper — and nothing # lost is later — when the AGENT decides mid-task to write a helper — and nothing
+9 -1
View File
@@ -97,7 +97,15 @@ logger = logging.getLogger(__name__)
# surfacing is a claim ("this rule may apply to what you are doing") that a pull # surfacing is a claim ("this rule may apply to what you are doing") that a pull
# can confirm or refute, while an ambient one is a delivery nobody decided on. # can confirm or refute, while an ambient one is a delivery nobody decided on.
# Add a source here only when a ranker picked it. # Add a source here only when a ranker picked it.
RANKED_SOURCES = ("write_path_rule", "pre_tool_rule") RANKED_SOURCES = (
"write_path_rule", "pre_tool_rule", "prompt_rule",
# A reserved slot is a ranker's choice twice over — it ran a query AND
# decided a kind was worth guaranteeing a place. Left out, its line would
# be counted as bulk delivery and drop out of the denominator, so the one
# surface built because a record class kept losing would be the one whose
# hits nobody could confirm.
"preference_slot",
)
def is_ambient(source: str) -> bool: def is_ambient(source: str) -> bool:
+388 -8
View File
@@ -444,6 +444,165 @@ async def _run_tool_arm(hits, recorder, command="curl -s https://git.example/api
return await pc.build_tool_rule_hint(1, tool, command, **kwargs) return await pc.build_tool_rule_hint(1, tool, command, **kwargs)
# ── the prompt-boundary arm (#3852) ─────────────────────────────────────
#
# The third arm, and it joins _ARMS rather than getting a test file of its
# own. That is the point of the shared parametrisation: #3497's history is
# that the pre-tool arm inherited a defect from its sibling by being MODELLED
# on it instead of sharing with it, and a third arm modelled on two is two
# chances to repeat that. Everything in the family — repeat rendering,
# fresh-only counting, the log-before-bailout order, the kind register, the
# two recorders reading one list — is a property of every arm or of none.
#
# Fewer patches than its siblings because it does less: no prior-art menu, no
# config object, no concept query. Just a bar, a search, and two recorders.
def _prompt_patches(pc, hits, recorder, retrieval_log=None):
return (
# The arm reads its own threshold key rather than a shared config
# object — a third corpus with a bar nothing has yet tuned for it.
patch.object(pc, "get_setting", AsyncMock(return_value="0.6")),
patch.object(pc, "semantic_search_rules", AsyncMock(return_value=hits)),
patch.object(pc, "record_retrieval", retrieval_log or MagicMock()),
patch.object(pc, "record_rule_surfaced", recorder),
)
async def _run_prompt_arm(hits, recorder, prompt="please merge to main",
retrieval_log=None, **kwargs):
from scribe.services import plugin_context as pc
with ExitStack() as stack:
for ctx in _prompt_patches(pc, hits, recorder, retrieval_log=retrieval_log):
stack.enter_context(ctx)
return await pc.build_prompt_rule_hint(1, prompt, **kwargs)
@pytest.mark.asyncio
async def test_the_prompt_arm_retrieves_against_what_the_operator_SAID():
"""The gap this arm closes, stated as the thing that had no trigger.
Both other arms are keyed on an act — a file write, a command. A rule that
governs what to SAY has no act in front of it: extract intent from loose
phrasing, raise a conflict before acting, end a finding with an offer all
bind on a response. Before this, the operator's message reached only
`semantic_search_notes`, so no rule had ever been retrieved against a
thing the operator actually said.
"""
rec = MagicMock()
search = AsyncMock(return_value=[(0.79, fake_rule(
id=2, title="`main` — never without explicit request",
when_to_apply="opening or merging a dev→main pull request",
))])
from scribe.services import plugin_context as pc
with ExitStack() as stack:
stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
stack.enter_context(patch.object(pc, "semantic_search_rules", search))
stack.enter_context(patch.object(pc, "record_retrieval", MagicMock()))
stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
out = await pc.build_prompt_rule_hint(1, "please merge to main")
# The PROMPT is the query — not a path, not a command.
assert search.call_args.args[1] == "please merge to main"
assert "get_rule(2)" in out["context"]
assert rec.call_args.kwargs["source"] == "prompt_rule"
@pytest.mark.asyncio
async def test_the_prompt_arm_addresses_the_request_not_a_tool_call():
"""`where` has to name the moment, and this arm's moment is the asking.
"may apply to this Bash call" would be a lie here — there is no Bash call,
which is the entire reason the arm exists.
"""
out = await _run_prompt_arm(
[(0.79, fake_rule(id=77, title="Extract intent from loose phrasing"))],
MagicMock(),
)
assert "may apply to this request" in out["context"], out["context"]
@pytest.mark.asyncio
async def test_the_prompt_arm_says_nothing_when_asked_nothing():
"""A blank prompt is not a query, and searching on one would put a row in
retrieval_logs that no operator action produced."""
search = AsyncMock(return_value=[])
log = MagicMock()
from scribe.services import plugin_context as pc
with ExitStack() as stack:
stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
stack.enter_context(patch.object(pc, "semantic_search_rules", search))
stack.enter_context(patch.object(pc, "record_retrieval", log))
stack.enter_context(patch.object(pc, "record_rule_surfaced", MagicMock()))
out = await pc.build_prompt_rule_hint(1, " ")
assert out == {"context": "", "rule_ids": []}
search.assert_not_called()
log.assert_not_called()
@pytest.mark.asyncio
async def test_the_prompt_arm_logs_the_calls_that_found_nothing():
"""#3497's defect, pinned on the arm that most needs it.
This bar is INHERITED from the act arms and unverified against prose. The
zero rows are therefore the whole evidence base for whether 0.72 belongs
here at all — an arm that logged only the calls it liked would report a
flawless clear-rate however wrong the number is.
"""
log = MagicMock()
out = await _run_prompt_arm([], MagicMock(), retrieval_log=log)
assert out["context"] == ""
# BY SOURCE, not by count. The reserved slot (#3894) logs its own query on
# the same call, so a bare call_count would pin the number of arms rather
# than the property — and would go red the next time one is added, which
# is rule 167's false alarm about the very thing being protected.
general = [c for c in log.call_args_list
if c.kwargs.get("source") == "prompt_rule"]
assert len(general) == 1, (
"the prompt arm returned early without logging a call that found "
"nothing — the only evidence its inherited threshold is too high"
)
assert general[0].kwargs["results"] == []
@pytest.mark.asyncio
async def test_turning_off_the_notes_menu_does_not_turn_off_rules():
"""Why this is a separate function and not a branch in the notes arm.
`build_autoinject_hint` returns early when auto-inject is disabled, when
the query is blank, and when nothing clears the note bar. Every one of
those is a statement about NOTES. Folded together, an operator who turned
the awareness menu off would silently stop receiving RULES — a coupling
with no symptom, since both failure modes look like a quiet hook.
Pinned as "the rule arm never asks the notes arm's config", which is the
structural fact rather than a simulation of the setting. A future refactor
that reaches for that config here fails, whatever it then does with it.
"""
from scribe.services import plugin_context as pc
notes_cfg = AsyncMock(return_value={
"enabled": False, "threshold": 0.55, "top_k": 3,
})
with ExitStack() as stack:
stack.enter_context(patch.object(pc, "get_autoinject_config", notes_cfg))
for ctx in _prompt_patches(
pc, [(0.79, fake_rule(id=2, title="`main` — never without explicit request"))],
MagicMock(),
):
stack.enter_context(ctx)
out = await pc.build_prompt_rule_hint(1, "please merge to main")
assert "get_rule(2)" in out["context"], (
"the rule arm went quiet while the notes menu was disabled — the two "
"are different claims with different costs of being missed, and one "
"operator setting should not silence both"
)
notes_cfg.assert_not_called()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_the_tool_arm_names_a_rule_for_the_command_about_to_run(): async def test_the_tool_arm_names_a_rule_for_the_command_about_to_run():
"""The 2026-09-03 incident in one test: reaching for curl against the forge """The 2026-09-03 incident in one test: reaching for curl against the forge
@@ -871,7 +1030,14 @@ _THREE_HITS = [
(0.74, fake_rule(id=161, title="Reach the forge through its MCP tools")), (0.74, fake_rule(id=161, title="Reach the forge through its MCP tools")),
] ]
_ARMS = [("write_path_rule", _run_arm), ("pre_tool_rule", _run_tool_arm)] _ARMS = [
("write_path_rule", _run_arm),
("pre_tool_rule", _run_tool_arm),
# The prompt arm joins the family rather than being modelled on it — see
# the block above _prompt_patches for why that distinction is the whole
# lesson of #3497.
("prompt_rule", _run_prompt_arm),
]
def _both_ends(log, rec, source): def _both_ends(log, rec, source):
@@ -896,7 +1062,7 @@ def _both_ends(log, rec, source):
return logged, surfaced return logged, surfaced
@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"]) @pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
@pytest.mark.parametrize( @pytest.mark.parametrize(
("excluded", "expected"), ("excluded", "expected"),
[([], 3), ([157], 2), ([156, 157, 161], 0)], [([], 3), ([157], 2), ([156, 157, 161], 0)],
@@ -954,7 +1120,7 @@ _FRESH_TAIL = "not in this session's loaded set"
_SEEN_TAIL = "You saw it earlier this session" _SEEN_TAIL = "You saw it earlier this session"
@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"]) @pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_a_rule_already_on_the_ledger_still_produces_a_line(source, run): async def test_a_rule_already_on_the_ledger_still_produces_a_line(source, run):
"""THE REGRESSION, stated as the thing that used to be absent. """THE REGRESSION, stated as the thing that used to be absent.
@@ -983,7 +1149,7 @@ async def test_a_rule_already_on_the_ledger_still_produces_a_line(source, run):
) )
@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"]) @pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_the_two_tails_are_distinguishable_and_say_the_true_one(source, run): async def test_the_two_tails_are_distinguishable_and_say_the_true_one(source, run):
"""One clause differs, and it is the clause that would otherwise be false. """One clause differs, and it is the clause that would otherwise be false.
@@ -1029,7 +1195,7 @@ _RULE_FORCE = "before deciding it does not apply"
_PREF_FORCE = "for how this has been done before" _PREF_FORCE = "for how this has been done before"
@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"]) @pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_a_preference_does_not_speak_in_the_rules_voice(source, run): async def test_a_preference_does_not_speak_in_the_rules_voice(source, run):
"""The register, pinned on the two places force is actually asserted. """The register, pinned on the two places force is actually asserted.
@@ -1053,7 +1219,7 @@ async def test_a_preference_does_not_speak_in_the_rules_voice(source, run):
) )
@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"]) @pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_kind_and_seen_do_not_read_each_other(source, run): async def test_kind_and_seen_do_not_read_each_other(source, run):
"""The structural claim the design rests on: two INDEPENDENT axes. """The structural claim the design rests on: two INDEPENDENT axes.
@@ -1093,7 +1259,7 @@ async def test_kind_and_seen_do_not_read_each_other(source, run):
) )
@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"]) @pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_neither_tail_injects_the_rule_statement(source, run): async def test_neither_tail_injects_the_rule_statement(source, run):
"""The budget, pinned on both branches. """The budget, pinned on both branches.
@@ -1138,7 +1304,7 @@ async def test_neither_tail_injects_the_rule_statement(source, run):
# checkable, so it is asserted rather than described. # checkable, so it is asserted rather than described.
@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"]) @pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_a_reference_is_rendered_but_not_counted(source, run): async def test_a_reference_is_rendered_but_not_counted(source, run):
"""The counters must read exactly as they did before #3750.""" """The counters must read exactly as they did before #3750."""
@@ -1174,3 +1340,217 @@ async def test_a_reference_is_rendered_but_not_counted(source, run):
"whole context will not do: the write-path arm fills it from four " "whole context will not do: the write-path arm fills it from four "
"other sources." "other sources."
) )
# ── the reserved preference slot (#3894) ────────────────────────────────
#
# A rule and a preference are not equally served by one ranking, because their
# losses are not equal. A rule crowded out at the prompt boundary still fires
# at an act arm — a push reaches pre_tool_rule, a write reaches
# write_path_rule. A preference about how to ANSWER has no later act: the
# response is the act, so crowded out here it is never delivered at all.
#
# The failure is invisible without this slot. The rule that won is a
# legitimate hit, the telemetry reads healthy, and the only symptom is a
# preference that quietly never arrives — which is `reuse_slot`'s shape one
# corpus over (#2463), where snippets kept losing to project records that
# merely resembled the query.
def _search_by_kind(general, preference):
"""Stand in for the two calls the arm makes against one corpus.
The arm searches twice: once across every kind, once filtered to
preferences for the slot. A single return value cannot tell those apart,
and a test that could not tell them apart would pass against an arm that
never filtered at all — which is the one thing making the slot a slot.
"""
async def _search(*_args, **kwargs):
return preference if kwargs.get("kind") == "preference" else general
return AsyncMock(side_effect=_search)
async def _run_slot(general, preference, recorder=None, retrieval_log=None, **kwargs):
from scribe.services import plugin_context as pc
rec = recorder or MagicMock()
with ExitStack() as stack:
stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
stack.enter_context(patch.object(
pc, "semantic_search_rules", _search_by_kind(general, preference)))
stack.enter_context(patch.object(
pc, "record_retrieval", retrieval_log or MagicMock()))
stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
out = await pc.build_prompt_rule_hint(1, "please merge to main", **kwargs)
return out, rec
_PREF_HIT = [(0.74, fake_rule(
id=140, kind="preference", title="Let each action land before the next",
when_to_apply="before starting an action while a previous one is settling",
))]
_RULES_FILLING_THE_LIMIT = [
(0.81, fake_rule(id=2, title="`main` — never without explicit request")),
(0.79, fake_rule(id=1, title="`dev` is home")),
(0.77, fake_rule(id=153, title="Merge dev→main with a plain merge commit")),
]
@pytest.mark.asyncio
async def test_a_preference_that_lost_the_ranking_still_gets_a_line():
"""The whole point. Three rules fill the limit; the preference places
fourth on score and would never be seen without the slot."""
out, _rec = await _run_slot(_RULES_FILLING_THE_LIMIT, _PREF_HIT)
assert "get_rule(140)" in out["context"], (
"a preference cleared the bar, placed behind the rules, and was "
"dropped — which is the outcome with no symptom: the rules that won "
"are legitimate hits and nothing in the telemetry looks wrong"
)
assert "Preference that may apply" in out["context"]
@pytest.mark.asyncio
async def test_the_slot_is_not_spent_when_a_preference_already_placed():
"""A guaranteed slot is a floor, not a quota. A preference that earned its
place on score does not entitle the corpus to a second one."""
general = [(0.81, fake_rule(id=140, kind="preference", title="Let each action land"))]
search_log = MagicMock()
out, _rec = await _run_slot(general, _PREF_HIT, retrieval_log=search_log)
sources = [c.kwargs.get("source") for c in search_log.call_args_list]
assert "preference_slot" not in sources, (
"the slot ran while a preference had already placed — a second "
"reserved line for a kind already represented is noise the general "
"ranking had already decided against"
)
assert out["context"].count("Preference that may apply") == 1
@pytest.mark.asyncio
async def test_the_slot_query_can_only_answer_with_a_preference():
"""Filtered at the QUERY, not verified afterwards.
An unfiltered search that happened to return a rule would spend the slot
on it, and that line would be indistinguishable from one that earned its
place on score — a slot silently spent on the wrong kind is worse than no
slot at all.
"""
from scribe.services import plugin_context as pc
search = _search_by_kind(_RULES_FILLING_THE_LIMIT, _PREF_HIT)
with ExitStack() as stack:
stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
stack.enter_context(patch.object(pc, "semantic_search_rules", search))
stack.enter_context(patch.object(pc, "record_retrieval", MagicMock()))
stack.enter_context(patch.object(pc, "record_rule_surfaced", MagicMock()))
await pc.build_prompt_rule_hint(1, "please merge to main")
kinds = [c.kwargs.get("kind") for c in search.await_args_list]
assert "preference" in kinds, "the slot searched without filtering by kind"
assert kinds.count("preference") == 1, "the slot searched more than once"
@pytest.mark.asyncio
async def test_the_slot_runs_even_when_the_general_search_found_nothing():
"""The ordering the arm gets wrong by default.
An empty general result is not proof no preference qualifies: that search
overfetches by distance and then collapses, so a preference ranked below
the window is invisible to it while a kind-filtered query finds it at
once. Bailing out first would make the slot dead in exactly the corpus it
exists for — one where rules outnumber preferences.
"""
out, _rec = await _run_slot([], _PREF_HIT)
assert "get_rule(140)" in out["context"], (
"the arm returned early on an empty general result and never asked "
"for a preference"
)
@pytest.mark.asyncio
async def test_the_slot_extends_rather_than_displacing():
"""Nothing the general search returned goes un-shown.
`reuse_slot` evicts its menu's weakest hit; this one does not, and the
reason is the ledger. A displaced hit sits in `prompt_rule`'s
retrieval_logs row while never being surfaced, so that source's two
tables stop agreeing — and #3668's identity is the cheapest true
statement available about this pair. Milestone #379 is what losing it
costs: five steps planned against two counters disagreeing.
"""
log, rec = MagicMock(), MagicMock()
out, _ = await _run_slot(
_RULES_FILLING_THE_LIMIT, _PREF_HIT, recorder=rec, retrieval_log=log,
)
for rule_id in (2, 1, 153):
assert f"get_rule({rule_id})" in out["context"], (
f"rule {rule_id} was returned and logged, then pushed out by the "
f"slot — surfaced and logged now disagree for prompt_rule"
)
logged = [
r.id for c in log.call_args_list if c.kwargs.get("source") == "prompt_rule"
for _s, r in c.kwargs["results"]
]
surfaced = [
rid for c in rec.call_args_list if c.kwargs.get("source") == "prompt_rule"
for rid in c.kwargs["rule_ids"]
]
assert logged == surfaced, (
f"prompt_rule logged {logged} and surfaced {surfaced} — the identity "
f"#3668 pins, broken by the slot rather than by a write path"
)
@pytest.mark.asyncio
async def test_the_slot_accounts_for_itself_under_its_own_source():
"""Both sides of the trade, logged.
#2463's own finding is the warning rather than the precedent: the hit
that slot pushed OUT was in retrieval_logs while the query that pushed it
out was not, so the slot could never be judged against what it displaced.
This one logs its query AND records its surfacing, under a source of its
own, so it can be evaluated separately from the ranking it bypassed.
"""
log, rec = MagicMock(), MagicMock()
await _run_slot(_RULES_FILLING_THE_LIMIT, _PREF_HIT,
recorder=rec, retrieval_log=log)
slot_logs = [c for c in log.call_args_list
if c.kwargs.get("source") == "preference_slot"]
assert len(slot_logs) == 1, "the slot ran without logging its own query"
assert [r.id for _s, r in slot_logs[0].kwargs["results"]] == [140]
slot_surfacings = [c for c in rec.call_args_list
if c.kwargs.get("source") == "preference_slot"]
assert len(slot_surfacings) == 1
assert slot_surfacings[0].kwargs["rule_ids"] == [140]
@pytest.mark.asyncio
async def test_a_preference_on_the_ledger_keeps_the_slot_and_is_not_recounted():
"""Repeats hold the slot; they just do not count twice.
A preference is the kind of record where being reminded is the point, so
one already on the ledger occupies the slot rather than being skipped for
a fresh one — rendered with the repeat tail (#3750). What it must not do
is register a second surfacing, which would count one delivery twice in
the denominator pull-through is read from (#3752).
"""
log, rec = MagicMock(), MagicMock()
out, _ = await _run_slot(
_RULES_FILLING_THE_LIMIT, _PREF_HIT,
recorder=rec, retrieval_log=log, exclude_rule_ids=[140],
)
assert "get_rule(140)" in out["context"]
assert _SEEN_TAIL in out["context"], "the repeat was rendered as a first surfacing"
assert not [c for c in rec.call_args_list
if c.kwargs.get("source") == "preference_slot"], (
"a preference the session had already been shown was counted as a "
"fresh surfacing"
)
assert 140 not in out["rule_ids"], (
"a repeat was written back to the hook's ledger, which would keep "
"pushing its stamp forward so it never aged out (#3751)"
)