bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity
Files
57781770c349c3995e764daf509337c3618673bb
FabledScribe/tests/test_write_path_trigger.py
T
bvandeusenandClaude Opus 5 57781770c3
CI & Build / Python lint (push) Successful in 3s
Details
CI & Build / Plugin hooks (push) Successful in 7s
Details
CI & Build / integration (push) Successful in 20s
Details
CI & Build / TypeScript typecheck (push) Successful in 33s
Details
CI & Build / Python tests (push) Successful in 42s
Details
CI & Build / Build & push image (push) Successful in 26s
Details
feat(write-path): query snippets by concept, not by raw code 
Implements #2242, from the operator's question: would it make more sense to
search by the concept of the snippet than by the code itself?

It would, measurably. A snippet's embedded text is f"{title}\n{body}", and a
snippet's body is composed markdown — When to use / Signature / Location, then
the fenced code — so `when_to_use` appears TWICE in the vector and the document
is prose-forward. The semantic arm was interrogating it with raw code carrying
no prose at all. Measured on the deployed instance against snippet #2222:

  query built from            score   best unrelated   separation
  raw code body               0.743   0.630            0.11
  name + docstring            0.823   0.602            0.22
  hand-written concept prose  0.835   0.583            0.25

A 12-word description beats a near-verbatim reimplementation of the function,
and code-as-query RAISES the noise floor. It's also the cleanest explanation for
the fragment miss recorded on #2223: a short excerpt carries almost no prose to
match a document that is mostly prose.

So build the query from what the code says it's FOR — declarations plus the
first docstring / JSDoc / leading comment block — shaped as "name(params) — what
it does", mirroring a snippet's own title, which is the form that measured 0.823.

Server-side rather than in the hook: no manifest bump, so installed 0.1.20
plugins get this immediately; multi-language parsing in bash would be miserable;
and it's unit-testable here.

Two rules worth calling out, both measured rather than chosen:

- NO DOC, NO REWRITE. A bare identifier is not a concept and scored 0.671 vs the
  code body's 0.743. Separation from noise is identical either way (0.113), but
  the absolute drops under the 0.68 bar, so preferring a bare name would convert
  a comfortable hit into a miss. Undocumented code keeps the raw payload.
- The 48-char floor still judges the RAW payload, before the rewrite. A concept
  query is allowed to be shorter than the floor — that is the point, the best
  queries are short — but a sub-floor edit stays silent even with a docstring.
  Applying the floor after extraction would discard the best queries.

Regex, not a parser: this is on a PreToolUse critical path and an Edit's
new_string is rarely a valid module, so a miss must cost only a fallback. Every
unrecognised language (Vue SFC, config files) degrades to exactly the previous
behaviour.

Telemetry now logs the concept query rather than the code, since retrieval_logs
is what the threshold gets tuned from and the two aren't comparable.

0.68 is left alone: signal rises to 0.82 while noise FALLS to 0.58, so the bar
sits mid-gap instead of near the edge. To be re-measured against the deployed
instance rather than assumed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs
2026-07-30 09:58:06 -04:00

693 lines
32 KiB
Python
Raw Blame History

