diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py
index 3c63f98..3a6b449 100644
--- a/src/scribe/services/plugin_context.py
+++ b/src/scribe/services/plugin_context.py
@@ -31,6 +31,12 @@ from scribe.services.embeddings import semantic_search_notes, semantic_search_ru
 from scribe.services.note_usage import record_surfaced
 from scribe.services.rule_usage import record_rule_surfaced
 from scribe.services.supersession import superseded_ids
+from scribe.services.retrieval_surfaces import (
+    MAX_BUDGET,
+    SURFACES,
+    budget_for,
+    floor_for,
+)
 from scribe.services.retrieval_telemetry import record_retrieval
 from scribe.services.settings import get_setting
 
@@ -54,12 +60,16 @@ _GOAL_CHARS = 200
 # and are meant to be tuned from retrieval_logs (source='auto_inject') once data
 # accrues — they're exposed in the Settings UI, no restart needed.
 AUTOINJECT_ENABLED_KEY = "kb_autoinject_enabled"
-AUTOINJECT_THRESHOLD_KEY = "kb_autoinject_threshold"
-AUTOINJECT_TOP_K_KEY = "kb_autoinject_top_k"
+# The key and the value both live in the registry now (#4102); these names
+# survive because the comments above them are where each number's measurement
+# is recorded, and because the Settings UI agreement test pairs them with the
+# Vue refs by name. One definition, two readable places.
+AUTOINJECT_THRESHOLD_KEY = SURFACES["auto_inject"].floor_key
+AUTOINJECT_TOP_K_KEY = SURFACES["auto_inject"].budget_key
 
 AUTOINJECT_DEFAULT_ENABLED = True
-AUTOINJECT_DEFAULT_THRESHOLD = 0.55
-AUTOINJECT_DEFAULT_TOP_K = 3
+AUTOINJECT_DEFAULT_THRESHOLD = SURFACES["auto_inject"].floor_default
+AUTOINJECT_DEFAULT_TOP_K = SURFACES["auto_inject"].budget_default
 
 # The write-path trigger (#2082) gets its own on/off switch, its own threshold,
 # and shares only top-k. It originally shared the threshold too, on the argument
@@ -83,9 +93,9 @@ AUTOINJECT_DEFAULT_TOP_K = 3
 # pull-through (#2085) once a real corpus accrues; a cross-encoder rerank
 # (#1038) would subsume this bump.
 WRITEPATH_ENABLED_KEY = "kb_writepath_enabled"
-WRITEPATH_THRESHOLD_KEY = "kb_writepath_threshold"
+WRITEPATH_THRESHOLD_KEY = SURFACES["write_path"].floor_key
 WRITEPATH_DEFAULT_ENABLED = True
-WRITEPATH_DEFAULT_THRESHOLD = 0.68
+WRITEPATH_DEFAULT_THRESHOLD = SURFACES["write_path"].floor_default
 
 # The standing-rule arm (milestone 307) gets its own bar — the split #2223 made
 # one surface down, now made for the THIRD corpus. It inherited 0.68 above, and
@@ -133,8 +143,8 @@ WRITEPATH_DEFAULT_THRESHOLD = 0.68
 # case 0.72 was calibrated on, and the telemetry says it is working — the
 # write-path rule arm speaks on 37% of its calls and its refused mass sits at
 # p50 0.6989, comfortably under the bar rather than piled against it.
-RULEHINT_THRESHOLD_KEY = "kb_rulehint_threshold"
-RULEHINT_DEFAULT_THRESHOLD = 0.72
+RULEHINT_THRESHOLD_KEY = SURFACES["write_path_rule"].floor_key
+RULEHINT_DEFAULT_THRESHOLD = SURFACES["write_path_rule"].floor_default
 
 # THE COMMAND ARM'S OWN BAR, AND WHY IT IS NOT THE WRITE PATH'S (#3853).
 #
@@ -193,8 +203,8 @@ RULEHINT_DEFAULT_THRESHOLD = 0.72
 # eligible. Retrieval is ownership-scoped, not project-scoped. Scoping it
 # would drop that ceiling and widen the 0.0115, which is the larger fix and
 # the reason to settle project scoping before tuning this number twice.
-TOOLRULE_THRESHOLD_KEY = "kb_toolrule_threshold"
-TOOLRULE_DEFAULT_THRESHOLD = 0.68
+TOOLRULE_THRESHOLD_KEY = SURFACES["pre_tool_rule"].floor_key
+TOOLRULE_DEFAULT_THRESHOLD = SURFACES["pre_tool_rule"].floor_default
 
 # A SET OF RULES PER ACT, NOT THE SINGLE BEST ONE (#3851).
 #
@@ -223,7 +233,10 @@ TOOLRULE_DEFAULT_THRESHOLD = 0.68
 # and nothing else, so a moment with one clearly-relevant rule still shows
 # one, and a moment with four shows four. The corpus decides, not a constant.
 # The cap survives as a ceiling on the worst case, not as the usual answer.
-RULEHINT_LIMIT = 5
+# NOW A DEFAULT RATHER THAN A CAP (#4102): both rule act-arms read their own
+# budget from the registry, so this is the value an install starts at and not
+# the value it is stuck with. The reasoning below is why 5 is where it starts.
+RULEHINT_LIMIT = SURFACES["write_path_rule"].budget_default
 
 # MEASURED, NOT REASONED — and the reasoning it replaced was wrong (#3851).
 #
@@ -433,7 +446,11 @@ _CONCEPT_MIN_CHARS = 16
 _AUTOINJECT_BAND = 0.10
 # Hard ceiling on top-k regardless of the user's setting — this is an
 # awareness menu (titles only), never a content dump.
-_AUTOINJECT_MAX_TOP_K = 10
+# The budget ceiling, now shared by every surface rather than owned by this
+# one (#4102). It bounds the same thing everywhere — how many lines a single
+# unsolicited injection may occupy — so one surface having a private ceiling
+# was an accident of which arm got a configurable budget first.
+_AUTOINJECT_MAX_TOP_K = MAX_BUDGET
 
 # --- the prompt-boundary rule arm (#3852) ------------------------------------
 #
@@ -446,7 +463,7 @@ _AUTOINJECT_MAX_TOP_K = 10
 # 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"
+PROMPTRULE_THRESHOLD_KEY = SURFACES["prompt_rule"].floor_key
 # 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
@@ -458,7 +475,7 @@ PROMPTRULE_THRESHOLD_KEY = "kb_promptrule_threshold"
 # 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
+PROMPTRULE_DEFAULT_THRESHOLD = SURFACES["prompt_rule"].floor_default
 
 # 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
@@ -469,7 +486,7 @@ PROMPTRULE_DEFAULT_THRESHOLD = 0.72
 # 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
+PROMPTRULE_LIMIT = SURFACES["prompt_rule"].budget_default
 
 # THE COMPLETION-REPORT ARM'S OWN BAR (services/reply_preferences.py).
 #
@@ -487,8 +504,8 @@ PROMPTRULE_LIMIT = 3
 # Kept at the prose arm's starting value rather than tuned: the split is what
 # makes the two independently movable, and a default is a product decision
 # that this install's corpus cannot settle (rule 115).
