bvandeusenandClaude Opus 5
390846a3d5
CI & Build / TypeScript typecheck (push) Failing after 2s
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 16s
CI & Build / Python tests (push) Successful in 42s
CI & Build / Build & push image (push) Has been skipped
Closes #2244. Retrieval matches on concept, and concepts are language-agnostic: asking about a TypeScript union-find scores 0.72-0.73 against a PYTHON snippet, comfortably over the 0.68 bar. That is useful — a different-language solution gives you the shape even when the code isn't reusable — but the menu line said nothing about it, so the reader either dismissed a good structural reference or pasted Python into a .ts file. Worth noting this predates the concept-query change: raw TS code already matched the Python snippet at 0.73, because the embedder reads identifiers and structure semantically rather than syntactically. The fail state has been shipping quietly; #2242 only made it an intended use rather than an accident. - knowledge._note_to_item projects `language` from the data mirror, same shape as the existing verification projection — a plain column read, no body parsing. - The semantic arm carries language through on the item it builds; it is the arm where these arise, since a snippet recorded AT the path you're editing is almost never in another language. - _prior_art_line folds it into the marker: [similar 0.72 · python]. Together with the score rather than after the title, because the two jointly are the judgement being offered. - One explanatory line is added to the menu, and only when something on it is actually tagged. Two deliberate calls: LABEL, DON'T FILTER. A stricter threshold for foreign-language hits would suppress exactly the shape-borrowing this exists for. They were never the problem; their being undisclosed was. ONLY CLAIM A MISMATCH YOU CAN ESTABLISH. _foreign_language returns "" when either side is unknown — unrecognised extension, or a snippet with no recorded language. A wrong "· python" is worse than no tag. Same-language hits stay unlabelled, so the common case keeps a clean line and the preamble stays off the menu entirely. Operator-typed language names fold through an alias table first (py/python3 → python, tsx → typescript, c++ → cpp); unrecognised names pass through lowercased, which still makes an unknown-but-equal pair compare equal. Trap found while building: _note() in the tests is a MagicMock, so `note.data` auto-created a truthy mock that would have rendered its repr into a menu line. Both test helpers now set data = None explicitly. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs
785 lines
37 KiB
Python
785 lines
37 KiB
Python
"""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
|
|
# Explicitly None, not left to MagicMock's auto-attribute: the semantic arm
|
|
# reads `note.data` for the snippet's language (#2244), and an auto-created
|
|
# mock there is truthy, so it would render its repr into the menu line.
|
|
n.data = None
|
|
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()
|
|
|
|
|
|
# --- cross-language prior art is disclosed, not filtered (#2244) --------------
|
|
|
|
def test_language_for_path_and_canonicalisation():
|
|
from scribe.services import plugin_context as pc
|
|
assert pc._language_for_path("src/a/b.ts") == "typescript"
|
|
assert pc._language_for_path("src/a/b.py") == "python"
|
|
assert pc._language_for_path("ops/Dockerfile") == "dockerfile"
|
|
assert pc._language_for_path("README") == ""
|
|
assert pc._language_for_path("weird/thing.qqq") == ""
|
|
# Operator-typed spellings that mean the same thing must compare equal.
|
|
assert pc._canonical_language("Python3") == pc._canonical_language("py")
|
|
assert pc._canonical_language("TSX") == pc._canonical_language("typescript")
|
|
assert pc._canonical_language("C++") == "cpp"
|
|
# Unknown-but-equal still compares equal, which is all this must get right.
|
|
assert pc._canonical_language("Brainfuck") == pc._canonical_language("brainfuck")
|
|
|
|
|
|
def test_foreign_language_only_claims_a_mismatch_it_can_establish():
|
|
from scribe.services import plugin_context as pc
|
|
assert pc._foreign_language({"language": "python"}, "typescript") == "python"
|
|
assert pc._foreign_language({"language": "py"}, "python") == "" # same, aliased
|
|
assert pc._foreign_language({"language": ""}, "typescript") == "" # theirs unknown
|
|
assert pc._foreign_language({"language": "python"}, "") == "" # target unknown
|
|
assert pc._foreign_language({}, "typescript") == "" # no field at all
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_cross_language_hit_is_labelled_and_explained():
|
|
"""The fail state this closes: an unlabelled Python hit offered while writing
|
|
TypeScript is either dismissed as irrelevant or pasted into the .ts file."""
|
|
from scribe.services import plugin_context as pc
|
|
note = _note(7, "group_pairs — collapse related pairs into groups")
|
|
note.data = {"language": "python"}
|
|
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.72, note)])), \
|
|
patch.object(pc, "record_retrieval", MagicMock()), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
|
|
out = await pc.build_write_path_hint(1, "frontend/src/lib/cluster.ts", code=REAL_CODE)
|
|
assert "[similar 0.72 · python]" in out["context"]
|
|
assert "shape of a solution to adapt, not code to copy" in out["context"]
|
|
# Disclosed, NOT filtered — a stricter bar for foreign hits would suppress
|
|
# exactly the shape-borrowing this exists for.
|
|
assert out["note_ids"] == [7]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_same_language_hit_is_not_labelled_and_gets_no_preamble():
|
|
"""The common case keeps a clean line; the explanation only appears when
|
|
there is something on the menu it explains."""
|
|
from scribe.services import plugin_context as pc
|
|
note = _note(7, "debounce — rate-limit a callback")
|
|
note.data = {"language": "python"}
|
|
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.72, note)])), \
|
|
patch.object(pc, "record_retrieval", MagicMock()), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
|
|
out = await pc.build_write_path_hint(1, "src/scribe/services/x.py", code=REAL_CODE)
|
|
assert "[similar 0.72]" in out["context"]
|
|
assert "·" not in out["context"]
|
|
assert "shape of a solution" not in out["context"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_unknown_target_extension_never_invents_a_mismatch():
|
|
"""A wrong "· python" tag is worse than no tag at all."""
|
|
from scribe.services import plugin_context as pc
|
|
note = _note(7, "helper — does a thing")
|
|
note.data = {"language": "python"}
|
|
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.72, note)])), \
|
|
patch.object(pc, "record_retrieval", MagicMock()), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
|
|
out = await pc.build_write_path_hint(1, "scripts/thing.qqq", code=REAL_CODE)
|
|
assert "· python" not in out["context"]
|
|
|
|
|
|
def test_prior_art_line_keeps_language_and_attribution_together():
|
|
from scribe.services.plugin_context import _prior_art_line
|
|
item = {"id": 12, "title": "group_pairs — collapse pairs"}
|
|
assert _prior_art_line(item, "similar 0.72", None, "python") == \
|
|
'> - #12 [similar 0.72 · python] "group_pairs — collapse pairs"'
|
|
both = _prior_art_line(item, "similar 0.72", "alex", "python")
|
|
assert "· python" in both and "shared by alex" in both
|
|
|
|
|
|
# --- 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)
|