"""Tests for the write-path trigger (#2082) — prior art at the moment code is written.
Covers the service (`build_write_path_hint`): the two arms and their precedence,
the anti-bloat gates carried over from milestone 93, telemetry under its own
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 json
import re
import subprocess
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
PLUGIN = Path(__file__).resolve().parents[1] / "plugin"
HOOK = PLUGIN / "hooks" / "scribe_prior_art.sh"
def _snippet_item(nid, title, user_id=1):
"""A row as services.snippets.list_snippets returns it (a dict, not a Note)."""
return {"id": nid, "title": title, "user_id": user_id, "note_type": "snippet"}
def _note(nid, title, user_id=1):
n = MagicMock()
n.id, n.title, n.user_id = nid, title, user_id
return n
def _cfg(**over):
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
async def test_disabled_returns_empty_and_touches_nothing():
from scribe.services import plugin_context as pc
search, listing = AsyncMock(), AsyncMock()
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=REAL_CODE)
assert out["context"] == "" and out["note_ids"] == []
listing.assert_not_called()
search.assert_not_called()
@pytest.mark.asyncio
async def test_no_path_returns_empty():
"""Nothing to look up by place, and the semantic arm alone isn't this feature."""
from scribe.services import plugin_context as pc
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=REAL_CODE)
assert out["context"] == ""
search.assert_not_called()
@pytest.mark.asyncio
async def test_nothing_recorded_is_silent():
from scribe.services import plugin_context as pc
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", AsyncMock(return_value=[])), \
patch.object(pc, "record_retrieval", MagicMock()):
out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
assert out["context"] == "" and out["note_ids"] == []
# --- arm 1: by place ---------------------------------------------------------
@pytest.mark.asyncio
async def test_snippet_at_this_path_is_surfaced_without_a_score():
"""A snippet recorded HERE is prior art by definition, not by resemblance —
it must not be subject to the similarity threshold."""
from scribe.services import plugin_context as pc
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
patch.object(pc.snippets_svc, "list_snippets",
AsyncMock(return_value=([_snippet_item(12, "debounce — rate-limit a callback")], 1))), \
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=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"]
assert "src/x.py" in out["context"]
@pytest.mark.asyncio
async def test_directory_is_only_consulted_when_the_file_leaves_room():
from scribe.services import plugin_context as pc
calls = []
async def _listing(uid, **kw):
calls.append(kw["path"])
if kw["path"] == "src/lib/x.py":
return ([_snippet_item(i, f"s{i}") for i in (1, 2, 3)], 3)
return ([_snippet_item(9, "nearby")], 1)
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
patch.object(pc.snippets_svc, "list_snippets", _listing), \
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=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]
@pytest.mark.asyncio
async def test_directory_hits_are_marked_nearby_not_here():
from scribe.services import plugin_context as pc
async def _listing(uid, **kw):
if kw["path"] == "src/lib":
return ([_snippet_item(9, "sibling helper")], 1)
return ([], 0)
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
patch.object(pc.snippets_svc, "list_snippets", _listing), \
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=REAL_CODE)
assert '#9 [nearby] "sibling helper"' in out["context"]
@pytest.mark.asyncio
async def test_a_failing_location_lookup_does_not_sink_the_hint():
"""The place arm is best-effort — a broken query must not cost the semantic one."""
from scribe.services import plugin_context as pc
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
patch.object(pc.snippets_svc, "list_snippets", AsyncMock(side_effect=RuntimeError("boom"))), \
patch.object(pc, "semantic_search_notes",
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=REAL_CODE)
assert out["note_ids"] == [5]
# --- arm 2: by meaning, under the milestone-93 gates -------------------------
@pytest.mark.asyncio
async def test_semantic_arm_is_snippet_only_and_browse_scoped():
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=([], 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=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"
# Nobody asked for this, so it must not reach a one-to-one direct share.
assert kwargs["scope"] == "browse"
@pytest.mark.asyncio
async def test_margin_gate_applies_to_the_semantic_arm():
from scribe.services import plugin_context as pc
hits = [(0.80, _note(11, "near")), (0.74, _note(22, "alsoNear")), (0.61, _note(33, "far"))]
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=5))), \
patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
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=REAL_CODE)
assert out["note_ids"] == [11, 22]
assert "#33" not in out["context"]
@pytest.mark.asyncio
async def test_no_code_means_no_semantic_arm():
"""An Edit whose payload we couldn't read still gets the location answer."""
from scribe.services import plugin_context as pc
search = AsyncMock()
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="")
assert out["note_ids"] == [12]
search.assert_not_called()
@pytest.mark.asyncio
async def test_place_beats_meaning_and_the_cap_covers_both_arms():
from scribe.services import plugin_context as pc
async def _listing(uid, **kw):
if kw["path"] == "src/x.py":
return ([_snippet_item(1, "placed")], 1)
return ([], 0)
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=2))), \
patch.object(pc.snippets_svc, "list_snippets", _listing), \
patch.object(pc, "semantic_search_notes",
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=REAL_CODE)
# Location hit first; the whole menu stays within top_k.
assert out["note_ids"] == [1, 7]
assert len(out["note_ids"]) <= 2
assert out["context"].index("#1") < out["context"].index("#7")
@pytest.mark.asyncio
async def test_semantic_arm_only_asks_for_the_budget_the_place_arm_left():
from scribe.services import plugin_context as pc
search = AsyncMock(return_value=[])
async def _listing(uid, **kw):
if kw["path"] == "src/x.py":
return ([_snippet_item(1, "placed")], 1)
return ([], 0)
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
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=REAL_CODE)
assert search.await_args.kwargs["limit"] == 2
@pytest.mark.asyncio
async def test_session_dedup_excludes_ids_from_both_arms():
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, "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=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"]
# --- attribution + telemetry -------------------------------------------------
@pytest.mark.asyncio
async def test_another_users_snippet_is_attributed():
"""Rule #47 / the #2084 lesson: an unattributed line reads as your own record."""
from scribe.services import plugin_context as pc
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
patch.object(pc.snippets_svc, "list_snippets",
AsyncMock(return_value=([_snippet_item(12, "theirs", user_id=42)], 1))), \
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=REAL_CODE)
assert "shared by alex, treat as a suggestion" in out["context"]
@pytest.mark.asyncio
async def test_telemetry_uses_its_own_source():
"""Separate source is what lets this surface be tuned apart from auto_inject."""
from scribe.services import plugin_context as pc
rec = 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",
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=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"
assert rec.call_args.kwargs["project_id"] == 4
@pytest.mark.asyncio
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",
AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
cfg = await pc.get_writepath_config(1)
# Its own switch is off while auto-inject stays on...
assert cfg["enabled"] is False
# ...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()))
# --- concept extraction: query by what the code IS FOR, not by the code (#2242) ---
def test_concept_query_prefers_the_name_and_docstring():
"""Snippet documents are prose-forward — `when_to_use` appears in both the
title and the body — so a concept query out-scores the code itself (0.823 vs
0.743 measured against #2222). The shape mirrors a snippet's own title."""
from scribe.services.plugin_context import concept_query
q = concept_query(
'def collapse_into_clusters(edges):\n'
' """Collapse similar-pairs into connected components."""\n'
' parent = {}\n'
' return parent\n'
)
assert q == "collapse_into_clusters(edges) — Collapse similar-pairs into connected components."
def test_concept_query_reads_jsdoc_and_the_const_arrow_form():
"""The dominant shape in this repo's frontend isn't `function foo()`."""
from scribe.services.plugin_context import concept_query
q = concept_query(
"/**\n"
" * Rate-limit a callback so it fires once after the last call.\n"
" */\n"
"export const debounce = (fn: Fn, wait = 250) => {\n"
" let t: number | undefined;\n"
"};\n"
)
assert q == "debounce(fn: Fn, wait = 250) — Rate-limit a callback so it fires once after the last call."
def test_concept_query_does_not_let_a_shebang_become_the_description():
"""`#!` matches the leading-comment pattern but says nothing about the code."""
from scribe.services.plugin_context import concept_query
q = concept_query(
"#!/usr/bin/env bash\n"
"# Resolve the repo remote and url-encode it for the plugin API.\n"
"encode_remote() {\n"
" git remote get-url origin\n"
"}\n"
)
assert q == "encode_remote() — Resolve the repo remote and url-encode it for the plugin API."
assert "usr/bin/env" not in q
def test_concept_query_falls_back_when_there_is_no_doc():
"""MEASURED RULE, not a taste call. An identifier alone scored 0.671 against
#2222 where the full code body scored 0.743 — same separation from the noise
floor (0.113), but below the 0.68 bar, so preferring the bare name would turn
a comfortable hit into a miss. Undocumented code keeps the raw payload."""
from scribe.services.plugin_context import concept_query
assert concept_query(
"def collapse_into_clusters(edges):\n"
" parent = {}\n"
" return parent\n"
) == ""
def test_concept_query_falls_back_on_a_doc_that_says_nothing():
from scribe.services.plugin_context import concept_query
assert concept_query('def f(x):\n """TODO"""\n return x\n') == ""
def test_concept_query_falls_back_when_nothing_parses():
"""A Vue SFC has neither a recognised declaration nor a doc block. Every
unhandled language must degrade to exactly the pre-#2242 behaviour."""
from scribe.services.plugin_context import concept_query
assert concept_query(
'<script setup lang="ts">\n'
"const props = defineProps<{ modelValue: string }>()\n"
"</script>\n"
) == ""
def test_concept_query_accepts_a_doc_without_a_declaration():
"""A module docstring is already the concept; it needs no signature."""
from scribe.services.plugin_context import concept_query
q = concept_query('"""Turn a raw git remote into the project it belongs to."""\nMAPPING = {}\n')
assert q == "Turn a raw git remote into the project it belongs to."
def test_concept_query_caps_what_it_collects():
"""A 40-function Write must not become a wall of signatures, and a long module
docstring must not drown out the declaration."""
from scribe.services import plugin_context as pc
many = '"""Utilities."""\n' + "".join(
f"def helper_{i}(a, b):\n return a\n" for i in range(12)
)
q = pc.concept_query(many)
assert q.count("helper_") == pc._CONCEPT_MAX_DECLS
long_doc = '"""' + ("word " * 400) + '"""\ndef f(a):\n return a\n'
assert len(pc.concept_query(long_doc)) <= pc._CONCEPT_MAX_DOC_CHARS + 64
@pytest.mark.asyncio
async def test_semantic_arm_queries_the_concept_not_the_raw_payload():
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=([], 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=REAL_CODE)
sent = search.await_args.args[1] if len(search.await_args.args) > 1 else search.await_args.kwargs["query"]
assert sent.startswith("debounce(fn, wait=0.25)")
assert "Rate-limit a callback so it fires once after the last call." in sent
assert "nonlocal timer" not in sent # the implementation body is gone
@pytest.mark.asyncio
async def test_telemetry_records_the_query_actually_sent():
"""retrieval_logs is what the threshold gets tuned from, so it has to hold the
concept query — logging the raw code would make the scores uninterpretable."""
from scribe.services import plugin_context as pc
rec = 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", AsyncMock(return_value=[])), \
patch.object(pc, "record_retrieval", rec):
await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
logged = rec.call_args.kwargs["query"]
assert logged.startswith("debounce(fn, wait=0.25)")
assert "nonlocal timer" not in logged
@pytest.mark.asyncio
async def test_the_floor_judges_the_raw_payload_not_the_concept():
"""ORDER OF OPERATIONS. The floor asks "is this a helper being written?" of the
RAW payload; the rewrite happens after. A concept query is allowed to be
shorter than the floor — that's the whole point, since the best queries are
short — but a sub-floor payload must stay silent even when it has a docstring."""
from scribe.services import plugin_context as pc
# Clears the floor, and its concept is deliberately SHORTER than the floor.
verbose = (
'def slugify(text):\n'
' """Turn text into a url slug."""\n'
' out = re.sub(r"[^a-z0-9]+", "-", text.lower())\n'
' return out.strip("-")\n'
)
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=verbose)
concept = pc.concept_query(verbose)
assert len("".join(concept.split())) < pc.WRITEPATH_MIN_CODE_CHARS
search.assert_called_once() # ...and it still searched
# Under the floor, docstring notwithstanding.
tiny = 'def f(x):\n """Slug it."""\n return x\n'
search2 = 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", search2), \
patch.object(pc, "record_retrieval", MagicMock()):
await pc.build_write_path_hint(1, "src/x.py", code=tiny)
assert len("".join(tiny.split())) < pc.WRITEPATH_MIN_CODE_CHARS
search2.assert_not_called()
# --- the route between them --------------------------------------------------
def test_route_reads_every_arg_the_hook_sends():
"""Rule #33: the hook's query string and the route's reader are one contract,
and the hook is the only caller — a rename on either side is silent."""
import inspect
from scribe.routes import plugin as routes
src = inspect.getsource(routes.write_path_prior_art)
for arg in ("path", "code", "repo", "project_id", "exclude_ids"):
assert f'request.args.get("{arg}"' in src, f"route ignores {arg}"
hook = HOOK.read_text()
for arg in ("path=", "code=", "repo=", "exclude_ids="):
assert arg in hook, f"hook never sends {arg}"
def test_route_resolves_repo_to_a_project_not_to_a_location_filter():
"""A snippet's `repo` is a label the operator typed; the hook sends a git
remote. Matching one against the other would silently return nothing, so the
remote may only reach resolve_project."""
import inspect
from scribe.routes import plugin as routes
src = inspect.getsource(routes.write_path_prior_art)
assert "resolve_project" in src
assert "repo=repo" not in src
# --- the plugin surface ------------------------------------------------------
def test_hook_is_registered_for_write_and_edit():
cfg = json.loads((PLUGIN / "hooks" / "hooks.json").read_text())
entries = cfg["hooks"]["PreToolUse"]
assert any(
"Write" in e.get("matcher", "") and "Edit" in e.get("matcher", "")
and any("scribe_prior_art.sh" in h["command"] for h in e["hooks"])
for e in entries
)
def test_hook_never_returns_a_permission_decision():
"""The load-bearing property: this hook informs a write, it cannot stop one.
A permissionDecision of deny/ask would put a recall aid in the way of the
operator's work."""
src = HOOK.read_text()
assert "additionalContext" in src
code_lines = [ln for ln in src.splitlines() if not ln.lstrip().startswith("#")]
assert not any("permissionDecision" in ln for ln in code_lines)
def test_hook_is_executable_and_shell_valid():
assert HOOK.stat().st_mode & 0o111, "hook must be executable"
subprocess.run(["bash", "-n", str(HOOK)], check=True)
def test_hook_reads_both_write_and_edit_payload_shapes():
"""Write and Edit name the payload differently, and the names have changed
across Claude Code versions — read whichever is present."""
src = HOOK.read_text()
for field in ("content", "file_content", "new_string", "new_str"):
assert f".tool_input.{field}" in src
assert ".tool_input.file_path" in src
def test_hook_stays_a_get_so_a_read_scoped_key_works():
"""Every other plugin hook works with a read-scoped key; a POST would demand
write scope (see auth.py) and silently break those installs."""
src = HOOK.read_text()
assert "-X POST" not in src and "--data" not in src
assert "/api/plugin/prior-art?" in src
def test_hook_exits_silently_when_unconfigured():
"""An install with no Scribe URL/token must produce no output at all."""
out = subprocess.run(
["bash", str(HOOK)],
input=json.dumps({
"session_id": "s1", "cwd": "/tmp", "tool_name": "Write",
"tool_input": {"file_path": "/tmp/x.py", "content": "def f(): ..."},
}),
capture_output=True, text=True,
env={"PATH": "/usr/bin:/bin", "SCRIBE_URL": "", "SCRIBE_TOKEN": ""},
)
assert out.returncode == 0
assert out.stdout.strip() == ""
def test_hook_skips_prose_and_data_files():
"""No round-trip for a markdown edit — the server would return nothing anyway."""
src = HOOK.read_text()
skip = re.search(r"case \"\$file_path\" in\n(.*?)esac", src, re.S)
assert skip, "expected an extension skip list"
for ext in ("*.md", "*.json", "*.lock", "*.png"):
assert ext in skip.group(1)
# Config formats are deliberately NOT skipped — a workflow file is reusable.
assert "*.yml" not in skip.group(1)
def test_plugin_version_bumped_with_the_hook():
"""The #1040 lesson: a plugin change clients can't see is a change that didn't
ship."""
manifest = json.loads((PLUGIN / ".claude-plugin" / "plugin.json").read_text())
version = tuple(int(p) for p in manifest["version"].split("."))
assert version >= (0, 1, 18)
Reference in New Issue View Git Blame Copy Permalink