-REPORTPREF_THRESHOLD_KEY = "kb_reportpref_threshold"
-REPORTPREF_DEFAULT_THRESHOLD = 0.72
+REPORTPREF_THRESHOLD_KEY = SURFACES["report_preference"].floor_key
+REPORTPREF_DEFAULT_THRESHOLD = SURFACES["report_preference"].floor_default
 
 
 def _slugify(text: str) -> str:
@@ -592,8 +609,18 @@ async def build_process_manifest(user_id: int) -> dict:
 async def get_autoinject_config(user_id: int) -> dict:
     """Resolve a user's auto-inject settings, falling back to the defaults.
 
-    Returns {"enabled": bool, "threshold": float, "top_k": int}, clamped to
-    sane ranges (threshold to [0,1]; top_k to [1, _AUTOINJECT_MAX_TOP_K]).
+    Returns {"enabled": bool, "threshold": float, "top_k": int}.
+
+    THE TWO NUMBERS COME FROM THE REGISTRY NOW (#4102). They used to be read and
+    clamped here, and identically again in `get_writepath_config`, and again in
+    three rule arms, and once more in `reply_preferences` — the duplication the
+    shape ledger counts as canon #2860 across 295 of 372 siblings. That was
+    tolerable while the values were shipped constants. It stops being tolerable
+    once a tool is expected to MOVE them, because a tuning surface cannot be
+    consistent across arms that each spell their configuration differently.
+
+    `enabled` stays here: it is this surface's own switch, not a tunable number,
+    and the registry deliberately holds only the pair a floor-tuner touches.
     """
     enabled_raw = await get_setting(
         user_id, AUTOINJECT_ENABLED_KEY,
@@ -601,21 +628,11 @@ async def get_autoinject_config(user_id: int) -> dict:
     )
     enabled = enabled_raw.strip().lower() in ("true", "1", "yes", "on")
 
-    try:
-        threshold = float(await get_setting(
-            user_id, AUTOINJECT_THRESHOLD_KEY, str(AUTOINJECT_DEFAULT_THRESHOLD)))
-    except (TypeError, ValueError):
-        threshold = AUTOINJECT_DEFAULT_THRESHOLD
-    threshold = min(1.0, max(0.0, threshold))
-
-    try:
-        top_k = int(float(await get_setting(
-            user_id, AUTOINJECT_TOP_K_KEY, str(AUTOINJECT_DEFAULT_TOP_K))))
-    except (TypeError, ValueError):
-        top_k = AUTOINJECT_DEFAULT_TOP_K
-    top_k = min(_AUTOINJECT_MAX_TOP_K, max(1, top_k))
-
-    return {"enabled": enabled, "threshold": threshold, "top_k": top_k}
+    return {
+        "enabled": enabled,
+        "threshold": await floor_for(user_id, "auto_inject"),
+        "top_k": await budget_for(user_id, "auto_inject"),
+    }
 
 
 def _record_kind(note) -> str:
@@ -1024,13 +1041,8 @@ async def build_prompt_rule_hint(
         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))
+        threshold = await floor_for(user_id, "prompt_rule")
+        limit = await budget_for(user_id, "prompt_rule")
 
         t0 = time.perf_counter()
         _rep: dict = {}
