refactor(retrieval): one registry for every surface's floor and budget (#4102)

Groundwork for the step's real change. The operator's decision is that the
floor is chosen and adjusted by the model using it, not shipped as a value
somebody has to defend:

  "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 consistent across six arms that each spell their
configuration differently, so the arms stop owning their numbers.
`retrieval_surfaces.SURFACES` names each one, its floor key and default, its
budget key and default, and — because they are rendered by the tuning tool and
the Settings UI — what it asks, over what corpus, and how often it fires. A
floor cannot be moved responsibly by anyone who does not know those three.

Three things fall out:

- **`k` becomes a real budget everywhere.** Only auto-inject had a configurable
  one; `RULEHINT_LIMIT`, `PROMPTRULE_LIMIT` and `reply_preferences.LIMIT` were
  constants. `k` is what binds under a low floor, so it has to be settable per
  surface — and per surface is the point, since `pre_tool_rule` fires before
  every Bash call while `prompt_rule` fires once a turn.
- **`write_path` gets its own budget, inherited not reset.** It shared
  auto-inject's outright on the argument that "how many titles at once" means
  the same thing on both. It does not, for the same reason. Unset, it still
  reads auto-inject's key, so an install that tuned the shared knob does not
  silently drop to a new default.
- **The duplicated read-and-clamp goes.** That shape is canon #2860 across 295
  of 372 judged siblings. Survivable while the numbers were constants; not once
  they are meant to move.

The long measurement comments stay exactly where they are — #2223's noise-floor
probe, #3853's command-vs-code split, #3851's band measurement. The constants
they annotate now alias the registry, so there is one value and the reasoning
still sits beside it.

Tests build the write-path config from the registry (`helpers.writepath_cfg`)
instead of from hand-written dicts. That is not tidiness: the rule arms read
their numbers inside a fail-open `except`, so a dict missing one key does not
raise where a reader would see it — the arm silently becomes a no-op that reads
exactly like "fired and found nothing". Ten hand-written dicts each looked
complete on the day they were typed.

tests/test_retrieval_surfaces.py pins the identity everything rests on: a
surface's name IS its telemetry source. Nothing in the type system says so —
`record_retrieval(source="pre_tool_rule")` is a literal in another file — and
renaming one without the other yields an arm that can be tuned and not
measured, or measured and not tuned, with no symptom either way.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy

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

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):

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,

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"

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()))

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)), \

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