fix(write-path): give the semantic arm its own threshold + a payload floor

Closes #2223. The write-path prior-art trigger's semantic arm shared
auto-inject's 0.55 threshold, which was tuned on prose. Code embeddings
sit on a much higher similarity floor — any two Python-shaped payloads
share keywords, indentation and structure — so 0.55 landed INSIDE the
noise band. Measured against the live instance:

  near-duplicate of a recorded helper   0.73-0.74   true positive
  unrelated colour math / Vue SFC / CSS 0.55-0.63   false positive
  `x = 1`                               0.58        false positive

6 of 8 probe payloads produced a nudge; 4 were noise. The margin gate
couldn't help — _AUTOINJECT_BAND is relative to the top hit, so with a
single hit it never engages.

Two gates are now the write-path arm's own:

- kb_writepath_threshold, default 0.68 — above every measured false
  positive, still 0.05 below both true positives. Auto-inject keeps
  0.55; it was tuned on prose and is not implicated. The comment this
  replaces explicitly reserved the split for when telemetry showed the
  surfaces wanted different values, so this is the change it described,
  not a reversal of it.
- WRITEPATH_MIN_CODE_CHARS = 48 non-whitespace chars, below which the
  semantic arm doesn't run at all. Whitespace is excluded so a deeply
  indented one-liner can't pass on padding. 48 sits under the smallest
  plausible reusable helper (~60) and well over a degenerate edit, so it
  errs toward keeping recall — precision is the threshold's job. This is
  the cheap half of the operator's #89 idea; the full length<->threshold
  curve stays open there, since they asked to brainstorm it rather than
  have a scale invented for them.

top_k stays shared — "how many titles at once" means the same thing on
both surfaces.

The existing tests were passing `code="x"` / `code="def f(): ..."` into
the semantic arm, i.e. exactly the payloads the floor now drops, so the
gate tests were never exercising a realistic payload. They now use a
REAL_CODE fixture, plus new coverage for the floor (trivial payload,
padding, place-arm unaffected, real helper passes) and a guard on the
constant itself.

Settings UI carries the new knob with the reasoning in its hint, and the
write-path checkbox no longer claims it shares the threshold.

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

tests/test_write_path_trigger.py

@@ -30,11 +30,31 @@ def _note(nid, title, user_id=1):
 
 
 def _cfg(**over):
-    base = {"enabled": True, "threshold": 0.55, "top_k": 3}
+    base = {"enabled": True, "threshold": 0.68, "top_k": 3}
     base.update(over)
     return base
 
 
+# The semantic arm ignores payloads carrying less than WRITEPATH_MIN_CODE_CHARS
+# of non-whitespace substance (#2223), so any test that expects it to RUN has to
+# pass something a real Edit would plausibly contain. `code="x"` used to be the
+# fixture here, which meant the gate tests were passing payloads that in
+# production are exactly the noise the floor now drops.
+REAL_CODE = '''def debounce(fn, wait=0.25):
+    """Rate-limit a callback so it fires once after the last call."""
+    timer = None
+
+    def wrapped(*a, **kw):
+        nonlocal timer
+        if timer:
+            timer.cancel()
+        timer = threading.Timer(wait, fn, a, kw)
+        timer.start()
+
+    return wrapped
+'''
+
+
 # --- silence: the common case ------------------------------------------------
 
 @pytest.mark.asyncio
@@ -44,7 +64,7 @@ async def test_disabled_returns_empty_and_touches_nothing():
     with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(enabled=False))), \
          patch.object(pc.snippets_svc, "list_snippets", listing), \
          patch.object(pc, "semantic_search_notes", search):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert out["context"] == "" and out["note_ids"] == []
     listing.assert_not_called()
     search.assert_not_called()
@@ -57,7 +77,7 @@ async def test_no_path_returns_empty():
     search = AsyncMock()
     with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
          patch.object(pc, "semantic_search_notes", search):
-        out = await pc.build_write_path_hint(1, "   ", code="def f(): ...")
+        out = await pc.build_write_path_hint(1, "   ", code=REAL_CODE)
     assert out["context"] == ""
     search.assert_not_called()
 
@@ -69,7 +89,7 @@ async def test_nothing_recorded_is_silent():
          patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
          patch.object(pc, "record_retrieval", MagicMock()):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert out["context"] == "" and out["note_ids"] == []
 
 
@@ -86,7 +106,7 @@ async def test_snippet_at_this_path_is_surfaced_without_a_score():
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
          patch.object(pc, "record_retrieval", MagicMock()), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert out["note_ids"] == [12]
     assert '#12 [here] "debounce — rate-limit a callback"' in out["context"]
     assert "get_snippet(id)" in out["context"]