@@ -1041,7 +1053,7 @@ async def build_prompt_rule_hint(
         # sessions — this surface speaks unasked, and a whole-rulebook answer
         # is only right for someone who asked the whole rulebook.
         hits = await semantic_search_rules(
-            user_id, q, limit=PROMPTRULE_LIMIT, threshold=threshold,
+            user_id, q, limit=limit, threshold=threshold,
             report=_rep, project_id=project_id or None,
         )
         duration_ms = (time.perf_counter() - t0) * 1000.0
@@ -1057,7 +1069,7 @@ async def build_prompt_rule_hint(
         # 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,
+            threshold=threshold, limit=limit,
             project_id=project_id,
             is_task=None, results=fresh, duration_ms=duration_ms,
             best_available=_rep.get("best_available_score"),
@@ -1294,52 +1306,37 @@ def concept_query(code: str) -> str:
 
 
 async def get_writepath_config(user_id: int) -> dict:
-    """Write-path trigger settings: its own `enabled` and `threshold`, auto-inject's top_k.
+    """Write-path trigger settings — and the two rule arms' numbers alongside.
 
-    The threshold OVERRIDES the inherited auto-inject value — code embeddings
-    have a much higher similarity floor than prose, so the two surfaces need
-    different bars. See WRITEPATH_DEFAULT_THRESHOLD for the measurements (#2223).
-    top_k is still shared: "how many titles at once" means the same thing on
-    both surfaces, and nothing suggests they want different ceilings.
+    Three surfaces' worth of configuration arrives in one call because one hook
+    request drives all three arms. They are still three SURFACES with three
+    independent pairs, resolved from the registry (#4102).
+
+    The floors were split apart one at a time, each on its own measurement, and
+    those measurements are recorded where the defaults are: WRITEPATH (#2223 —
+    code embeddings sit on a much higher similarity floor than prose), RULEHINT
+    (a third corpus again), TOOLRULE (#3853 — a shell command is a different
+    query shape from a code payload and scores lower for the same relevance).
+
+    THE BUDGET IS NOW SPLIT TOO. `top_k` used to be auto-inject's outright, on
+    the argument that "how many titles at once" means the same thing on both
+    surfaces. It does not: this arm fires before every Write and Edit while
+    auto-inject fires once a turn, so the same number buys wildly different
+    amounts of attention. `write_path` inherits auto-inject's value when it has
+    none of its own, so no install that tuned the shared knob loses it.
     """
-    cfg = await get_autoinject_config(user_id)
     enabled_raw = await get_setting(
         user_id, WRITEPATH_ENABLED_KEY,
         "true" if WRITEPATH_DEFAULT_ENABLED else "false",
     )
-
-    try:
-        threshold = float(await get_setting(
-            user_id, WRITEPATH_THRESHOLD_KEY, str(WRITEPATH_DEFAULT_THRESHOLD)))
-    except (TypeError, ValueError):
-        threshold = WRITEPATH_DEFAULT_THRESHOLD
-    threshold = min(1.0, max(0.0, threshold))
-
-    try:
-        rule_threshold = float(await get_setting(
-            user_id, RULEHINT_THRESHOLD_KEY, str(RULEHINT_DEFAULT_THRESHOLD)))
-    except (TypeError, ValueError):
-        rule_threshold = RULEHINT_DEFAULT_THRESHOLD
-    rule_threshold = min(1.0, max(0.0, rule_threshold))
-
-    try:
-        tool_rule_threshold = float(await get_setting(
-            user_id, TOOLRULE_THRESHOLD_KEY, str(TOOLRULE_DEFAULT_THRESHOLD)))
-    except (TypeError, ValueError):
-        tool_rule_threshold = TOOLRULE_DEFAULT_THRESHOLD
-    tool_rule_threshold = min(1.0, max(0.0, tool_rule_threshold))
-
     return {
-        **cfg,
         "enabled": enabled_raw.strip().lower() in ("true", "1", "yes", "on"),
-        "threshold": threshold,
-        # Its own bar, for a third corpus — see RULEHINT_DEFAULT_THRESHOLD.
-        "rule_threshold": rule_threshold,
-        # And the COMMAND arm's own bar again, for the same reason one level
-        # down: a shell command is a different query shape from a code payload
-        # and scores lower for the same relevance (#3853). Separate keys, so an
-        # install can move one without the other — which is the whole finding.
-        "tool_rule_threshold": tool_rule_threshold,
+        "threshold": await floor_for(user_id, "write_path"),
+        "top_k": await budget_for(user_id, "write_path"),
+        "rule_threshold": await floor_for(user_id, "write_path_rule"),
+        "rule_top_k": await budget_for(user_id, "write_path_rule"),
+        "tool_rule_threshold": await floor_for(user_id, "pre_tool_rule"),
+        "tool_rule_top_k": await budget_for(user_id, "pre_tool_rule"),
     }
 
 def _rule_band(hits: list) -> list:
@@ -1988,7 +1985,7 @@ async def build_write_path_hint(
         rule_t0 = time.perf_counter()
         _rep_wpr: dict = {}
         hits = await semantic_search_rules(
-            user_id, code or path, limit=RULEHINT_LIMIT,
+            user_id, code or path, limit=cfg["rule_top_k"],
             threshold=cfg["rule_threshold"],
             report=_rep_wpr, project_id=project_id or None,
         )
@@ -2049,7 +2046,7 @@ async def build_write_path_hint(
         # the same readout.
         record_retrieval(
             user_id=user_id, source="write_path_rule", query=code or path,
-            threshold=cfg["rule_threshold"], limit=RULEHINT_LIMIT,
+            threshold=cfg["rule_threshold"], limit=cfg["rule_top_k"],
             project_id=project_id,
             is_task=None, results=fresh, duration_ms=rule_ms,
             best_available=_rep_wpr.get("best_available_score"),
@@ -2145,7 +2142,7 @@ async def build_tool_rule_hint(
         t0 = time.perf_counter()
         _rep_ptr: dict = {}
         hits = await semantic_search_rules(
-            user_id, query, limit=RULEHINT_LIMIT,
+            user_id, query, limit=cfg["tool_rule_top_k"],
             threshold=cfg["tool_rule_threshold"],
             report=_rep_ptr, project_id=project_id or None,
         )
@@ -2168,7 +2165,7 @@ async def build_tool_rule_hint(
         # failure the arm was built to stop.
         record_retrieval(
             user_id=user_id, source="pre_tool_rule", query=query,
-            threshold=cfg["tool_rule_threshold"], limit=RULEHINT_LIMIT,
+            threshold=cfg["tool_rule_threshold"], limit=cfg["tool_rule_top_k"],
             project_id=project_id,
             is_task=None, results=fresh, duration_ms=duration_ms,
             best_available=_rep_ptr.get("best_available_score"),
diff --git a/src/scribe/services/reply_preferences.py b/src/scribe/services/reply_preferences.py
index fd04b5d..40bb968 100644
--- a/src/scribe/services/reply_preferences.py
+++ b/src/scribe/services/reply_preferences.py
@@ -61,13 +61,9 @@ import logging
 import time
 
 from scribe.services.embeddings import semantic_search_rules
-from scribe.services.plugin_context import (
-    REPORTPREF_DEFAULT_THRESHOLD,
-    REPORTPREF_THRESHOLD_KEY,
-)
+from scribe.services.retrieval_surfaces import SURFACES, budget_for, floor_for
 from scribe.services.retrieval_telemetry import record_retrieval
 from scribe.services.rule_usage import record_rule_surfaced
-from scribe.services.settings import get_setting
 
 logger = logging.getLogger(__name__)
 
@@ -85,16 +81,20 @@ COMPLETION_QUERY = (
 # A handful, not a menu. More than a few shape preferences for ONE kind of
 # reply would contradict each other before they helped; the limit is here to
 # keep one noisy corpus from turning a status change into a wall of text.
-LIMIT = 3
+# The STARTING budget, not the budget (#4102). Both numbers this arm runs on
+# now come from the surface registry, so the model that reads this arm's
+# telemetry can move either — which matters more here than anywhere else,
+# because a fixed query makes this arm's score a constant and a floor a hair
+# above it produces a dead arm no amount of traffic will ever reveal.
+LIMIT = SURFACES["report_preference"].budget_default
 
 
 async def _threshold(user_id: int) -> float:
-    try:
-        value = float(await get_setting(
-            user_id, REPORTPREF_THRESHOLD_KEY, str(REPORTPREF_DEFAULT_THRESHOLD)))
-    except (TypeError, ValueError):
-        value = REPORTPREF_DEFAULT_THRESHOLD
-    return min(1.0, max(0.0, value))
+    return await floor_for(user_id, SOURCE)
+
+
+async def _limit(user_id: int) -> int:
+    return await budget_for(user_id, SOURCE)
 
 
 async def completion_preferences(user_id: int, *, project_id: int | None = None) -> list[dict]:
@@ -115,16 +115,17 @@ async def completion_preferences(user_id: int, *, project_id: int | None = None)
     """
     try:
         threshold = await _threshold(user_id)
+        limit = await _limit(user_id)
         report: dict = {}
         t0 = time.perf_counter()
         hits = await semantic_search_rules(
-            user_id, COMPLETION_QUERY, limit=LIMIT, threshold=threshold,
+            user_id, COMPLETION_QUERY, limit=limit, threshold=threshold,
             kind="preference", report=report, project_id=project_id,
         )
         hits = [(score, rule) for score, rule in hits if rule.kind == "preference"]
         record_retrieval(
             user_id=user_id, source=SOURCE, query=COMPLETION_QUERY,
-            threshold=threshold, limit=LIMIT, project_id=project_id,
+            threshold=threshold, limit=limit, project_id=project_id,
             is_task=None, results=hits,
             best_available=report.get("best_available_score"),
             best_available_id=report.get("best_available_id"),
diff --git a/src/scribe/services/retrieval_surfaces.py b/src/scribe/services/retrieval_surfaces.py
new file mode 100644
index 0000000..2fcdbd8
--- /dev/null
+++ b/src/scribe/services/retrieval_surfaces.py
@@ -0,0 +1,243 @@
+"""One registry of the retrieval surfaces, and the two numbers each one has (#4102).
+
+WHY THIS EXISTS
+
+Six push arms each carried their own loose copy of the same shape: a settings
+key, a default, and a limit that was usually a module constant nobody could
+change. `_threshold()` was duplicated so widely that the shape ledger counts it
+as canon across 295 of 372 judged siblings (snippet #2860). That was survivable
+while the numbers were shipped constants an operator occasionally edited.
+
+It stops being survivable once the numbers are meant to MOVE. The operator's
+decision for this step:
+
+    "the floor should be chosen and adjusted by the model using it. we've come
+    back to something either fails or has to be looked at by the user we need a
+    model consistent surface for the adjustment of these floor values. the user
+    should be able to touch it but the model should be the thing handling it 9
+    times out of 10."
+
+A tuning surface cannot be "model consistent" if every arm spells its own
+configuration differently. So the arms stop owning their numbers and read them
+from here instead, and the tuning tool, the routes and the Settings UI all
+enumerate THIS table rather than hard-coding six special cases.
+
+WHAT A FLOOR IS NOW, AND WHAT IT IS NOT
+
+It is not a relevance judgement. Relevance is decided by the reader, which is
+the only participant that can read a trigger against a situation — that is the
+milestone's whole argument, and the injected line has always said so out loud
+("read it before deciding it does not apply").
+
+A floor answers the cheaper question: *is this worth ranking at all*. `k` is
+what binds, and `k` is a BUDGET — how much of this surface's attention a
+candidate list may spend. An arm that fires before every Bash call cannot
+afford what an arm that fires once a turn can.
+
+WHY THE DEFAULTS BELOW ARE STARTING POINTS AND NOT ANSWERS
+
+A cosine score is a distance in BAAI/bge-small-en-v1.5's vector space, measured
+against THIS corpus. It cannot transfer to an install with different records,
+and rule 115 forbids defending a shipped default from this instance's
+telemetry — which is what made the old design unbuildable: every number was a
+guess everywhere except here, and there was no mechanism that could ever
+improve it.
+
+The mechanism is the fix. Scribe ships a starting point and the means to
+correct it, so the values below carry the measurement that motivated them (see
+the long comments in `plugin_context.py`, which are kept where they are because
+they record how each number was first arrived at) without claiming to be right
+for anybody else.
+
+HOW A FLOOR SHOULD ACTUALLY BE MOVED
+
+By reading the records the floor refused — `retrieval_telemetry(
+near_miss_samples=N)` returns them by id — and never by the percentile alone.
+That is not a style preference; it is the one case where the two disagreed and
+was checked. `report_preference` logged 69 consecutive declines with the
+refused score 0.0006 under the bar, and every percentile said "lower it".
+Reading the refused record showed it was rule 77 "Extract intent from loose
+phrasing", a false positive, so lowering the bar would have delivered that rule
+on every completion report ever written. The statistic and the correct action
+pointed in opposite directions, and only opening the record could tell.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from scribe.services.settings import get_setting
+
+# A budget nobody should be able to set past. Not a tuning value — a guard on
+# the worst case, so a mistyped setting cannot turn a menu into a wall of text.
+# Shared by every surface because it bounds the same thing everywhere: how many
+# lines a single unsolicited injection may occupy.
+MAX_BUDGET = 10
+
+
+@dataclass(frozen=True)
+class Surface:
+    """One push arm's tunable pair, plus enough prose to tune it responsibly.
+
+    `asks` / `over` / `fires` are not documentation for this file — they are
+    rendered by the tuning tool and the Settings UI. A floor cannot be moved
+    sensibly by anyone, model or human, who does not know what the query is, what
+    corpus it runs against, or how often it costs something. Those three facts
+    are exactly what separates these arms from each other, and they were
+    previously recoverable only by reading `plugin_context.py`.
+    """
+
+    name: str
+    """The telemetry `source` value, and the join key.
+
+    MUST equal the string this arm passes to `record_retrieval`. Everything
+    useful about tuning depends on that identity: the tool that moves a floor
+    and the table that says what the floor did have to be talking about the same
+    arm. A test asserts it rather than a comment asking nicely.
+    """
+
+    floor_key: str
+    floor_default: float
+    budget_key: str
+    budget_default: int
+    asks: str
+    over: str
+    fires: str
+    budget_falls_back_to: str = ""
+    """A budget key to inherit when this surface has none of its own set.
+
+    Only `write_path` uses it, and only because it USED to share auto-inject's
+    `top_k` outright. Giving it a key without this would silently reset the
+    budget of every install that had tuned the shared one — a behaviour change
+    delivered as a default, which is the shape of regression nobody reports
+    because nothing looks broken.
+    """
+
+
+SURFACES: dict[str, Surface] = {
+    "auto_inject": Surface(
+        name="auto_inject",
+        floor_key="kb_autoinject_threshold",
+        floor_default=0.55,
+        budget_key="kb_autoinject_top_k",
+        budget_default=3,
+        asks="the operator's message, as they typed it",
+        over="notes, snippets, processes and issues",
+        fires="once per operator turn",
+    ),
+    "write_path": Surface(
+        name="write_path",
+        floor_key="kb_writepath_threshold",
+        floor_default=0.68,
+        budget_key="kb_writepath_top_k",
+        budget_default=3,
+        budget_falls_back_to="kb_autoinject_top_k",
+        asks="the code being written, rewritten as a concept query",
+        over="snippets and recorded issues",
+        fires="before every Write and Edit",
+    ),
+    "write_path_rule": Surface(
+        name="write_path_rule",
+        floor_key="kb_rulehint_threshold",
+        floor_default=0.72,
+        budget_key="kb_rulehint_top_k",
+        budget_default=5,
+        asks="the code being written, against rule triggers",
+        over="global rules plus the bound project's own",
+        fires="before every Write and Edit",
+    ),
+    "pre_tool_rule": Surface(
+        name="pre_tool_rule",
+        floor_key="kb_toolrule_threshold",
+        floor_default=0.68,
+        budget_key="kb_toolrule_top_k",
+        budget_default=5,
+        asks="the command about to run, against rule triggers",
+        over="global rules plus the bound project's own",
+        fires="before every Bash call — the busiest arm there is",
+    ),
+    "prompt_rule": Surface(
+        name="prompt_rule",
+        floor_key="kb_promptrule_threshold",
+        floor_default=0.72,
+        budget_key="kb_promptrule_top_k",
+        budget_default=3,
+        asks="the operator's message, against rule triggers",
+        over="global rules plus the bound project's own",
+        fires="once per operator turn",
+    ),
+    "report_preference": Surface(
+        name="report_preference",
+        floor_key="kb_reportpref_threshold",
+        floor_default=0.72,
+        budget_key="kb_reportpref_top_k",
+        budget_default=3,
+        # THE ONE FIXED QUERY, and the reason this arm behaves unlike the rest.
+        # The others score something that varies per call; this one scores a
+        # constant string, so its top score for a given corpus is also a
+        # constant. A floor a hair above that constant is not a quiet arm, it is
+        # a dead one, and no amount of traffic will ever reveal it — which is
+        # precisely how this arm spent 69 calls declining the same record.
+        asks="a fixed question about how to lay out a completion report",
+        over="preferences",
+        fires="when a task finishes",
+    ),
+}
+
+# Reserved slots are deliberately absent. `preference_slot` and `reuse_slot`
+# borrow their parent arm's floor and are hard-limited to one hit each, because
+# their entire purpose is to guarantee a single line to a kind of record that
+# keeps losing a general score contest (#2246, #3894). A budget of "1" is the
+# feature; exposing it as tunable would invite setting it to 0 and silently
+# removing the guarantee.
+
+
+def surface_names() -> list[str]:
+    """Every tunable surface, in a stable order for menus and listings."""
+    return list(SURFACES)
+
+
+def get_surface(name: str) -> Surface:
+    """Look one up, refusing an unknown name loudly.
+
+    A typo'd surface must not be writable. Settings keys are free-form strings
+    in a generic table, so a tuning call naming `pretool_rule` would otherwise
+    write a key nothing ever reads — a change that appears to succeed, reports a
+    new value, and alters nothing.
+    """
+    try:
+        return SURFACES[name]
+    except KeyError:
+        raise ValueError(
+            f"unknown retrieval surface {name!r}. Tunable surfaces are: "
+            + ", ".join(surface_names())
+        ) from None
+
+
+async def floor_for(user_id: int, name: str) -> float:
+    """This install's current floor for a surface, clamped to [0, 1]."""
+    s = get_surface(name)
+    try:
+        value = float(await get_setting(user_id, s.floor_key, str(s.floor_default)))
+    except (TypeError, ValueError):
+        value = s.floor_default
+    return min(1.0, max(0.0, value))
+
+
+async def budget_for(user_id: int, name: str) -> int:
+    """This install's current budget for a surface, clamped to [1, MAX_BUDGET].
+
+    The lower clamp is 1, never 0: a surface turned off is turned off by its
+    `enabled` switch, which says so. A budget of zero would be an arm that runs
+    a search, logs a retrieval, and renders nothing — indistinguishable in the
+    telemetry from a bar nothing cleared, which is the exact confusion this
+    milestone exists to remove.
+    """
+    s = get_surface(name)
+    raw = await get_setting(user_id, s.budget_key, "")
+    if not raw and s.budget_falls_back_to:
+        raw = await get_setting(user_id, s.budget_falls_back_to, "")
+    try:
+        value = int(float(raw)) if raw else s.budget_default
+    except (TypeError, ValueError):
+        value = s.budget_default
+    return min(MAX_BUDGET, max(1, value))
diff --git a/tests/helpers.py b/tests/helpers.py
index 73ce17e..0dcd784 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -322,3 +322,32 @@ def http_sink(reply: bytes = b'{"context":"","note_ids":[]}'):
     finally:
         server.shutdown()
         server.server_close()
+
+
+def writepath_cfg(**over):
+    """A complete `get_writepath_config` stand-in, built from the registry (#4102).
+
+    DERIVED, NOT LITERAL, and the reason is a failure mode this file already
+    warned about in prose without being able to prevent: the write-path hint
+    drives three arms, each of which reads its numbers out of the config dict
+    inside a fail-open `except`. A dict missing one key does not raise where a
+    reader would see it — the arm silently becomes a no-op, which is
+    indistinguishable from the arm working and finding nothing.
+
+    So the keys come from `retrieval_surfaces.SURFACES`. A seventh surface, or a
+    rename, changes this helper for free and cannot quietly disable an arm in
+    ten hand-written dicts that each looked complete on the day they were typed.
+    """
+    from scribe.services.retrieval_surfaces import SURFACES
+
+    cfg = {
+        "enabled": True,
+        "threshold": SURFACES["write_path"].floor_default,
+        "top_k": SURFACES["write_path"].budget_default,
+        "rule_threshold": SURFACES["write_path_rule"].floor_default,
+        "rule_top_k": SURFACES["write_path_rule"].budget_default,
+        "tool_rule_threshold": SURFACES["pre_tool_rule"].floor_default,
+        "tool_rule_top_k": SURFACES["pre_tool_rule"].budget_default,
+    }
+    cfg.update(over)
+    return cfg
diff --git a/tests/test_ledger_references_not_silence.py b/tests/test_ledger_references_not_silence.py
index be1f161..a190a19 100644
--- a/tests/test_ledger_references_not_silence.py
+++ b/tests/test_ledger_references_not_silence.py
@@ -41,7 +41,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from tests.helpers import fake_note
+from tests.helpers import fake_note, writepath_cfg
 
 REAL_CODE = '''def debounce(fn, wait=0.25):
     """Rate-limit a callback so it fires once after the last call."""
@@ -59,9 +59,7 @@ REAL_CODE = '''def debounce(fn, wait=0.25):
 
 
 def _wp_cfg(**over):
-    base = {"enabled": True, "threshold": 0.68, "top_k": 3, "rule_threshold": 0.72}
-    base.update(over)
-    return base
+    return writepath_cfg(**over)
 
 
 def _snippet_item(nid, title, user_id=1):
diff --git a/tests/test_note_usage.py b/tests/test_note_usage.py
index 26a2fc0..5ab69cf 100644
--- a/tests/test_note_usage.py
+++ b/tests/test_note_usage.py
@@ -20,7 +20,7 @@ from scribe.services.note_usage import (
     record_surfaced,
     usage_for_notes,
 )
-from tests.helpers import fake_note
+from tests.helpers import fake_note, writepath_cfg
 
 
 # --- recording ------------------------------------------------------------
@@ -154,8 +154,7 @@ async def test_unscored_location_arms_are_recorded(lookups, expected_source):
         patch.object(
             plugin_context,
             "get_writepath_config",
-            AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3,
-                                    "rule_threshold": 0.72}),
+            AsyncMock(return_value=writepath_cfg(threshold=0.55)),
         ),
         patch.object(
             plugin_context.snippets_svc,
diff --git a/tests/test_retrieval_surfaces.py b/tests/test_retrieval_surfaces.py
new file mode 100644
index 0000000..1f6a720
--- /dev/null
+++ b/tests/test_retrieval_surfaces.py
@@ -0,0 +1,187 @@
+"""The surface registry, and the identity the whole tuning story rests on (#4102).
+
+WHY THIS EXISTS
+
+Six arms used to own their own numbers: a settings key, a default, and a limit
+that was usually a module constant nobody could change. That survived while the
+values were shipped constants. It stops surviving once the operator's decision
+is that the numbers MOVE:
+
+    "the floor should be chosen and adjusted by the model using it… the user
+    should be able to touch it but the model should be the thing handling it 9
+    times out of 10."
+
+A tuning surface cannot be model-consistent across arms that each spell their
+configuration differently, so the arms now read `{floor, budget}` from one
+registry.
+
+WHAT THIS PINS
+
+  1. **A surface's name IS its telemetry source.** This is the load-bearing
+     one. The tool that moves a floor and the table that reports what the floor
+     did have to be naming the same arm, and nothing in the type system says so
+     — `record_retrieval(source="pre_tool_rule")` is a string literal in a
+     different file. Renaming one without the other produces a surface that can
+     be tuned and cannot be measured, or measured and not tuned, and both fail
+     silently.
+  2. **Keys are unique.** Two surfaces sharing a settings key is how
+     `report_preference` spent its first release moving whenever the prompt arm
+     was tuned (#3860) — one dial wearing two labels.
+  3. **An unknown surface is refused.** Settings keys are free-form strings in
+     a generic table, so a typo'd name would write a key nothing reads: a
+     change that appears to succeed, reports a new value, and alters nothing.
+  4. **The budget clamps at 1, never 0.** A zero-budget arm searches, logs a
+     retrieval and renders nothing — indistinguishable in the telemetry from a
+     bar nothing cleared, which is the exact confusion this milestone exists to
+     remove. "Off" is what the `enabled` switch is for.
+  5. **`write_path` inherits auto-inject's budget when it has none.** It used
+     to share that key outright; giving it its own without a fallback would
+     silently reset the budget on every install that had tuned the shared one.
+"""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.services import retrieval_surfaces as rs
+
+SERVICES = ("plugin_context", "reply_preferences")
+
+
+def _service_source(name: str) -> str:
+    from pathlib import Path
+    root = Path(__file__).resolve().parents[1] / "src" / "scribe" / "services"
+    return (root / f"{name}.py").read_text()
+
+
+def test_every_surface_name_is_a_real_telemetry_source():
+    """The join key, checked against the arms that emit it.
+
+    Asserted on the source text rather than by calling the arms, because what
+    can rot here is the literal: a surface renamed in the registry and not in
+    the `record_retrieval(source=…)` call still runs, still logs, and still
+    tunes — just against two different names, so the readout an operator uses
+    to justify a change describes a different arm from the one the change hits.
+    """
+    blob = "\n".join(_service_source(n) for n in SERVICES)
+    # `report_preference` passes its name through a module constant rather than
+    # a literal, so that one name is satisfied by the constant holding it.
+    from scribe.services.reply_preferences import SOURCE
+
+    missing = [
+        s.name for s in rs.SURFACES.values()
+        if f'source="{s.name}"' not in blob and s.name != SOURCE
+    ]
+    assert not missing, (
+        f"these surfaces can be tuned but never measured: {missing}. The "
+        "registry name must match the string the arm passes to record_retrieval."
+    )
+
+
+def test_no_two_surfaces_share_a_settings_key():
+    """One dial, one label. #3860 is what the other way costs."""
+    floors = [s.floor_key for s in rs.SURFACES.values()]
+    budgets = [s.budget_key for s in rs.SURFACES.values()]
+    assert len(set(floors)) == len(floors), f"duplicate floor key in {floors}"
+    assert len(set(budgets)) == len(budgets), f"duplicate budget key in {budgets}"
+    assert not (set(floors) & set(budgets)), "a floor key doubles as a budget key"
+
+
+def test_every_surface_says_what_it_asks_over_what_and_how_often():
+    """The prose is rendered, not decorative.
+
+    A floor cannot be moved responsibly by anyone — model or human — who does
+    not know the query shape, the corpus, or how often the arm costs something.
+    An empty string here reaches the tuning tool and the Settings UI as a blank.
+    """
+    for s in rs.SURFACES.values():
+        for field in ("asks", "over", "fires"):
+            assert getattr(s, field).strip(), f"{s.name}.{field} is empty"
+
+
+def test_an_unknown_surface_is_refused_rather_than_written():
+    with pytest.raises(ValueError) as e:
+        rs.get_surface("pretool_rule")          # a real typo for pre_tool_rule
+    # The message has to name the alternatives, or the caller's next move is a
+    # second guess.
+    assert "pre_tool_rule" in str(e.value)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stored, expected", [
+    ("0.8", 0.8),
+    ("5", 1.0),            # clamped, not believed
+    ("-3", 0.0),
+    ("banana", 0.55),      # unparseable falls back to the surface's default
+    ("", 0.55),
+])
+async def test_a_floor_is_clamped_to_the_unit_interval(stored, expected):
+    with patch.object(rs, "get_setting", AsyncMock(return_value=stored)):
+        assert await rs.floor_for(1, "auto_inject") == expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stored, expected", [
+    ("4", 4),
+    ("999", rs.MAX_BUDGET),
+    ("0", 1),
+    ("-2", 1),
+    ("banana", 3),
+])
+async def test_a_budget_is_clamped_with_a_floor_of_one(stored, expected):
+    """Zero is the value that must not get through.
+
+    An arm with a budget of 0 runs its search, writes a `retrieval_logs` row
+    with `result_count == 0`, and renders nothing — which reads in the telemetry
+    exactly like a bar nothing cleared. Turning a surface off is the `enabled`
+    switch's job, and that one says so.
+    """
+    with patch.object(rs, "get_setting", AsyncMock(return_value=stored)):
+        assert await rs.budget_for(1, "auto_inject") == expected
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_budget_falls_back_to_auto_injects():
+    """The migration this fallback exists for.
+
+    `write_path` had no budget key of its own — it read auto-inject's. An
+    install that had tuned that shared knob to 6 must not silently drop to the
+    new key's default the day this ships; nothing would look broken and the
+    operator would never know to look.
+    """
+    async def _setting(_uid, key, default=""):
+        return {"kb_writepath_top_k": "", "kb_autoinject_top_k": "6"}.get(key, default)
+
+    with patch.object(rs, "get_setting", AsyncMock(side_effect=_setting)):
+        assert await rs.budget_for(1, "write_path") == 6
+
+
+@pytest.mark.asyncio
+async def test_its_own_budget_wins_once_it_is_set():
+    """And the fallback must not become a permanent override."""
+    async def _setting(_uid, key, default=""):
+        return {"kb_writepath_top_k": "2", "kb_autoinject_top_k": "6"}.get(key, default)
+
+    with patch.object(rs, "get_setting", AsyncMock(side_effect=_setting)):
+        assert await rs.budget_for(1, "write_path") == 2
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_config_carries_every_arm_it_drives():
+    """One hook request drives three arms, and each reads its pair out of this.
+
+    A missing key does not raise where a reader would see it — the rule arms
+    fail open, so a KeyError becomes an empty hint, and the arm reads as "fired
+    and found nothing". That is the failure this asserts against, and it is the
+    reason tests build the dict from the registry rather than by hand.
+    """
+    from scribe.services import plugin_context as pc
+
+    # Both modules read settings: `pc` for the enabled switch, `rs` for the
+    # pairs. Patching one and not the other reaches a real database.
+    with patch.object(pc, "get_setting", AsyncMock(return_value="")), \
+         patch.object(rs, "get_setting", AsyncMock(return_value="")):
+        cfg = await pc.get_writepath_config(1)
+
+    for key in ("threshold", "top_k", "rule_threshold", "rule_top_k",
+                "tool_rule_threshold", "tool_rule_top_k"):
+        assert key in cfg, f"{key} missing — its arm will silently no-op"
diff --git a/tests/test_rule_usage_wiring.py b/tests/test_rule_usage_wiring.py
index 6b383d6..554f2af 100644
--- a/tests/test_rule_usage_wiring.py
+++ b/tests/test_rule_usage_wiring.py
@@ -17,7 +17,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from tests.helpers import fake_note, fake_rule
+from scribe.services import retrieval_surfaces as rs
+from tests.helpers import fake_note, fake_rule, writepath_cfg
 
 # The MCP tool layer reads its caller from a ContextVar the HTTP transport sets
 # per request; a unit test has no request, so it binds the caller itself. The
