bvandeusenandClaude Fable 5
8664d8ad14
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / Python lint (push) Successful in 4s
CI & Build / TypeScript typecheck (push) Successful in 38s
CI & Build / integration (push) Failing after 44s
CI & Build / Python tests (push) Successful in 1m26s
CI & Build / Build & push image (push) Successful in 31s
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1438 lines
71 KiB
Python
1438 lines
71 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
|
|
from tests.helpers import fake_note, http_sink
|
|
|
|
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 _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"] == []
|
|
|
|
|
|
# --- the sync class, and arm 1 by place --------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snippet_at_this_path_becomes_the_sync_nudge():
|
|
"""A snippet recorded AT the exact file is not a reuse suggestion — it IS
|
|
the record of the file being edited (#2708). It surfaces without a score,
|
|
framed as "updating the record is part of this edit", and is reported in
|
|
sync_note_ids so the hook feeds the sync dedup channel."""
|
|
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 out["sync_note_ids"] == [12]
|
|
assert '#12 [records this file] "debounce — rate-limit a callback"' in out["context"]
|
|
# The load-bearing sentence: the record's freshness belongs to this edit.
|
|
assert "update_snippet" in out["context"]
|
|
assert "verify_snippet" in out["context"]
|
|
assert "src/x.py" in out["context"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reuse_dedup_never_suppresses_the_sync_nudge():
|
|
"""THE bug #2708 names: a title surfaced as a reuse hint twenty turns ago
|
|
landed in exclude_ids — and then silenced "you are editing the recorded
|
|
file right now", the one claim that must fire at the moment of change. The
|
|
sync class dedups only against its own channel."""
|
|
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, "seen as reuse already")], 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=REAL_CODE, exclude_ids=[12],
|
|
)
|
|
assert out["sync_note_ids"] == [12]
|
|
assert "[records this file]" in out["context"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_dedup_silences_a_sync_nudge_already_shown():
|
|
"""Once the session has been told "snippet #12 records this file", every
|
|
further edit of the same file stays quiet about it — on every arm: the
|
|
directory query and the semantic arm would both re-surface it otherwise."""
|
|
from scribe.services import plugin_context as pc
|
|
|
|
async def _listing(uid, **kw):
|
|
# Recorded at the exact file, so it matches the file AND dir queries.
|
|
return ([_snippet_item(12, "already nudged")], 1)
|
|
|
|
search = AsyncMock(return_value=[])
|
|
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", search), \
|
|
patch.object(pc, "record_retrieval", MagicMock()):
|
|
out = await pc.build_write_path_hint(
|
|
1, "src/x.py", code=REAL_CODE, exclude_sync_ids=[12],
|
|
)
|
|
assert out["note_ids"] == [] and out["sync_note_ids"] == []
|
|
assert "#12" not in out["context"]
|
|
assert 12 in search.await_args.kwargs["exclude_ids"]
|
|
|
|
|
|
@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, fake_note(id=5, title="throttle — …", user_id=1, note_type="snippet"))])), \
|
|
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 is snippets AND recorded experience (#2246) — an issue saying
|
|
# "we tried this and it broke" belongs here. What stays out is the open
|
|
# to-do list, which resembles the code and answers nothing.
|
|
assert kwargs["note_type"] == ("snippet", "note")
|
|
assert kwargs["task_kind"] == "issue"
|
|
# 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, fake_note(id=11, title="near", user_id=1, note_type="snippet")), (0.74, fake_note(id=22, title="alsoNear", user_id=1, note_type="snippet")), (0.61, fake_note(id=33, title="far", user_id=1, note_type="snippet"))]
|
|
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, fake_note(id=7, title="scored", user_id=1, note_type="snippet"))])), \
|
|
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_the_reuse_arms():
|
|
"""exclude_ids governs the REUSE classes — nearby and semantic. (The sync
|
|
class has its own channel; see the tests above.)"""
|
|
from scribe.services import plugin_context as pc
|
|
|
|
async def _listing(uid, **kw):
|
|
# Recorded at a SIBLING file, so it reaches only the directory query.
|
|
if kw["path"] == "src":
|
|
return ([_snippet_item(12, "already shown")], 1)
|
|
return ([], 0)
|
|
|
|
search = AsyncMock(return_value=[])
|
|
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", 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 nearby 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, fake_note(id=7, title="scored", user_id=1, note_type="snippet"))])), \
|
|
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_sync_surfacing_is_measured_under_its_own_usage_source():
|
|
"""The sync class's pull-through rate is #2708's scoreboard — whether
|
|
sessions actually update the record when told they're editing it. Folding
|
|
it into write_path_place would make that unmeasurable."""
|
|
from scribe.services import plugin_context as pc
|
|
|
|
async def _listing(uid, **kw):
|
|
if kw["path"] == "src/x.py":
|
|
return ([_snippet_item(12, "records me")], 1)
|
|
return ([_snippet_item(9, "sibling")], 1)
|
|
|
|
surfaced = MagicMock()
|
|
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, "record_surfaced", surfaced), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
|
|
await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
|
|
by_source = {
|
|
c.kwargs["source"]: c.kwargs["note_ids"] for c in surfaced.call_args_list
|
|
}
|
|
assert by_source["write_path_sync"] == [12]
|
|
assert by_source["write_path_place"] == [9]
|
|
|
|
|
|
@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, fake_note(id=5, title="debounce — …", user_id=1, note_type="snippet"))])
|
|
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 = fake_note(id=7, title="group_pairs — collapse related pairs into groups", user_id=1, note_type="snippet")
|
|
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 = fake_note(id=7, title="debounce — rate-limit a callback", user_id=1, note_type="snippet")
|
|
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 = fake_note(id=7, title="helper — does a thing", user_id=1, note_type="snippet")
|
|
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
|
|
# The handler plus the query-scope helper it delegates repo/project_id
|
|
# to — the contract is what the ROUTE MODULE reads, wherever it reads it.
|
|
src = inspect.getsource(routes.write_path_prior_art) + inspect.getsource(
|
|
routes._project_scope
|
|
)
|
|
for arg in ("path", "code", "repo", "project_id", "exclude_ids",
|
|
"exclude_sync_ids", "shapes"):
|
|
assert f'request.args.get("{arg}"' in src, f"route ignores {arg}"
|
|
|
|
hook = HOOK.read_text()
|
|
for arg in ("path=", "code=", "repo=", "exclude_ids=", "exclude_sync_ids=",
|
|
"shapes="):
|
|
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) + inspect.getsource(
|
|
routes._project_scope
|
|
)
|
|
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. The list lives in the shared library (#2901) and the hook asks it."""
|
|
lib = (PLUGIN / "hooks" / "scribe_defs.sh").read_text()
|
|
skip = re.search(r"scribe_skip_path\(\) \{\n case \"\$1\" in\n(.*?)esac", lib, re.S)
|
|
assert skip, "expected an extension skip list in scribe_defs.sh"
|
|
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)
|
|
src = HOOK.read_text()
|
|
assert 'scribe_skip_path "$file_path" && exit 0' in src
|
|
assert '/scribe_defs.sh"' in src # sourced, not copied
|
|
|
|
|
|
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, 31)
|
|
|
|
|
|
def test_hook_keeps_sync_and_reuse_dedup_apart():
|
|
"""#2708's dedup audit, pinned: the hook holds TWO per-session id files and
|
|
feeds each its own class — sync ids (snippets recording the edited file) to
|
|
the sync file, the rest to the reuse file. One shared file is exactly the
|
|
bug this replaced: a reuse hint early in the session silencing the record-
|
|
sync nudge when the recorded file is edited later."""
|
|
src = HOOK.read_text()
|
|
assert ".sync.ids" in src # its own state file
|
|
assert "exclude_sync_ids=" in src # its own query channel
|
|
# The reuse file must NOT swallow sync ids — the write-back subtracts them.
|
|
assert "(.note_ids // []) - (.sync_note_ids // [])" in src
|
|
assert "(.sync_note_ids // [])[]?" in src
|
|
|
|
|
|
def _hook_runtime_env():
|
|
"""Env for tests that need the hook's tools to actually RUN.
|
|
|
|
The silence-contract tests above deliberately restrict PATH — the hook must
|
|
exit quietly when its tools are missing, and asserting on empty output
|
|
doesn't care why it was empty. These tests assert on CONTENT, so the tools
|
|
must resolve wherever the image installed them; skip (don't fail) on an
|
|
image that lacks them, because that image cannot exercise this behaviour
|
|
at all.
|
|
"""
|
|
import os
|
|
import shutil
|
|
|
|
for tool in ("git", "jq", "curl", "bash"):
|
|
if shutil.which(tool) is None:
|
|
pytest.skip(f"hook runtime tool {tool!r} not installed")
|
|
return {"PATH": os.environ["PATH"],
|
|
"SCRIBE_URL": "http://127.0.0.1:9", "SCRIBE_TOKEN": "t"}
|
|
|
|
|
|
def _dup_repo(tmp_path, env):
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
|
|
(repo / "a.py").write_text("def debounce(fn):\n return fn\n")
|
|
# git grep searches the index, so the existing copy must be staged.
|
|
subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
|
|
return repo
|
|
|
|
|
|
def _write_event(repo, session="s-nudge"):
|
|
return json.dumps({
|
|
"session_id": session, "cwd": str(repo), "tool_name": "Write",
|
|
"tool_input": {"file_path": str(repo / "b.py"),
|
|
"content": "def debounce(fn):\n return fn\n"},
|
|
})
|
|
|
|
|
|
def test_hook_nudges_recording_when_copies_exist_but_nothing_is_recorded(tmp_path):
|
|
"""#2664: the local arm proves duplication; when Scribe ANSWERS that it has
|
|
no record of it, the same context block must ask for create_snippet — the
|
|
one moment the recording nudge is earned rather than noise."""
|
|
with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, seen):
|
|
env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
|
|
repo = _dup_repo(tmp_path, env)
|
|
out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo),
|
|
capture_output=True, text=True, env=env)
|
|
assert out.returncode == 0
|
|
assert seen and seen[0]["path"] == ["b.py"]
|
|
assert out.stdout.strip(), (
|
|
"hook produced no output — the local arm should have found the "
|
|
"staged duplicate and nudged"
|
|
)
|
|
ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
|
|
assert "already defined" in ctx # the duplication finding
|
|
assert "create_snippet" in ctx # the recording ask riding it
|
|
assert "did not answer" not in ctx
|
|
|
|
|
|
def test_hook_says_when_scribe_did_not_answer_once_per_outage(tmp_path):
|
|
"""#2932: a configured instance that does not answer (refused connection)
|
|
is SAID — the write went unchecked — instead of the hook failing open in
|
|
silence; the record nudge's "nothing recorded" claim is withheld. Once per
|
|
outage: a second miss is quiet, an answer clears the marker, and the next
|
|
miss speaks again. The marker is shared with the after-write hook."""
|
|
env = _hook_runtime_env() # SCRIBE_URL → a refused port
|
|
repo = _dup_repo(tmp_path, env)
|
|
marker = tmp_path / "scribe-priorart" / "s-out.unreached"
|
|
env["TMPDIR"] = str(tmp_path)
|
|
out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
|
|
capture_output=True, text=True, env=env)
|
|
assert out.returncode == 0
|
|
ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
|
|
assert "already defined" in ctx
|
|
assert "Scribe did not answer the prior-art check for `b.py` within 5s" in ctx
|
|
assert "UNCHECKED" in ctx and "list_shapes" in ctx
|
|
assert "None of those existing copies is recorded" not in ctx
|
|
assert marker.is_file()
|
|
# Second miss inside the quiet window: local arm only.
|
|
out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
|
|
capture_output=True, text=True, env=env)
|
|
ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
|
|
assert "already defined" in ctx and "did not answer" not in ctx
|
|
# An answer clears the marker …
|
|
with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, _seen):
|
|
up = dict(env, SCRIBE_URL=f"http://127.0.0.1:{port}")
|
|
subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
|
|
capture_output=True, text=True, env=up)
|
|
assert not marker.exists()
|
|
# … so the next outage is announced afresh.
|
|
out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
|
|
capture_output=True, text=True, env=env)
|
|
assert "did not answer" in json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
|
|
# A write the hook had nothing local to say about still carries the line
|
|
# (the line is the whole message then): a fresh session, no duplicate.
|
|
(repo / "a.py").unlink()
|
|
subprocess.run(["git", "add", "-A"], cwd=repo, check=True, env=env)
|
|
out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out-2"),
|
|
capture_output=True, text=True, env=env)
|
|
ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
|
|
assert ctx.startswith("> Scribe did not answer")
|
|
|
|
|
|
def test_hook_stays_quiet_about_recording_when_nothing_is_duplicated(tmp_path):
|
|
"""A brand-new helper with no other copies earns no nudge — a reflex that
|
|
fires on every Write is one sessions learn to skip."""
|
|
env = _hook_runtime_env()
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
|
|
out = subprocess.run(
|
|
["bash", str(HOOK)],
|
|
input=json.dumps({
|
|
"session_id": "s-quiet", "cwd": str(repo), "tool_name": "Write",
|
|
"tool_input": {"file_path": str(repo / "b.py"),
|
|
"content": "def debounce(fn):\n return fn\n"},
|
|
}),
|
|
capture_output=True, text=True, env=env,
|
|
)
|
|
assert out.returncode == 0
|
|
assert "create_snippet" not in out.stdout
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("fname", "definition"),
|
|
[
|
|
("scanner.go", "func Resolve(x int) error {\n\treturn nil\n}\n"),
|
|
("scanner_m.go",
|
|
"func (s *Scanner) Resolve(x int) error {\n\treturn nil\n}\n"),
|
|
("queue.kt", "suspend fun refreshQueue(id: Long) {\n}\n"),
|
|
("fetch.rs", "pub async fn fetch_all() -> u32 {\n 0\n}\n"),
|
|
("adapter.go", "type ForgeAdapter struct {\n\tname string\n}\n"),
|
|
],
|
|
ids=["go-func", "go-method", "kotlin-fun", "rust-fn", "go-type"],
|
|
)
|
|
def test_local_arm_finds_duplicates_in_every_language_family(
|
|
tmp_path, fname, definition
|
|
):
|
|
"""#2682: the definition detector must cover ALL code, not the languages of
|
|
the repo it was born in. Its original CSS/JS/Python-only patterns silently
|
|
amputated the local arm — and the #2664 recording nudge gated on it — for
|
|
every Go/Kotlin/Rust project, which is exactly where the operator observed
|
|
recording never happening. Each case stages an existing copy and writes the
|
|
same definition to a second file; the hook must prove the duplication and
|
|
ask for the record (the instance ANSWERS "nothing recorded" — since #2932
|
|
an unanswered call withholds the nudge, so a sink stands in for it)."""
|
|
with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, _seen):
|
|
env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
|
|
(repo / fname).write_text(definition)
|
|
subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
|
|
ext = fname.rsplit(".", 1)[1]
|
|
out = subprocess.run(
|
|
["bash", str(HOOK)],
|
|
input=json.dumps({
|
|
"session_id": f"s-lang-{ext}", "cwd": str(repo),
|
|
"tool_name": "Write",
|
|
"tool_input": {"file_path": str(repo / f"copy.{ext}"),
|
|
"content": definition},
|
|
}),
|
|
capture_output=True, text=True, env=env,
|
|
)
|
|
assert out.returncode == 0
|
|
assert out.stdout.strip(), (
|
|
f"hook produced no output for {fname} — the local arm should have "
|
|
f"found the staged duplicate definition"
|
|
)
|
|
ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
|
|
assert "already defined" in ctx
|
|
assert "create_snippet" in ctx
|
|
|
|
|
|
# --- #2791: the write-path feed — hook evidence lands as ledger rows ----------
|
|
|
|
|
|
def _ts():
|
|
from datetime import datetime, timezone
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stamping_needs_named_shapes_and_a_recent_pull():
|
|
"""Offered-but-ignored stamps nothing: without a PULLED event there is no
|
|
evidence, and without the hook naming shapes there is nothing to stamp.
|
|
Neither case may even read the pull stream."""
|
|
from scribe.services import plugin_context as pc
|
|
pulls = AsyncMock(return_value={})
|
|
stamp = 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", AsyncMock(return_value=[])), \
|
|
patch.object(pc, "record_retrieval", MagicMock()), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", pulls), \
|
|
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp):
|
|
out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, project_id=4)
|
|
assert out["stamped"] == []
|
|
pulls.assert_not_awaited() # no shapes → no read
|
|
out = await pc.build_write_path_hint(
|
|
1, "src/x.py", code=REAL_CODE, project_id=4,
|
|
stamp_shapes=[("sym", "debounce")],
|
|
)
|
|
pulls.assert_awaited_once()
|
|
stamp.assert_not_awaited() # shapes, but no pull
|
|
assert out["stamped"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_pulled_snippet_already_seen_is_evidence_not_menu():
|
|
"""The pulled-then-written flow IS the dedup-excluded flow: the hint offered
|
|
#7 earlier (so it sits in exclude_ids), the session pulled it, and now
|
|
writes code resembling it. #7 must be scored for this payload — and handed
|
|
to the stamp as resemblance — without being re-listed in the menu."""
|
|
from scribe.services import plugin_context as pc
|
|
search = AsyncMock(return_value=[(0.91, fake_note(id=7, title="pulled", user_id=1, note_type="snippet")), (0.80, fake_note(id=8, title="fresh", user_id=1, note_type="snippet"))])
|
|
stamp = AsyncMock(return_value=[{
|
|
"path": "src/x.py", "symbol": "debounce", "kind": "sym",
|
|
"snippet_id": 7, "reason": "hook: pulled #7; payload resembles it (0.91)",
|
|
}])
|
|
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
|
|
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={})), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={7: _ts()})), \
|
|
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp):
|
|
out = await pc.build_write_path_hint(
|
|
1, "src/x.py", code=REAL_CODE, project_id=4, exclude_ids=[7],
|
|
stamp_shapes=[("sym", "debounce")], repo_key="git.example.com/a/b",
|
|
)
|
|
# The query kept #7 eligible (and widened the budget by one for it)...
|
|
kw = search.call_args.kwargs
|
|
assert 7 not in kw["exclude_ids"]
|
|
assert kw["limit"] == 4
|
|
# ...but the menu still honours the session dedup.
|
|
assert out["note_ids"] == [8]
|
|
assert "#7" not in "\n".join(
|
|
line for line in out["context"].splitlines() if "[similar" in line
|
|
)
|
|
# The stamp saw the pull and the resemblance score for this payload.
|
|
skw = stamp.call_args.kwargs
|
|
assert skw["pulled"] == {7: skw["pulled"][7]}
|
|
assert skw["resembles"] == {7: 0.91}
|
|
assert skw["shapes"] == [("sym", "debounce")]
|
|
assert skw["repo_key"] == "git.example.com/a/b"
|
|
assert out["stamped"][0]["snippet_id"] == 7
|
|
# And the session is told what landed, with the way to correct it.
|
|
assert "Shape accounting" in out["context"]
|
|
assert "`debounce` → instance of #7" in out["context"]
|
|
assert "classify_shapes" in out["context"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_stamp_renders_even_when_the_hint_is_otherwise_silent():
|
|
"""After dedup the common case is an empty hint; the stamp must still run
|
|
and still be reported — silence about accounting is how hook rows would
|
|
become invisible."""
|
|
from scribe.services import plugin_context as pc
|
|
stamped = [{"path": "web/b.css", "symbol": "btn-primary", "kind": "css",
|
|
"snippet_id": 5, "reason": "hook: pulled #5; payload references `btn-primary`"}]
|
|
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()), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={5: _ts()})), \
|
|
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances",
|
|
AsyncMock(return_value=stamped)):
|
|
out = await pc.build_write_path_hint(
|
|
1, "web/b.css", code=".btn-primary { color: red; }" * 4, project_id=4,
|
|
stamp_shapes=[("css", "btn-primary")],
|
|
)
|
|
assert out["note_ids"] == []
|
|
assert out["stamped"] == stamped
|
|
assert "`.btn-primary` → instance of #5" in out["context"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_failing_stamp_does_not_sink_the_hint():
|
|
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, "records me")], 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={})), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={12: _ts()})), \
|
|
patch.object(pc.shape_ledger_svc, "stamp_write_path_instances",
|
|
AsyncMock(side_effect=RuntimeError("ledger down"))):
|
|
out = await pc.build_write_path_hint(
|
|
1, "src/x.py", code=REAL_CODE, project_id=4, stamp_shapes=[("sym", "f")],
|
|
)
|
|
assert out["sync_note_ids"] == [12]
|
|
assert out["stamped"] == []
|
|
|
|
|
|
def test_route_stamps_only_for_a_caller_allowed_to_write():
|
|
"""A read-scoped key gets the hint — every plugin hook works on a read key
|
|
— but a GET must never change accounting for it. The route passes the
|
|
shapes through only when the key is write-scoped (or it's a session)."""
|
|
import inspect
|
|
|
|
from scribe.routes import plugin as routes
|
|
src = inspect.getsource(routes.write_path_prior_art)
|
|
assert 'request.args.get("shapes"' in src
|
|
assert '== "write"' in src
|
|
assert "stamp_shapes=shapes if may_stamp else None" in src
|
|
assert "normalize_repo_key(repo)" in src
|
|
hook = HOOK.read_text()
|
|
assert "&shapes=" in hook
|
|
|
|
|
|
def test_hook_does_not_name_a_type_import_specifier_as_a_shape(tmp_path):
|
|
"""#2904, the awk mirror of the server rule: `type Foo,` inside an import
|
|
list is not a definition; `type Baz = …` on its own line is."""
|
|
env = _hook_runtime_env()
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
|
|
seen = _run_hook_against_sink(tmp_path, {
|
|
"session_id": "s-type", "cwd": str(repo), "tool_name": "Write",
|
|
"tool_input": {"file_path": str(repo / "x.ts"),
|
|
"content": 'import { type Foo, bar } from "./y";\n'
|
|
"type Baz = { a: number };\n"
|
|
"export function use(): Baz {\n return { a: 1 };\n}\n"},
|
|
})
|
|
assert seen["shapes"] == ["sym:Baz,sym:use"]
|
|
|
|
|
|
def test_hook_names_the_shapes_being_written():
|
|
"""The feed's two inputs: every definition in the payload, or — for an Edit
|
|
that changes a body, not a signature — the definition enclosing the edit,
|
|
found by walking the target file upward from the edited lines."""
|
|
src = HOOK.read_text()
|
|
lib = (PLUGIN / "hooks" / "scribe_defs.sh").read_text()
|
|
assert "scribe_defs()" in lib # one extractor, shared (#2901)
|
|
assert "scribe_defs()" not in src # ...not a second copy here
|
|
assert ".tool_input.old_string" in src # the Edit's anchor
|
|
assert "| tac | scribe_defs | head -1" in src # nearest definition above
|
|
# The ledger feed sends NAMES, never bodies, and stays on the one GET.
|
|
assert src.count("/api/plugin/prior-art?") == 1
|
|
|
|
|
|
def _run_hook_against_sink(tmp_path, payload):
|
|
"""Run the hook with SCRIBE_URL pointed at a throwaway local listener and
|
|
return the query the hook sent (tests.helpers.http_sink)."""
|
|
with http_sink() as (port, seen):
|
|
env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
|
|
out = subprocess.run(
|
|
["bash", str(HOOK)], input=json.dumps(payload),
|
|
capture_output=True, text=True, env=env,
|
|
)
|
|
assert out.returncode == 0, out.stderr
|
|
return seen[0] if seen else {}
|
|
|
|
|
|
def test_hook_sends_every_definition_in_a_write(tmp_path):
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
env = _hook_runtime_env()
|
|
subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
|
|
seen = _run_hook_against_sink(tmp_path, {
|
|
"session_id": "s-feed-w", "cwd": str(repo), "tool_name": "Write",
|
|
"tool_input": {
|
|
"file_path": str(repo / "new.ts"),
|
|
"content": "export async function onDelete(): Promise<void> {\n"
|
|
" const ok = await confirmed({ title: 'x' });\n}\n"
|
|
".btn-primary {\n color: red;\n}\n",
|
|
},
|
|
})
|
|
assert seen["path"] == ["new.ts"]
|
|
assert seen["shapes"] == ["css:btn-primary,sym:onDelete"]
|
|
|
|
|
|
def test_hook_sends_the_enclosing_definition_for_a_body_edit(tmp_path):
|
|
"""An Edit to the inside of onTrash names no definition itself; the hook
|
|
must walk the file upward from the edited line and send onTrash."""
|
|
import shutil
|
|
if shutil.which("tac") is None:
|
|
pytest.skip("the enclosing-definition walk needs tac")
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
env = _hook_runtime_env()
|
|
subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
|
|
target = repo / "comp.vue"
|
|
target.write_text(
|
|
"<script setup lang=\"ts\">\n"
|
|
"async function onTrash(): Promise<void> {\n"
|
|
" const ok = await confirmed({ title: 'Move to the trash?' });\n"
|
|
" if (!ok) return;\n"
|
|
"}\n"
|
|
"const other = () => {\n return 1;\n};\n"
|
|
"</script>\n"
|
|
)
|
|
seen = _run_hook_against_sink(tmp_path, {
|
|
"session_id": "s-feed-e", "cwd": str(repo), "tool_name": "Edit",
|
|
"tool_input": {
|
|
"file_path": str(target),
|
|
"old_string": " if (!ok) return;",
|
|
"new_string": " if (!ok) return;\n await guarded(() => store.trashNode(id));",
|
|
},
|
|
})
|
|
assert seen["shapes"] == ["sym:onTrash"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_write_time_derive_check_names_a_family_or_a_canon_in_band():
|
|
"""#2900: the ledger's own word on the names being written — a duplicate
|
|
family with no canon, or a canon recorded elsewhere — rendered at the
|
|
write even when nothing else does; keyed so the session's exclude
|
|
channel silences a family already named."""
|
|
from scribe.services import plugin_context as pc
|
|
found = [
|
|
{"symbol": "log-empty", "kind": "css", "key": "name:css:log-empty",
|
|
"family": {"group": "name:css:log-empty", "label": ".log-empty", "identical": False,
|
|
"files": ["a/TaskLogSection.vue", "a/WorkspaceTaskPanel.vue"],
|
|
"file_count": 5, "size": 6,
|
|
"consumers": {"count": 6, "paths": ["a/TaskLogSection.vue", "a/V.vue"]}}},
|
|
{"symbol": "btn-primary", "kind": "css", "key": "canon:2855",
|
|
"canon": {"snippet_id": 2855, "path": "frontend/src/assets/components.css",
|
|
"label": ".btn-primary"}},
|
|
{"symbol": "slugify", "kind": "sym", "key": "dup:483a",
|
|
"family": {"group": "dup:483a", "label": "slugify", "identical": True,
|
|
"files": ["a/x.py", "a/y.py"], "file_count": 2, "size": 3}},
|
|
{"symbol": "load", "kind": "sym", "key": "name:sym:load",
|
|
"family": {"group": "name:sym:load", "label": "load", "identical": False,
|
|
"files": ["a/X.vue", "a/Y.vue", "a/Z.vue"], "file_count": 3, "size": 4}},
|
|
]
|
|
check = AsyncMock(return_value=found)
|
|
patches = dict(
|
|
get_writepath_config=AsyncMock(return_value=_cfg()),
|
|
semantic_search_notes=AsyncMock(return_value=[]),
|
|
record_retrieval=MagicMock(), owner_names_for=AsyncMock(return_value={}),
|
|
)
|
|
with patch.multiple(pc, **patches), \
|
|
patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_derive", check):
|
|
out = await pc.build_write_path_hint(
|
|
1, "frontend/src/components/New.vue", code=REAL_CODE, project_id=24,
|
|
stamp_shapes=[("css", "log-empty"), ("css", "btn-primary"), ("sym", "slugify"),
|
|
("sym", "load")],
|
|
exclude_derive=["name:sym:load"],
|
|
)
|
|
check.assert_awaited_once_with(24, "frontend/src/components/New.vue",
|
|
[("css", "log-empty"), ("css", "btn-primary"),
|
|
("sym", "slugify"), ("sym", "load")])
|
|
# The excluded family is gone; the other three render and are keyed.
|
|
assert [d["key"] for d in out["derive"]] == ["name:css:log-empty", "canon:2855", "dup:483a"]
|
|
assert out["derive_keys"] == ["name:css:log-empty", "canon:2855", "dup:483a"]
|
|
ctx = out["context"]
|
|
assert "Shape ledger at `frontend/src/components/New.vue`" in ctx
|
|
# A CSS family is a repeated NAME (note 2917) and its dismissal is scoped-css;
|
|
# a code dup family is an identical body and dismisses as convention-plumbing.
|
|
# A css family says what renders it (milestone 302) before the ask.
|
|
assert "`.log-empty` is a repeated name with no canon — defined in 5 other file(s): " \
|
|
"`a/TaskLogSection.vue`, `a/WorkspaceTaskPanel.vue` +3 more; used by 6 templates: " \
|
|
"`a/TaskLogSection.vue`, `a/V.vue` +4 more; derive it now" in ctx
|
|
assert "`slugify` is a duplicate family with no canon — identical body in 2 other file(s): " \
|
|
"`a/x.py`, `a/y.py`; derive it now" in ctx
|
|
assert "`.btn-primary` is canon — snippet #2855 at `frontend/src/assets/components.css`" in ctx
|
|
assert 'reason_code=\"scoped-css\")` dismisses' in ctx
|
|
assert 'reason_code=\"convention-plumbing\")` dismisses' in ctx
|
|
assert "`load`" not in ctx
|
|
# No project → no check; a failing check never sinks the hint.
|
|
check.reset_mock()
|
|
with patch.multiple(pc, **patches), \
|
|
patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_derive", check):
|
|
out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, stamp_shapes=[("sym", "f")])
|
|
check.assert_not_awaited()
|
|
assert out["derive"] == [] and out["derive_keys"] == []
|
|
boom = AsyncMock(side_effect=RuntimeError("ledger down"))
|
|
with patch.multiple(pc, **patches), \
|
|
patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_derive", boom):
|
|
out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, project_id=24,
|
|
stamp_shapes=[("sym", "f")])
|
|
assert out["context"] == "" and out["derive"] == []
|
|
|
|
|
|
def test_the_hook_keeps_a_derive_channel_and_sends_it_back(tmp_path):
|
|
"""#2900: derive keys the server returns land in the session's own
|
|
`.derive.ids` file and go back as `exclude_derive` on the next write —
|
|
a family is named once per session, not at every edit."""
|
|
reply = (b'{"context":"> family","note_ids":[],"sync_note_ids":[],'
|
|
b'"derive_keys":["dup:483a","canon:2855"]}')
|
|
with http_sink(reply) as (port, seen):
|
|
env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}",
|
|
TMPDIR=str(tmp_path))
|
|
payload = {"session_id": "s-derive-1", "cwd": str(tmp_path), "tool_name": "Write",
|
|
"tool_input": {"file_path": str(tmp_path / "a.css"),
|
|
"content": ".log-empty {\n color: red;\n}\n"}}
|
|
for _ in range(2):
|
|
out = subprocess.run(["bash", str(HOOK)], input=json.dumps(payload),
|
|
capture_output=True, text=True, env=env)
|
|
assert out.returncode == 0, out.stderr
|
|
assert "exclude_derive" not in seen[0]
|
|
assert seen[1]["exclude_derive"] == ["dup:483a,canon:2855"]
|
|
state = tmp_path / "scribe-priorart" / "s-derive-1.derive.ids"
|
|
assert state.read_text().split() == ["dup:483a", "canon:2855", "dup:483a", "canon:2855"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_write_time_divergence_check_is_named_in_band():
|
|
"""#2793: the hook named a shape at a path whose directory a canon
|
|
dominates, and the stamp didn't make it that canon's instance — the hint
|
|
must say so at the write, even when nothing else renders."""
|
|
from scribe.services import plugin_context as pc
|
|
div = [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": 2761,
|
|
"instances": 20, "judged": 21}]
|
|
check = AsyncMock(return_value=div)
|
|
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()), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_divergence", check):
|
|
out = await pc.build_write_path_hint(
|
|
1, "frontend/src/components/Danger.vue", code=REAL_CODE, project_id=24,
|
|
stamp_shapes=[("sym", "confirmDanger")],
|
|
)
|
|
check.assert_awaited_once_with(24, "frontend/src/components/Danger.vue",
|
|
[("sym", "confirmDanger")], [])
|
|
assert out["divergence"] == div
|
|
assert "Divergence check at `frontend/src/components/Danger.vue`" in out["context"]
|
|
assert "`confirmDanger` → #2761 (20 of 21 judged siblings are its instances)" in out["context"]
|
|
assert "variant" in out["context"]
|
|
# No project → no check at all.
|
|
check.reset_mock()
|
|
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()), \
|
|
patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
|
|
patch.object(pc.shape_ledger_svc, "write_time_divergence", check):
|
|
out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, stamp_shapes=[("sym", "f")])
|
|
check.assert_not_awaited()
|
|
assert out["divergence"] == []
|