@@ -109,7 +129,7 @@ async def test_directory_is_only_consulted_when_the_file_leaves_room():
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
          patch.object(pc, "record_retrieval", MagicMock()), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
-        out = await pc.build_write_path_hint(1, "src/lib/x.py", code="x")
+        out = await pc.build_write_path_hint(1, "src/lib/x.py", code=REAL_CODE)
     # The file alone filled the budget, so the directory was never queried.
     assert calls == ["src/lib/x.py"]
     assert out["note_ids"] == [1, 2, 3]
@@ -129,7 +149,7 @@ async def test_directory_hits_are_marked_nearby_not_here():
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
          patch.object(pc, "record_retrieval", MagicMock()), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
-        out = await pc.build_write_path_hint(1, "src/lib/x.py", code="x")
+        out = await pc.build_write_path_hint(1, "src/lib/x.py", code=REAL_CODE)
     assert '#9 [nearby] "sibling helper"' in out["context"]
 
 
@@ -143,7 +163,7 @@ async def test_a_failing_location_lookup_does_not_sink_the_hint():
                       AsyncMock(return_value=[(0.80, _note(5, "throttle — …"))])), \
          patch.object(pc, "record_retrieval", MagicMock()), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert out["note_ids"] == [5]
 
 
@@ -157,7 +177,7 @@ async def test_semantic_arm_is_snippet_only_and_browse_scoped():
          patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
          patch.object(pc, "semantic_search_notes", search), \
          patch.object(pc, "record_retrieval", MagicMock()):
-        await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+        await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     kwargs = search.await_args.kwargs
     # Prior art means snippets — not the dev-log that happens to mention one.
     assert kwargs["note_type"] == "snippet"
@@ -174,7 +194,7 @@ async def test_margin_gate_applies_to_the_semantic_arm():
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \
          patch.object(pc, "record_retrieval", MagicMock()), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert out["note_ids"] == [11, 22]
     assert "#33" not in out["context"]
 
@@ -210,7 +230,7 @@ async def test_place_beats_meaning_and_the_cap_covers_both_arms():
                       AsyncMock(return_value=[(0.9, _note(7, "scored"))])), \
          patch.object(pc, "record_retrieval", MagicMock()), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="x")
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     # Location hit first; the whole menu stays within top_k.
     assert out["note_ids"] == [1, 7]
     assert len(out["note_ids"]) <= 2
@@ -231,7 +251,7 @@ async def test_semantic_arm_only_asks_for_the_budget_the_place_arm_left():
          patch.object(pc.snippets_svc, "list_snippets", _listing), \
          patch.object(pc, "semantic_search_notes", search), \
          patch.object(pc, "record_retrieval", MagicMock()):
-        await pc.build_write_path_hint(1, "src/x.py", code="x")
+        await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert search.await_args.kwargs["limit"] == 2
 
 
@@ -244,7 +264,7 @@ async def test_session_dedup_excludes_ids_from_both_arms():
                       AsyncMock(return_value=([_snippet_item(12, "already shown")], 1))), \
          patch.object(pc, "semantic_search_notes", search), \
          patch.object(pc, "record_retrieval", MagicMock()):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="x", exclude_ids=[12])
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, exclude_ids=[12])
     # The location hit was already surfaced this session → dropped, not repeated.
     assert out["note_ids"] == []
     assert 12 in search.await_args.kwargs["exclude_ids"]
@@ -262,7 +282,7 @@ async def test_another_users_snippet_is_attributed():
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
          patch.object(pc, "record_retrieval", MagicMock()), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={42: "alex"})):
-        out = await pc.build_write_path_hint(1, "src/x.py", code="x")
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert "shared by alex, treat as a suggestion" in out["context"]
 
 
@@ -277,7 +297,7 @@ async def test_telemetry_uses_its_own_source():
                       AsyncMock(return_value=[(0.9, _note(7, "scored"))])), \
          patch.object(pc, "record_retrieval", rec), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
-        await pc.build_write_path_hint(1, "src/x.py", code="x", project_id=4)
+        await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, project_id=4)
     rec.assert_called_once()
     assert rec.call_args.kwargs["source"] == "write_path"
     assert rec.call_args.kwargs["source"] != "auto_inject"
@@ -285,7 +305,7 @@ async def test_telemetry_uses_its_own_source():
 
 
 @pytest.mark.asyncio