@@ -59,16 +60,15 @@ def _arm_patches(pc, hits, recorder, prior_art=None, cfg=None, rule_search=None,
     """
     return (
         patch.object(pc, "get_writepath_config",
-                     AsyncMock(return_value=cfg or {
-                         "enabled": True, "threshold": 0.6,
-                         "top_k": 3, "rule_threshold": 0.6,
-                         # The command arm reads its OWN bar since #3853, and
-                         # a stub missing this key does not fail where a
-                         # reader would see it: the arm fails open, so the
-                         # KeyError becomes an empty hint and every case in
-                         # _ARMS reports the arm went silent instead.
-                         "tool_rule_threshold": 0.6,
-                     })),
+                     # Every key, derived from the surface registry (#4102).
+                     # A stub missing one does not fail where a reader would
+                     # see it: the arm fails open, so the KeyError becomes an
+                     # empty hint and every case in _ARMS reports the arm went
+                     # silent instead.
+                     AsyncMock(return_value=cfg or writepath_cfg(
+                         threshold=0.6, rule_threshold=0.6,
+                         tool_rule_threshold=0.6,
+                     ))),
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))),
         patch.object(pc, "semantic_search_notes",
                      AsyncMock(return_value=_PRIOR_ART if prior_art is None
@@ -156,8 +156,7 @@ async def test_the_arm_searches_on_its_OWN_bar_not_the_code_one():
     with ExitStack() as stack:
         for ctx in _arm_patches(
             pc, [], MagicMock(), rule_search=search,
-            cfg={"enabled": True, "threshold": 0.60,
-                 "top_k": 3, "rule_threshold": 0.81},
+            cfg=writepath_cfg(threshold=0.60, top_k=3, rule_threshold=0.81),
         ):
             stack.enter_context(ctx)
         await pc.build_write_path_hint(
@@ -392,16 +391,15 @@ def test_every_rules_payload_caller_names_itself():
 def _tool_patches(pc, hits, recorder, cfg=None, retrieval_log=None):
     return (
         patch.object(pc, "get_writepath_config",
-                     AsyncMock(return_value=cfg or {
-                         "enabled": True, "threshold": 0.6,
-                         "top_k": 3, "rule_threshold": 0.6,
-                         # The command arm reads its OWN bar since #3853, and
-                         # a stub missing this key does not fail where a
-                         # reader would see it: the arm fails open, so the
-                         # KeyError becomes an empty hint and every case in
-                         # _ARMS reports the arm went silent instead.
-                         "tool_rule_threshold": 0.6,
-                     })),
+                     # Every key, derived from the surface registry (#4102).
+                     # A stub missing one does not fail where a reader would
+                     # see it: the arm fails open, so the KeyError becomes an
+                     # empty hint and every case in _ARMS reports the arm went
+                     # silent instead.
+                     AsyncMock(return_value=cfg or writepath_cfg(
+                         threshold=0.6, rule_threshold=0.6,
+                         tool_rule_threshold=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),
@@ -433,9 +431,10 @@ async def _run_tool_arm(hits, recorder, command="curl -s https://git.example/api
 
 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")),
+        # The arm reads its own floor and budget from the surface registry
+        # rather than a shared config object (#4102) — a third corpus, with a
+        # pair nothing has yet tuned for it.
+        patch.object(rs, "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),
@@ -469,7 +468,7 @@ async def test_the_prompt_arm_retrieves_against_what_the_operator_SAID():
     ))])
     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(rs, "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))
@@ -503,7 +502,7 @@ async def test_the_prompt_arm_says_nothing_when_asked_nothing():
     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(rs, "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()))
@@ -1416,7 +1415,7 @@ async def _run_slot(general, preference, recorder=None, retrieval_log=None, **kw
     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(rs, "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(
@@ -1480,7 +1479,7 @@ async def test_the_slot_query_can_only_answer_with_a_preference():
     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(rs, "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()))
@@ -1617,8 +1616,7 @@ async def test_each_act_arm_searches_at_its_own_bar():
     """The split, where it actually takes effect."""
     from scribe.services import plugin_context as pc
 
-    cfg = {"enabled": True, "threshold": 0.6, "top_k": 3,
-           "rule_threshold": 0.77, "tool_rule_threshold": 0.61}
+    cfg = writepath_cfg(threshold=0.6, top_k=3, rule_threshold=0.77, tool_rule_threshold=0.61)
 
     search = AsyncMock(return_value=list(_THREE_HITS))
     with ExitStack() as stack:
@@ -1646,8 +1644,7 @@ async def test_an_act_arm_reports_the_bar_it_actually_searched_at():
     """
     from scribe.services import plugin_context as pc
 
-    cfg = {"enabled": True, "threshold": 0.6, "top_k": 3,
-           "rule_threshold": 0.77, "tool_rule_threshold": 0.61}
+    cfg = writepath_cfg(threshold=0.6, top_k=3, rule_threshold=0.77, tool_rule_threshold=0.61)
 
     search = AsyncMock(return_value=list(_THREE_HITS))
     log = MagicMock()
@@ -1712,8 +1709,7 @@ async def test_the_act_arms_scope_their_search_to_the_bound_project(bound, scope
     global rules only, which the search spells as `project_id=None`."""
     from scribe.services import plugin_context as pc
 
-    cfg = {"enabled": True, "threshold": 0.6, "top_k": 3,
-           "rule_threshold": 0.6, "tool_rule_threshold": 0.6}
+    cfg = writepath_cfg(threshold=0.6, top_k=3, rule_threshold=0.6, tool_rule_threshold=0.6)
     tool_search = AsyncMock(return_value=[])
     with ExitStack() as stack:
         stack.enter_context(patch.object(
@@ -1726,7 +1722,7 @@ async def test_the_act_arms_scope_their_search_to_the_bound_project(bound, scope
 
     prompt_search = AsyncMock(return_value=[])
     with ExitStack() as stack:
-        stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
+        stack.enter_context(patch.object(rs, "get_setting", AsyncMock(return_value="0.6")))
         stack.enter_context(patch.object(pc, "semantic_search_rules", prompt_search))
         stack.enter_context(patch.object(pc, "record_retrieval", MagicMock()))
         stack.enter_context(patch.object(pc, "record_rule_surfaced", MagicMock()))
diff --git a/tests/test_services_plugin_context.py b/tests/test_services_plugin_context.py
index fff5f62..9e159ae 100644
--- a/tests/test_services_plugin_context.py
+++ b/tests/test_services_plugin_context.py
@@ -1,7 +1,8 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from tests.helpers import fake_note
+from scribe.services import retrieval_surfaces as rs
+from tests.helpers import fake_note, writepath_cfg
 
 
 pytestmark = pytest.mark.usefixtures("_no_supersession")
@@ -17,7 +18,8 @@ async def test_get_autoinject_config_defaults_and_clamps():
     from scribe.services import plugin_context as pc
 
     # No settings stored → defaults.
-    with patch.object(pc, "get_setting", AsyncMock(side_effect=lambda uid, k, d: d)):
+    with patch.object(pc, "get_setting", AsyncMock(side_effect=lambda uid, k, d: d)), \
+         patch.object(rs, "get_setting", AsyncMock(side_effect=lambda uid, k, d="": d)):
         cfg = await pc.get_autoinject_config(1)
     assert cfg == {
         "enabled": pc.AUTOINJECT_DEFAULT_ENABLED,
@@ -31,8 +33,12 @@ async def test_get_autoinject_config_defaults_and_clamps():
         pc.AUTOINJECT_THRESHOLD_KEY: "5",
         pc.AUTOINJECT_TOP_K_KEY: "999",
     }
-    with patch.object(pc, "get_setting",
-                      AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+    # The switch comes from plugin_context, the two numbers from the registry.
+    def _side(uid, k, d=""):
+        return stored.get(k, d)
+
+    with patch.object(pc, "get_setting", AsyncMock(side_effect=_side)), \
+         patch.object(rs, "get_setting", AsyncMock(side_effect=_side)):
         cfg = await pc.get_autoinject_config(1)
     assert cfg["enabled"] is False
     assert cfg["threshold"] == 1.0
@@ -437,9 +443,7 @@ async def test_write_path_semantic_arm_asks_for_experience_not_just_snippets():
     search = AsyncMock(return_value=hits)
     rec = MagicMock()
     with patch.object(pc, "get_writepath_config",
-                      AsyncMock(return_value={"enabled": True, "threshold": 0.6,
-                                              "top_k": 3,
-                                              "rule_threshold": 0.72})), \
+                      AsyncMock(return_value=writepath_cfg(threshold=0.6))), \
          patch.object(pc.snippets_svc, "list_snippets",
                       AsyncMock(return_value=([], 0))), \
          patch.object(pc, "semantic_search_notes", search), \
@@ -470,9 +474,7 @@ async def test_write_path_labels_a_non_snippet_hit_with_its_kind():
     hits = [(0.72, fake_note(id=9, title="debounce helper", user_id=1, note_type="snippet")),
             (0.71, fake_note(id=7, title="Debounce dropped the trailing call", user_id=1, is_task=True, task_kind="issue"))]
     with patch.object(pc, "get_writepath_config",
-                      AsyncMock(return_value={"enabled": True, "threshold": 0.6,
-                                              "top_k": 3,
-                                              "rule_threshold": 0.72})), \
+                      AsyncMock(return_value=writepath_cfg(threshold=0.6))), \
          patch.object(pc.snippets_svc, "list_snippets",
                       AsyncMock(return_value=([], 0))), \
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \
diff --git a/tests/test_write_path_trigger.py b/tests/test_write_path_trigger.py
index 7fd3bcd..31a1632 100644
--- a/tests/test_write_path_trigger.py
+++ b/tests/test_write_path_trigger.py
@@ -6,6 +6,7 @@ source, and the two ways this must stay silent. Plus the plugin hook contract 
 a PreToolUse hook that returns a permission decision would be able to block the
 operator's edit, which this feature must never do.
 """
+import contextlib
 import json
 import re
 import subprocess
@@ -13,7 +14,30 @@ from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from tests.helpers import fake_note, http_sink
+from tests.helpers import fake_note, http_sink, writepath_cfg
+
+
+@contextlib.contextmanager
+def _stored_settings(stored):
+    """Patch BOTH readers, because one call now uses two (#4102).
+
+    `get_writepath_config` reads its `enabled` switch through
+    `plugin_context.get_setting` and all six tunable numbers through
+    `retrieval_surfaces.get_setting`. Patching only the first leaves the numbers
+    talking to a real database — which in a unit job is a connection error, and
+    in an integration job is worse: the test would pass or fail on whatever the
+    instance happened to have stored.
+    """
+    from scribe.services import plugin_context as pc
+    from scribe.services import retrieval_surfaces as rs
+
+    def _side(uid, k, d=""):
+        return stored.get(k, d)
+
+    with patch.object(pc, "get_setting", AsyncMock(side_effect=_side)), \
+         patch.object(rs, "get_setting", AsyncMock(side_effect=_side)):
+        yield
+
 
 PLUGIN = Path(__file__).resolve().parents[1] / "plugin"
 HOOK = PLUGIN / "hooks" / "scribe_prior_art.sh"
@@ -31,9 +55,7 @@ def _cfg(**over):
     # missing key raises inside its fail-open except and turns the arm into a
     # silent no-op — which is indistinguishable from it working and finding
     # nothing.
-    base = {"enabled": True, "threshold": 0.68, "top_k": 3, "rule_threshold": 0.72}
-    base.update(over)
-    return base
+    return writepath_cfg(**over)
 
 
 # The semantic arm ignores payloads carrying less than WRITEPATH_MIN_CODE_CHARS
@@ -423,11 +445,10 @@ async def test_sync_surfacing_is_measured_under_its_own_usage_source():
 
 
 @pytest.mark.asyncio
-async def test_config_has_its_own_switch_and_threshold_but_shares_top_k():
+async def test_config_has_its_own_switch_threshold_and_inherited_budget():
     from scribe.services import plugin_context as pc
     stored = {pc.WRITEPATH_ENABLED_KEY: "false"}
-    with patch.object(pc, "get_setting",
-                      AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+    with _stored_settings(stored):
         cfg = await pc.get_writepath_config(1)
     # Its own switch is off while auto-inject stays on...
     assert cfg["enabled"] is False
@@ -436,8 +457,12 @@ async def test_config_has_its_own_switch_and_threshold_but_shares_top_k():
     # made unrelated code — including `x = 1` at 0.58 — clear the bar.
     assert cfg["threshold"] == pc.WRITEPATH_DEFAULT_THRESHOLD
     assert cfg["threshold"] > pc.AUTOINJECT_DEFAULT_THRESHOLD
-    # ...and top_k is still shared: "how many titles at once" means the same
-    # thing on both surfaces.
+    # ...and the budget is INHERITED rather than shared (#4102). This arm has
+    # its own key now, because it fires before every Write and Edit while
+    # auto-inject fires once a turn, so the same number buys very different
+    # amounts of attention. Unset, it still reads auto-inject's — which is what
+    # stops the split from silently resetting an install that had tuned the
+    # knob when it was shared.
     assert cfg["top_k"] == pc.AUTOINJECT_DEFAULT_TOP_K
 
 
@@ -449,8 +474,7 @@ async def test_writepath_threshold_is_operator_tunable_and_clamped():
 
     async def _cfg_with(raw):
         stored = {pc.WRITEPATH_THRESHOLD_KEY: raw}
-        with patch.object(pc, "get_setting",
-                          AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+        with _stored_settings(stored):
             return await pc.get_writepath_config(1)
 
     assert (await _cfg_with("0.9"))["threshold"] == 0.9
@@ -474,8 +498,7 @@ async def test_the_rule_arm_has_its_own_tunable_bar():
 
     async def _cfg_with(raw):
         stored = {pc.RULEHINT_THRESHOLD_KEY: raw}
-        with patch.object(pc, "get_setting",
-                          AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+        with _stored_settings(stored):
             return await pc.get_writepath_config(1)
 
     assert (await _cfg_with("0.8"))["rule_threshold"] == 0.8
@@ -494,8 +517,7 @@ async def test_the_two_write_path_bars_are_independent():
     from scribe.services import plugin_context as pc
 
     stored = {pc.WRITEPATH_THRESHOLD_KEY: "0.90", pc.RULEHINT_THRESHOLD_KEY: "0.61"}
-    with patch.object(pc, "get_setting",
-                      AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+    with _stored_settings(stored):
         cfg = await pc.get_writepath_config(1)
 
     assert cfg["threshold"] == 0.90
@@ -516,8 +538,7 @@ async def test_the_two_act_arms_read_independent_rule_bars():
     from scribe.services import plugin_context as pc
 
     stored = {pc.RULEHINT_THRESHOLD_KEY: "0.75", pc.TOOLRULE_THRESHOLD_KEY: "0.61"}
-    with patch.object(pc, "get_setting",
-                      AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+    with _stored_settings(stored):
         cfg = await pc.get_writepath_config(1)
 
     assert cfg["rule_threshold"] == 0.75
@@ -536,8 +557,7 @@ async def test_a_garbage_command_bar_falls_back_to_its_own_default():
     from scribe.services import plugin_context as pc
 
     stored = {pc.TOOLRULE_THRESHOLD_KEY: "banana"}
-    with patch.object(pc, "get_setting",
-                      AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+    with _stored_settings(stored):
         cfg = await pc.get_writepath_config(1)
 
     assert cfg["tool_rule_threshold"] == pc.TOOLRULE_DEFAULT_THRESHOLD