CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 44s
CI & Build / TypeScript typecheck (push) Successful in 53s
CI & Build / Python tests (push) Failing after 1m7s
CI & Build / Build & push image (push) Skipped
CI 7124 Python tests: 14 failed. One root cause behind ten of them, and the failure was the exact one `tests/helpers.writepath_cfg`'s docstring already warns about in prose — while being unable to prevent this instance of it. Three arms read their numbers out of that config dict inside a fail-open `except`. A missing key raises where nobody sees it, so the arm becomes a silent no-op, indistinguishable from the arm working and finding nothing. The helper derives its keys from `retrieval_surfaces.SURFACES` precisely to stop that — and `checkpoint_threshold` is deliberately NOT a surface, because everything in that table is a floor/budget pair belonging to one query and the checkpoint runs none. The derivation therefore could not see it, the write-path rule arm died before `record_retrieval`, and ten tests went red at once. Fixed at the helper, from the module constant, so there is still exactly one literal and it lives in the product. And the guard the docstring claimed now exists: `test_the_config_stand_in_carries_every_key_the_real_one_does` compares the stand-in's key set against the real `get_writepath_config`, so the next key that is not a surface fails loudly here instead of quietly disabling an arm under test. `test_retrieval_surfaces`'s hand-written key list gains it for the same reason, spelled out in place. The other four were the contract widening itself: `checkpoint` is present on every return of the tool arm, including its early ones, so four assertions comparing the whole dict needed it. That key is deliberately always present — the two arms feed one shell reader where an absent key and an empty one are read the same, so the difference is invisible exactly where it would bite. Verified statically: every cfg in the suite now routes through `writepath_cfg` (test_write_path_trigger's local `_cfg` delegates to it), no hand-written config dict survives, and the real config's eight keys match the stand-in's. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy
555 lines
26 KiB
Python
555 lines
26 KiB
Python
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from scribe.services import plugin_context as pc_module
|
|
from scribe.services import retrieval_surfaces as rs
|
|
from scribe.services.lessons import LESSON_NOTE_TYPE
|
|
from tests.helpers import fake_note, writepath_cfg
|
|
|
|
|
|
pytestmark = pytest.mark.usefixtures("_no_supersession")
|
|
|
|
|
|
|
|
|
|
# ─── knowledge auto-inject (Path A) ──────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_autoinject_config_defaults_and_clamps():
|
|
from scribe.services import plugin_context as pc
|
|
|
|
# No settings stored → defaults.
|
|
with patch.object(pc, "get_setting", AsyncMock(side_effect=lambda uid, k, d: d)), \
|
|
patch.object(rs, "get_setting", AsyncMock(side_effect=lambda uid, k, d="": d)):
|
|
cfg = await pc.get_autoinject_config(1)
|
|
assert cfg == {
|
|
"enabled": pc.AUTOINJECT_DEFAULT_ENABLED,
|
|
"threshold": pc.AUTOINJECT_DEFAULT_THRESHOLD,
|
|
"top_k": pc.AUTOINJECT_DEFAULT_TOP_K,
|
|
}
|
|
|
|
# Out-of-range values are clamped; top_k capped at the hard ceiling.
|
|
stored = {
|
|
pc.AUTOINJECT_ENABLED_KEY: "false",
|
|
pc.AUTOINJECT_THRESHOLD_KEY: "5",
|
|
pc.AUTOINJECT_TOP_K_KEY: "999",
|
|
}
|
|
# The switch comes from plugin_context, the two numbers from the registry.
|
|
def _side(uid, k, d=""):
|
|
return stored.get(k, d)
|
|
|
|
with patch.object(pc, "get_setting", AsyncMock(side_effect=_side)), \
|
|
patch.object(rs, "get_setting", AsyncMock(side_effect=_side)):
|
|
cfg = await pc.get_autoinject_config(1)
|
|
assert cfg["enabled"] is False
|
|
assert cfg["threshold"] == 1.0
|
|
assert cfg["top_k"] == pc._AUTOINJECT_MAX_TOP_K
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_autoinject_hint_disabled_returns_empty_and_skips_search():
|
|
from scribe.services import plugin_context as pc
|
|
search = AsyncMock()
|
|
with patch.object(pc, "get_autoinject_config",
|
|
AsyncMock(return_value={"enabled": False, "threshold": 0.55, "top_k": 3})), \
|
|
patch.object(pc, "semantic_search_notes", search), \
|
|
patch.object(pc, "record_retrieval", MagicMock()):
|
|
out = await pc.build_autoinject_hint(1, "anything")
|
|
assert out["context"] == "" and out["note_ids"] == []
|
|
search.assert_not_called() # disabled → no retrieval at all
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_autoinject_hint_titles_only_with_margin_gate():
|
|
from scribe.services import plugin_context as pc
|
|
# top=0.80; 0.74 within band (0.10), 0.61 outside → dropped.
|
|
hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
|
|
(0.74, fake_note(id=22, title="run_maintenance thresholds", user_id=1)),
|
|
(0.61, fake_note(id=33, title="unrelated-ish", user_id=1))]
|
|
rec = MagicMock()
|
|
with patch.object(pc, "get_autoinject_config",
|
|
AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \
|
|
patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \
|
|
patch.object(pc, "record_retrieval", rec):
|
|
out = await pc.build_autoinject_hint(1, "postgres pool", project_id=2,
|
|
exclude_ids=[99])
|
|
# Margin gate kept the top two, dropped the straggler.
|
|
assert out["note_ids"] == [11, 22]
|
|
assert '#11 [note] "Pool sizing decision" (0.80)' in out["context"]
|
|
assert "#33" not in out["context"]
|
|
# Title-first: no body text, ever.
|
|
assert "get_note(id)" in out["context"]
|
|
# Telemetry fired for BOTH retrievals this path runs: the scored menu and
|
|
# the reuse-slot query competing against it. The slot's query used to be
|
|
# the one unlogged retrieval on this path — the hit it displaced was in
|
|
# retrieval_logs, the query that displaced it was not (#2463).
|
|
sources = [c.kwargs["source"] for c in rec.call_args_list]
|
|
assert sources == ["auto_inject", "reuse_slot", "lesson_slot"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_autoinject_hint_blank_query_returns_empty():
|
|
from scribe.services import plugin_context as pc
|
|
search = AsyncMock()
|
|
with patch.object(pc, "get_autoinject_config",
|
|
AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \
|
|
patch.object(pc, "semantic_search_notes", search):
|
|
out = await pc.build_autoinject_hint(1, " ")
|
|
assert out["context"] == ""
|
|
search.assert_not_called()
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_session_context_includes_project_when_scoped():
|
|
# design_system_id explicitly None: a bare MagicMock would hand back a truthy
|
|
# auto-attribute and send this through the design branch, which is the
|
|
# opposite of what this test is about.
|
|
project = MagicMock(id=2, title="FabledScribe", goal="ship it",
|
|
design_system_id=None)
|
|
with patch("scribe.services.plugin_context.projects_svc.get_project",
|
|
AsyncMock(return_value=project)), \
|
|
patch("scribe.services.plugin_context.notes_svc.list_notes",
|
|
AsyncMock(return_value=([], 4))):
|
|
from scribe.services.plugin_context import build_session_context
|
|
out = await build_session_context(user_id=7, project_id=2)
|
|
|
|
assert out["project"] == {"id": 2, "title": "FabledScribe"}
|
|
assert "## Active project: FabledScribe (id 2)" in out["context"]
|
|
assert "Open todo tasks: 4" in out["context"]
|
|
# No design system on the project -> no design block at all. An install with
|
|
# none is the ordinary case, not a degraded one.
|
|
assert "## Design system" not in out["context"]
|
|
# Live state only (decision #4027): how to work with Scribe is the
|
|
# using-scribe skill's to say. A restated reflex here is a copy that drifts.
|
|
assert 'content_type="rule"' not in out["context"]
|
|
assert "Reflex:" not in out["context"]
|
|
|
|
|
|
def test_a_short_goal_is_shown_whole():
|
|
from scribe.services.plugin_context import _goal_line
|
|
assert _goal_line("ship it", 2) == "Goal: ship it"
|
|
assert _goal_line("", 2) == ""
|
|
|
|
|
|
def test_a_long_goal_is_cut_at_a_word_and_says_where_the_rest_is():
|
|
"""#4036: a raw slice ended mid-word with nothing marking the cut, so a
|
|
reader took half a sentence for the whole goal."""
|
|
from scribe.services.plugin_context import _GOAL_CHARS, _goal_line
|
|
goal = "make the record reach the next session\nso a solution is recalled " * 10
|
|
line = _goal_line(goal, 2)
|
|
shown = line.removeprefix("Goal: ").split(" (full goal:")[0]
|
|
assert shown.endswith("…") and len(shown) <= _GOAL_CHARS
|
|
# The cut lands between words: what precedes the ellipsis is a whole word.
|
|
assert shown[:-1].rstrip().split()[-1] in goal.split()
|
|
assert "\n" not in line
|
|
assert line.endswith("(full goal: `enter_project(2)`)")
|
|
|
|
|
|
def test_an_unbroken_goal_still_shows_its_start():
|
|
from scribe.services.plugin_context import _GOAL_CHARS, _goal_line
|
|
shown = _goal_line("x" * 500, 2).removeprefix("Goal: ").split(" (full goal:")[0]
|
|
assert shown == "x" * (_GOAL_CHARS - 1) + "…"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_session_context_pushes_the_projects_design_system():
|
|
"""The gap this closes: a design system had no push channel, so its
|
|
standards reached a session only if the agent already knew to go looking —
|
|
the same silent failure as a token nobody declares."""
|
|
project = MagicMock(id=2, title="App", goal="", design_system_id=9)
|
|
design = {
|
|
"id": 9, "title": "App kit", "description": "",
|
|
"inherits_from": ["House"],
|
|
"guidance": [], "token_count": 95,
|
|
"token_groups": ["accent", "surface", "type"],
|
|
}
|
|
with patch("scribe.services.plugin_context.projects_svc.get_project",
|
|
AsyncMock(return_value=project)), \
|
|
patch("scribe.services.plugin_context.notes_svc.list_notes",
|
|
AsyncMock(return_value=([], 0))), \
|
|
patch("scribe.services.plugin_context.design_systems_svc.design_context",
|
|
AsyncMock(return_value=design)):
|
|
from scribe.services.plugin_context import build_session_context
|
|
out = await build_session_context(user_id=7, project_id=2)
|
|
|
|
ctx = out["context"]
|
|
assert "## Design system: App kit (id 9) (inherits House)" in ctx
|
|
assert "95 tokens across accent, surface, type" in ctx
|
|
# A pointer to the values, never the values themselves — a hundred token
|
|
# declarations would crowd out the context they are meant to inform. The
|
|
# summary carries counts and group NAMES only, which is why design_context
|
|
# returns those rather than the resolved tokens.
|
|
assert "resolve_design_system(9)" in ctx
|
|
assert "get_design_system_stylesheet(9)" in ctx
|
|
# The prose pointer names the call that has the MERGED guidance, not
|
|
# enter_project, which carries only the summary now (#4045).
|
|
assert "`get_design_system(9)` → `resolved_guidance`" in ctx
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_session_context_survives_an_unreadable_design_system():
|
|
"""design_context returns None when the caller may not read the system.
|
|
That must degrade to "no design block", not to a crash that costs the
|
|
session its rules too."""
|
|
project = MagicMock(id=2, title="App", goal="", design_system_id=9)
|
|
with patch("scribe.services.plugin_context.projects_svc.get_project",
|
|
AsyncMock(return_value=project)), \
|
|
patch("scribe.services.plugin_context.notes_svc.list_notes",
|
|
AsyncMock(return_value=([], 0))), \
|
|
patch("scribe.services.plugin_context.design_systems_svc.design_context",
|
|
AsyncMock(return_value=None)):
|
|
from scribe.services.plugin_context import build_session_context
|
|
out = await build_session_context(user_id=7, project_id=2)
|
|
|
|
assert "## Design system" not in out["context"]
|
|
assert "## Active project: App (id 2)" in out["context"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_session_context_unbound_repo_emits_bind_hint():
|
|
from scribe.services.plugin_context import build_session_context
|
|
out = await build_session_context(
|
|
user_id=7, project_id=0, unbound_repo="host/owner/repo",
|
|
)
|
|
|
|
ctx = out["context"]
|
|
assert out["project"] is None
|
|
assert "## Repository not yet bound" in ctx
|
|
assert 'bind_repo(repo_url="host/owner/repo"' in ctx
|
|
# No project block when unbound.
|
|
assert "## Active project" not in ctx
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_project_id_that_does_not_resolve_is_reported_not_dropped():
|
|
"""A `.scribe` marker names a project directly (#4085), so for the first
|
|
time a caller arrives holding a pointer it BELIEVES in. If the id is for
|
|
another instance, or names a deleted project, the session has to be told —
|
|
it cannot infer it from an absence.
|
|
|
|
This used to render nothing whatsoever: the branch hung off `if project_id`
|
|
as an `elif`, so an id that was sent and failed took the outer arm, found
|
|
no project, and fell out of the block having said nothing at all.
|
|
"""
|
|
from scribe.services.plugin_context import build_session_context
|
|
with patch("scribe.services.plugin_context.projects_svc.get_project",
|
|
AsyncMock(return_value=None)):
|
|
out = await build_session_context(user_id=7, project_id=41)
|
|
|
|
ctx = out["context"]
|
|
assert out["project"] is None
|
|
assert "## Project 41 could not be loaded" in ctx
|
|
assert "list_projects" in ctx
|
|
# Not mistaken for the repo case, which has a different remedy.
|
|
assert "## Repository not yet bound" not in ctx
|
|
assert "No Scribe project is bound" not in ctx
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_process_manifest_renders_stub_specs():
|
|
items = [
|
|
{"id": 5, "title": "Drift Audit", "tags": [], "snippet": "Find drifted docs."},
|
|
{"id": 9, "title": "DRY Pass", "tags": [], "snippet": ""},
|
|
]
|
|
with patch("scribe.services.plugin_context.knowledge_svc.query_knowledge",
|
|
AsyncMock(return_value=(items, 2))):
|
|
from scribe.services.plugin_context import build_process_manifest
|
|
out = await build_process_manifest(user_id=7)
|
|
|
|
assert out["total"] == 2
|
|
drift = out["processes"][0]
|
|
assert drift["id"] == 5
|
|
assert drift["name"] == "Drift Audit"
|
|
assert drift["slug"] == "drift-audit" # kebab-cased
|
|
assert "Drift Audit" in drift["description"] # auto-surface trigger
|
|
assert "Find drifted docs." in drift["description"] # preview folded in
|
|
# No-snippet process still gets a usable description.
|
|
assert "DRY Pass" in out["processes"][1]["description"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_process_manifest_dedupes_slugs_and_skips_blank_titles():
|
|
items = [
|
|
{"id": 1, "title": "My Process", "tags": [], "snippet": "a"},
|
|
{"id": 2, "title": "my process", "tags": [], "snippet": "b"}, # same slug
|
|
{"id": 3, "title": " ", "tags": [], "snippet": "skip me"}, # blank title
|
|
]
|
|
with patch("scribe.services.plugin_context.knowledge_svc.query_knowledge",
|
|
AsyncMock(return_value=(items, 3))):
|
|
from scribe.services.plugin_context import build_process_manifest
|
|
out = await build_process_manifest(user_id=7)
|
|
|
|
slugs = [p["slug"] for p in out["processes"]]
|
|
assert slugs == ["my-process", "my-process-2"] # collision suffixed with id
|
|
assert out["total"] == 2 # blank-title entry dropped
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_process_manifest_truncates_long_preview():
|
|
items = [{"id": 1, "title": "Big", "tags": [], "snippet": "x" * 500}]
|
|
with patch("scribe.services.plugin_context.knowledge_svc.query_knowledge",
|
|
AsyncMock(return_value=(items, 1))):
|
|
from scribe.services.plugin_context import build_process_manifest
|
|
out = await build_process_manifest(user_id=7)
|
|
|
|
assert "…" in out["processes"][0]["description"]
|
|
assert "x" * 500 not in out["processes"][0]["description"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_session_context_caps_length():
|
|
"""The cap still binds, and now has to be provoked rather than tripped.
|
|
|
|
This used to hand the block 200 long rule titles, because the preload made
|
|
overflow the easy case. Milestone 394 removed that block, so nothing this
|
|
function assembles on its own is big enough any more — which is a reason
|
|
to drive the cap directly, not a reason to drop it. The project and design
|
|
blocks are still unbounded in principle, and the hook passes this text
|
|
through verbatim.
|
|
|
|
Patching the cap rather than manufacturing 9,000 characters keeps the test
|
|
about the TRUNCATION PATH — that it cuts, and that it says it cut — which
|
|
is the part a reader depends on. The unbound-repo hint supplies the text:
|
|
since milestone 410 the block carries live state only, and a bare session
|
|
is a one-liner.
|
|
"""
|
|
from scribe.services import plugin_context as pc
|
|
|
|
with patch.object(pc, "_MAX_CHARS", 60):
|
|
out = await pc.build_session_context(user_id=7, unbound_repo="host/owner/repo")
|
|
|
|
assert len(out["context"]) <= 60 + 20 # cap + truncation note
|
|
assert "truncated" in out["context"], (
|
|
"the block was cut without saying so — a reader cannot tell a "
|
|
"truncated context from a short one"
|
|
)
|
|
|
|
|
|
# --- the reuse slot (#2246) --------------------------------------------------
|
|
|
|
_CFG = {"enabled": True, "threshold": 0.55, "top_k": 3}
|
|
|
|
|
|
def _asked_for_reuse(calls: list[dict]) -> bool:
|
|
"""Did the reuse slot issue its reserved query on this run?"""
|
|
return any(
|
|
tuple(c.get("note_type") or ()) == pc_module._REUSE_KINDS for c in calls
|
|
)
|
|
|
|
|
|
async def _autoinject(main_hits, reuse_hits, cfg=None, lesson_hits=None):
|
|
"""Run build_autoinject_hint with each semantic query stubbed by the kinds
|
|
it asks for: the unscoped pool, the reserved reuse query, and the reserved
|
|
lesson query (milestone 385 step 5).
|
|
|
|
ROUTED ON THE REQUESTED KINDS, not on call order, and that is the point of
|
|
the helper. Order-keyed stubbing was fine while there was one reserved
|
|
slot; with two it makes every test in this section depend on which slot
|
|
runs first, so adding a third would silently hand one slot another's
|
|
candidate list and the tests would still pass.
|
|
"""
|
|
calls: list[dict] = []
|
|
|
|
async def fake_search(*_a, **kw):
|
|
calls.append(kw)
|
|
kinds = kw.get("note_type") or ()
|
|
if LESSON_NOTE_TYPE in kinds:
|
|
return lesson_hits or []
|
|
if kinds:
|
|
return reuse_hits
|
|
return main_hits
|
|
|
|
with patch("scribe.services.plugin_context.get_autoinject_config",
|
|
AsyncMock(return_value=dict(cfg or _CFG))), \
|
|
patch("scribe.services.plugin_context.semantic_search_notes",
|
|
AsyncMock(side_effect=fake_search)), \
|
|
patch("scribe.services.plugin_context.record_retrieval", MagicMock()), \
|
|
patch("scribe.services.plugin_context.record_surfaced", MagicMock()), \
|
|
patch("scribe.services.plugin_context.owner_names_for",
|
|
AsyncMock(return_value={})):
|
|
from scribe.services.plugin_context import build_autoinject_hint
|
|
out = await build_autoinject_hint(1, "write a debounce helper")
|
|
return out, calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_snippet_takes_the_last_slot_when_none_won_on_score():
|
|
"""The measured failure: a prompt asking for a helper returned three project
|
|
records ABOUT building the retrieval system, and zero snippets. Scribe's own
|
|
records are about software work, so they share vocabulary with any coding
|
|
prompt while answering none of them."""
|
|
main = [(0.66, fake_note(id=1, title="Step 3: title-first auto-inject", user_id=1)),
|
|
(0.65, fake_note(id=2, title="Task-reminder dedup query crashes", user_id=1, is_task=True)),
|
|
(0.64, fake_note(id=3, title="Drafter hardening · write-path trigger", user_id=1, is_task=True))]
|
|
reuse = [(0.58, fake_note(id=9, title="debounce — collapse rapid calls", user_id=1, note_type="snippet"))]
|
|
|
|
out, calls = await _autoinject(main, reuse)
|
|
|
|
assert out["note_ids"] == [1, 2, 9] # last slot displaced, not the first
|
|
assert "#9" in out["context"] and "[snippet]" in out["context"]
|
|
# The reserved query is scoped to the reuse kinds and asks for exactly one.
|
|
assert calls[1]["note_type"] == ("snippet", "process")
|
|
assert calls[1]["limit"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_reserved_query_is_skipped_when_a_snippet_already_won():
|
|
"""No second query, and no slot spent twice, when ranking already did the
|
|
right thing — the fix must be invisible in the case it isn't needed."""
|
|
main = [(0.81, fake_note(id=9, title="debounce helper", user_id=1, note_type="snippet")),
|
|
(0.80, fake_note(id=1, title="some task", user_id=1, is_task=True))]
|
|
|
|
out, calls = await _autoinject(main, [])
|
|
|
|
# WHICH query ran, not how many. A count here is a claim about every
|
|
# reserved slot on this path at once, so it fails the day an unrelated one
|
|
# is added — and the thing being tested is that THIS slot stood down.
|
|
assert not _asked_for_reuse(calls)
|
|
assert out["note_ids"] == [9, 1]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_weak_snippet_does_not_buy_the_slot():
|
|
"""The reserved hit skips the MARGIN band — that band is what snippets lose
|
|
to — but never the threshold. Silence stays the default; a slot spent on an
|
|
irrelevant snippet is how a menu teaches people to ignore it."""
|
|
main = [(0.66, fake_note(id=1, title="a task", user_id=1, is_task=True))]
|
|
|
|
out, calls = await _autoinject(main, []) # threshold returned nothing
|
|
|
|
assert out["note_ids"] == [1]
|
|
assert _asked_for_reuse(calls) # it asked, and got nothing
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_reserved_hit_is_not_held_to_the_margin_band():
|
|
"""0.58 is 0.08 below the top hit. Under the band it would survive; the point
|
|
is that it must survive even when it wouldn't — a snippet losing to a
|
|
same-vocabulary project record by a wide margin is the whole bug."""
|
|
main = [(0.90, fake_note(id=1, title="a task", user_id=1, is_task=True))]
|
|
reuse = [(0.58, fake_note(id=9, title="debounce", user_id=1, note_type="snippet"))]
|
|
|
|
out, _calls = await _autoinject(main, reuse)
|
|
|
|
assert 9 in out["note_ids"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_process_counts_as_reuse_too():
|
|
"""A stored process answers 'how do we do X here' the same way a snippet
|
|
answers 'what do we already have' — both lose to the same project records."""
|
|
main = [(0.70, fake_note(id=1, title="a task", user_id=1, is_task=True))]
|
|
reuse = [(0.60, fake_note(id=8, title="DRY pass process", user_id=1, note_type="process"))]
|
|
|
|
out, _ = await _autoinject(main, reuse)
|
|
assert 8 in out["note_ids"]
|
|
|
|
# …and one already on the menu suppresses the reserved query.
|
|
out2, calls2 = await _autoinject(
|
|
[(0.70, fake_note(id=8, title="DRY pass process", user_id=1, note_type="process"))], [])
|
|
assert not _asked_for_reuse(calls2)
|
|
|
|
|
|
# --- write-path widened beyond snippets (#2246, the mirror half) -------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_path_semantic_arm_asks_for_experience_not_just_snippets():
|
|
"""The inverse of auto-inject's mistake. This arm was snippets-only, so an
|
|
issue saying "we tried this and it deadlocked" could never reach the moment
|
|
that code was about to be written — arguably the better prior art, because
|
|
it says what NOT to do."""
|
|
from scribe.services import plugin_context as pc
|
|
|
|
hits = [(0.72, fake_note(id=9, title="debounce helper", user_id=1, note_type="snippet")),
|
|
(0.70, fake_note(id=7, title="Debounce dropped the trailing call", user_id=1, is_task=True, task_kind="issue"))]
|
|
search = AsyncMock(return_value=hits)
|
|
rec = MagicMock()
|
|
with patch.object(pc, "get_writepath_config",
|
|
AsyncMock(return_value=writepath_cfg(threshold=0.6))), \
|
|
patch.object(pc.snippets_svc, "list_snippets",
|
|
AsyncMock(return_value=([], 0))), \
|
|
patch.object(pc, "semantic_search_notes", search), \
|
|
patch.object(pc, "record_retrieval", rec), \
|
|
patch.object(pc, "record_surfaced", MagicMock()), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
|
|
patch.object(pc, "concept_query", MagicMock(return_value="debounce a callback")):
|
|
out = await pc.build_write_path_hint(
|
|
1, "src/utils/debounce.ts", code="x" * 400,
|
|
)
|
|
|
|
kw = search.await_args.kwargs
|
|
assert kw["note_type"] == ("snippet", "note", LESSON_NOTE_TYPE)
|
|
# An open to-do resembling the code answers nothing; an ISSUE carries a root
|
|
# cause and a NOTE carries durable knowledge. Only the todo is excluded.
|
|
assert kw["task_kind"] == "issue"
|
|
# Telemetry must not claim this was a notes-only retrieval any more.
|
|
assert rec.call_args.kwargs["is_task"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_path_labels_a_non_snippet_hit_with_its_kind():
|
|
"""An unlabelled issue on this menu reads as "here is code to reuse", which
|
|
is the opposite of what it says. A snippet stays unlabelled — it is the
|
|
menu's default and the header's default reading."""
|
|
from scribe.services import plugin_context as pc
|
|
|
|
hits = [(0.72, fake_note(id=9, title="debounce helper", user_id=1, note_type="snippet")),
|
|
(0.71, fake_note(id=7, title="Debounce dropped the trailing call", user_id=1, is_task=True, task_kind="issue"))]
|
|
with patch.object(pc, "get_writepath_config",
|
|
AsyncMock(return_value=writepath_cfg(threshold=0.6))), \
|
|
patch.object(pc.snippets_svc, "list_snippets",
|
|
AsyncMock(return_value=([], 0))), \
|
|
patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \
|
|
patch.object(pc, "record_retrieval", MagicMock()), \
|
|
patch.object(pc, "record_surfaced", MagicMock()), \
|
|
patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
|
|
patch.object(pc, "concept_query", MagicMock(return_value="debounce a callback")):
|
|
out = await pc.build_write_path_hint(
|
|
1, "src/utils/debounce.ts", code="x" * 400,
|
|
)
|
|
|
|
ctx = out["context"]
|
|
# The issue says what it is — and, since #4154, where it stands: a piece of
|
|
# prior art offered as "already tried" reads differently when it is still open.
|
|
assert "· issue (todo)]" in ctx
|
|
assert '[similar 0.72] "debounce helper"' in ctx # the snippet does not
|
|
# The header now names the right opener for each kind.
|
|
assert "get_task(id)" in ctx and "get_snippet(id)" in ctx
|
|
|
|
|
|
# ── The config stand-in cannot fall behind the real one (#4214) ───────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_config_stand_in_carries_every_key_the_real_one_does():
|
|
"""THE GUARD `writepath_cfg`'s DOCSTRING ALREADY CLAIMED AND DID NOT HAVE.
|
|
|
|
Three arms read their numbers out of that dict inside a fail-open
|
|
`except`, so a missing key does not raise where anyone can see it — the
|
|
arm silently becomes a no-op, which is indistinguishable from the arm
|
|
working and finding nothing. The helper derives its SURFACE keys from the
|
|
registry to prevent exactly that, and then #4214 added a key that is
|
|
deliberately not a surface: the whole derivation missed it, ten tests went
|
|
red at once, and the diagnosis cost a CI round.
|
|
|
|
Asserting the KEY SETS match, not the values: the stand-in exists to let a
|
|
test set different numbers.
|
|
"""
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from scribe.services import plugin_context as pc
|
|
from tests.helpers import writepath_cfg
|
|
|
|
with patch.object(pc, "get_setting", AsyncMock(return_value="0.6")), \
|
|
patch.object(pc, "floor_for", AsyncMock(return_value=0.6)), \
|
|
patch.object(pc, "budget_for", AsyncMock(return_value=3)):
|
|
real = await pc.get_writepath_config(1)
|
|
|
|
assert set(writepath_cfg()) == set(real), (
|
|
"tests/helpers.writepath_cfg has fallen behind get_writepath_config; "
|
|
"a key the real config has and the stand-in does not turns an arm "
|
|
"into a silent no-op under test"
|
|
)
|