-async def test_config_has_its_own_switch_but_shares_the_gates():
+async def test_config_has_its_own_switch_and_threshold_but_shares_top_k():
     from scribe.services import plugin_context as pc
     stored = {pc.WRITEPATH_ENABLED_KEY: "false"}
     with patch.object(pc, "get_setting",
@@ -293,11 +313,116 @@ async def test_config_has_its_own_switch_but_shares_the_gates():
         cfg = await pc.get_writepath_config(1)
     # Its own switch is off while auto-inject stays on...
     assert cfg["enabled"] is False
-    # ...and the gates are auto-inject's, not a second set that could drift.
-    assert cfg["threshold"] == pc.AUTOINJECT_DEFAULT_THRESHOLD
+    # ...its threshold is its OWN and stricter than auto-inject's, because code
+    # embeddings have a higher similarity floor than prose (#2223). Sharing 0.55
+    # 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.
     assert cfg["top_k"] == pc.AUTOINJECT_DEFAULT_TOP_K
 
 
+@pytest.mark.asyncio
+async def test_writepath_threshold_is_operator_tunable_and_clamped():
+    """Rule #25 — it's a product knob, so it has to be settable, and a typo in
+    the settings box must not be able to open the gate to everything."""
+    from scribe.services import plugin_context as pc
+
+    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))):
+            return await pc.get_writepath_config(1)
+
+    assert (await _cfg_with("0.9"))["threshold"] == 0.9
+    assert (await _cfg_with("5"))["threshold"] == 1.0        # clamped up
+    assert (await _cfg_with("-3"))["threshold"] == 0.0       # clamped down
+    # Garbage falls back to the default rather than to 0.0, which would surface
+    # every snippet in the corpus on every edit.
+    assert (await _cfg_with("banana"))["threshold"] == pc.WRITEPATH_DEFAULT_THRESHOLD
+
+
+# --- the minimum-substance floor on the semantic arm (#2223) ------------------
+
+@pytest.mark.asyncio
+async def test_trivial_payload_skips_the_semantic_arm_entirely():
+    """`x = 1` scored 0.58 against an unrelated helper on the live instance — the
+    embedder's floor for "some code", not a resemblance. A payload too small to
+    carry meaning must not reach the embedder at all, and must not cost a
+    retrieval_logs row either."""
+    from scribe.services import plugin_context as pc
+    search, rec = AsyncMock(return_value=[]), MagicMock()
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", rec):
+        out = await pc.build_write_path_hint(1, "src/x.py", code="x = 1")
+    assert out["context"] == "" and out["note_ids"] == []
+    search.assert_not_called()
+    rec.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_floor_counts_substance_not_indentation():
+    """Code is indentation-heavy, so raw length would let a deeply-nested
+    one-liner through on padding alone."""
+    from scribe.services import plugin_context as pc
+    padded = " " * 400 + "return None"   # long, but almost no substance
+    search = AsyncMock(return_value=[])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", MagicMock()):
+        await pc.build_write_path_hint(1, "src/x.py", code=padded)
+    assert len(padded) > pc.WRITEPATH_MIN_CODE_CHARS   # would pass a raw-length check
+    search.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_floor_does_not_block_the_place_arm():
+    """The floor is a statement about the semantic query, not about the edit. A
+    snippet recorded at this exact path is prior art however small the change."""
+    from scribe.services import plugin_context as pc
+    search = AsyncMock(return_value=[])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets",
+                      AsyncMock(return_value=([_snippet_item(12, "here helper")], 1))), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+        out = await pc.build_write_path_hint(1, "src/x.py", code="x = 1")
+    assert out["note_ids"] == [12]
+    search.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_real_helper_clears_the_floor():
+    """The floor errs toward keeping recall — anything that plausibly IS a
+    reusable helper has to get through, or the feature stops working."""
+    from scribe.services import plugin_context as pc
+    search = AsyncMock(return_value=[(0.80, _note(5, "debounce — …"))])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
+    assert out["note_ids"] == [5]
+    search.assert_called_once()
+
+
+def test_the_floor_stays_below_the_smallest_plausible_helper():
+    """A guard on the constant itself: raising it far enough to reject a real
+    one-line helper would silently gut the arm rather than sharpen it."""
+    from scribe.services import plugin_context as pc
+    smallest_real_helper = 'def slug(t): return re.sub(r"[^a-z0-9]+", "-", t.lower()).strip("-")'
+    substance = len("".join(smallest_real_helper.split()))
+    assert pc.WRITEPATH_MIN_CODE_CHARS < substance
+    # And still strictly above a degenerate one-liner.
+    assert pc.WRITEPATH_MIN_CODE_CHARS > len("".join("x = 1".split()))
+
+
 # --- the route between them --------------------------------------------------
 
 def test_route_reads_every_arg_the_hook_sends():