bvandeusenandClaude Opus 5.5
7c1d98dd2a
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / TypeScript typecheck (push) Successful in 53s
CI & Build / integration (push) Successful in 1m2s
CI & Build / Python tests (push) Failing after 1m26s
CI & Build / Build & push image (push) Skipped
A rule arm logs its query, floor, budget and project, which is everything semantic_search_rules takes, so a logged call is re-run to a depth past its budget and judged like an auto_inject menu. within_budget is the arm's own cut: inside the budget and, for the banded command and write arms, inside the band of the top hit. "Opened after" reads rule pulls, and rules created after the call are dropped as notes are. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
391 lines
16 KiB
Python
391 lines
16 KiB
Python
"""The review pass: judging whether an injected line RELATED.
|
|
|
|
Unit tests pin the refusals and the re-run's fidelity to the arm it replays.
|
|
Integration tests pin the parts a mock would make true by construction: which
|
|
logged calls are offered, that a verdict lands and replaces, that "opened
|
|
after" reads the usage table, and that the readout counts by rank.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from grimoire.models import async_session
|
|
from grimoire.services import retrieval_review as review
|
|
from tests.helpers import ensure_user, fake_note
|
|
|
|
|
|
# ── refusals ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_only_an_arm_that_can_be_re_run_exactly_is_offered():
|
|
with pytest.raises(ValueError, match="source must be one of"):
|
|
await review.menus_to_review(1, source="write_path")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_verdict_outside_the_vocabulary_is_refused():
|
|
with pytest.raises(ValueError, match="verdict must be one of"):
|
|
await review.judge_menu(1, log_id=1, verdicts=[
|
|
{"record_id": 5, "verdict": "relevant", "reason": "x"},
|
|
])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_verdict_without_its_reason_is_refused():
|
|
with pytest.raises(ValueError, match="needs its reason"):
|
|
await review.judge_menu(1, log_id=1, verdicts=[
|
|
{"record_id": 5, "verdict": "unrelated", "reason": " "},
|
|
])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_empty_verdict_list_is_refused():
|
|
with pytest.raises(ValueError, match="at least one"):
|
|
await review.judge_menu(1, log_id=1, verdicts=[])
|
|
|
|
|
|
# ── the re-run replays the arm ───────────────────────────────────────────
|
|
|
|
|
|
_FILTERS = ("note_type", "task_kind", "include_global_kinds", "scope")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_re_run_searches_the_way_the_arm_does():
|
|
"""A re-run with different visibility or kinds from the arm's would judge
|
|
a menu nobody was shown. Both searches are RUN and their keywords
|
|
compared, so a change to the arm's spec fails here until the review
|
|
follows it — and the arm has to have filters for the check to mean
|
|
anything."""
|
|
from grimoire.services import retrieval_pipeline as rp
|
|
|
|
arm_kw: dict = {}
|
|
|
|
async def arm_search(_uid, _q, **kw):
|
|
arm_kw.update(kw)
|
|
return []
|
|
|
|
await rp.run_note_arm(
|
|
rp.AUTO_INJECT, rp.NoteMoment(user_id=1, query="q", project_id=None),
|
|
floor=0.5, budget=3,
|
|
io=rp.NoteIO(search=arm_search, record_retrieval=MagicMock(),
|
|
record_surfaced=MagicMock()),
|
|
)
|
|
rerun_kw: dict = {}
|
|
|
|
async def rerun_search(_uid, _q, **kw):
|
|
rerun_kw.update(kw)
|
|
return []
|
|
|
|
row = SimpleNamespace(source="auto_inject", query="q", threshold=0.5, limit_n=3, project_id=None,
|
|
created_at=CALL)
|
|
with patch("grimoire.services.embeddings.semantic_search_notes", side_effect=rerun_search):
|
|
await review._rerun(1, row, 3)
|
|
|
|
arm = {k: arm_kw[k] for k in _FILTERS if k in arm_kw}
|
|
assert "scope" in arm and "include_global_kinds" in arm, (
|
|
f"the arm no longer passes its visibility filters: {arm_kw}"
|
|
)
|
|
assert {k: rerun_kw[k] for k in _FILTERS if k in rerun_kw} == arm
|
|
|
|
|
|
CALL = datetime(2026, 9, 20, 12, 0, tzinfo=timezone.utc)
|
|
BEFORE = CALL - timedelta(days=1)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_re_run_ranks_from_one_and_marks_the_budget_cut():
|
|
row = SimpleNamespace(source="auto_inject", query="q", threshold=0.55, limit_n=2, project_id=None,
|
|
created_at=CALL)
|
|
hits = [
|
|
(0.81, fake_note(id=11, title="first", created_at=BEFORE, updated_at=BEFORE)),
|
|
(0.74, fake_note(id=12, title="second", created_at=BEFORE, updated_at=BEFORE)),
|
|
(0.66, fake_note(id=13, title="third", created_at=BEFORE, updated_at=BEFORE)),
|
|
]
|
|
|
|
async def search(user_id, query, **kw):
|
|
assert kw["limit"] == 3 + review.POSTDATED_SLACK and kw["threshold"] == 0.55
|
|
assert kw["project_id"] is None
|
|
kw["report"]["best_chunk"] = {12: {"text": "second\nthe passage that matched"}}
|
|
return hits
|
|
|
|
with patch("grimoire.services.embeddings.semantic_search_notes", side_effect=search):
|
|
lines = await review._rerun(1, row, 3)
|
|
assert [(ln["rank"], ln["record_id"], ln["within_budget"]) for ln in lines] == [
|
|
(1, 11, True), (2, 12, True), (3, 13, False),
|
|
]
|
|
assert lines[1]["passage"] == "the passage that matched"
|
|
assert lines[0]["passage"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_record_written_after_the_call_is_dropped_before_ranking():
|
|
"""The session that made a call goes on to write about it, and that record
|
|
outranks everything in a re-run. The call could never have been offered
|
|
it, so it must not take a rank — or a verdict — from what was."""
|
|
row = SimpleNamespace(source="auto_inject", query="q", threshold=0.55, limit_n=1, project_id=None,
|
|
created_at=CALL)
|
|
after = CALL + timedelta(hours=1)
|
|
hits = [
|
|
(0.90, fake_note(id=21, title="the decision", created_at=after, updated_at=after)),
|
|
(0.80, fake_note(id=22, title="held then", created_at=BEFORE, updated_at=BEFORE)),
|
|
(0.70, fake_note(id=23, title="edited since", created_at=BEFORE, updated_at=after)),
|
|
]
|
|
|
|
async def search(user_id, query, **kw):
|
|
return hits
|
|
|
|
rep: dict = {}
|
|
with patch("grimoire.services.embeddings.semantic_search_notes", side_effect=search):
|
|
lines = await review._rerun(1, row, 2, report=rep)
|
|
assert rep["postdated"] == [21]
|
|
assert [(ln["rank"], ln["record_id"], ln["within_budget"], ln["changed_since_call"])
|
|
for ln in lines] == [(1, 22, True, False), (2, 23, False, True)]
|
|
|
|
|
|
# ── the rule arms ────────────────────────────────────────────────────────
|
|
|
|
|
|
def _rule(rid, title, created=BEFORE, updated=BEFORE):
|
|
return SimpleNamespace(
|
|
id=rid, title=title, kind="rule", when_to_apply=f"when {title}",
|
|
statement="s", created_at=created, updated_at=updated,
|
|
)
|
|
|
|
|
|
def test_the_reviewable_rule_arms_are_the_ones_that_annotate():
|
|
"""Every rule arm that shows lines can be reviewed; the reply backstop
|
|
shows none, so there is no menu to judge."""
|
|
from grimoire.services import retrieval_pipeline as rp
|
|
|
|
assert set(review.RULE_REVIEWABLE) == {
|
|
a.source for a in rp.RULE_ARMS if not a.stop_only
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_rule_re_run_searches_the_way_the_arm_does():
|
|
"""The floor and the project the arm searched with, run and compared."""
|
|
from grimoire.services import retrieval_pipeline as rp
|
|
|
|
arm_kw: dict = {}
|
|
|
|
async def arm_search(_uid, _q, **kw):
|
|
if "kind" not in kw:
|
|
arm_kw.update(kw)
|
|
return []
|
|
|
|
await rp.run_rule_arm(
|
|
rp.PROMPT_RULE, rp.RuleMoment(user_id=1, query="q", project_id=7),
|
|
floor=0.7, budget=3,
|
|
io=rp.RuleIO(search=arm_search, record_retrieval=MagicMock(),
|
|
record_rule_surfaced=MagicMock()),
|
|
)
|
|
rerun_kw: dict = {}
|
|
|
|
async def rerun_search(_uid, _q, **kw):
|
|
rerun_kw.update(kw)
|
|
return []
|
|
|
|
row = SimpleNamespace(source="prompt_rule", query="q", threshold=0.7, limit_n=3,
|
|
project_id=7, created_at=CALL)
|
|
with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=rerun_search):
|
|
await review._rerun(1, row, 3)
|
|
|
|
for key in ("threshold", "project_id"):
|
|
assert rerun_kw[key] == arm_kw[key], key
|
|
assert "kind" not in rerun_kw
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(("source", "within"), [
|
|
# No band: the budget alone is the cut.
|
|
("prompt_rule", [True, True, True, False]),
|
|
# Banded: the third sits outside the band of the top hit, so it was
|
|
# not shown though the budget had room for it.
|
|
("pre_tool_rule", [True, True, False, False]),
|
|
])
|
|
async def test_a_rule_line_is_inside_the_cut_only_where_the_arm_showed_it(source, within):
|
|
row = SimpleNamespace(source=source, query="git push", threshold=0.7, limit_n=3,
|
|
project_id=None, created_at=CALL)
|
|
hits = [(0.86, _rule(1, "one")), (0.84, _rule(2, "two")),
|
|
(0.76, _rule(3, "three")), (0.74, _rule(4, "four"))]
|
|
|
|
async def search(_uid, _q, **kw):
|
|
assert kw["threshold"] == 0.7 and kw["project_id"] is None
|
|
return hits
|
|
|
|
with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=search):
|
|
lines = await review._rerun(1, row, 4)
|
|
assert [ln["within_budget"] for ln in lines] == within
|
|
assert lines[0]["passage"] == "when one"
|
|
assert lines[0]["kind"] == "rule"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_rule_written_after_the_call_is_dropped_before_ranking():
|
|
row = SimpleNamespace(source="write_path_rule", query="q", threshold=0.7, limit_n=2,
|
|
project_id=None, created_at=CALL)
|
|
after = CALL + timedelta(hours=1)
|
|
hits = [(0.9, _rule(31, "written after", created=after, updated=after)),
|
|
(0.8, _rule(32, "held then"))]
|
|
|
|
async def search(_uid, _q, **kw):
|
|
return hits
|
|
|
|
rep: dict = {}
|
|
with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=search):
|
|
lines = await review._rerun(1, row, 2, report=rep)
|
|
assert rep["postdated"] == [31]
|
|
assert [(ln["rank"], ln["record_id"]) for ln in lines] == [(1, 32)]
|
|
|
|
|
|
# ── against Postgres ─────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def logged_calls():
|
|
"""A reviewer's logged calls: two reviewable, one that offered nothing,
|
|
one of another arm — and another user's reviewable call."""
|
|
from grimoire.models.retrieval_log import RetrievalLog
|
|
|
|
async with async_session() as s:
|
|
me = await ensure_user(s, "review_owner")
|
|
other = await ensure_user(s, "review_other")
|
|
now = datetime.now(timezone.utc)
|
|
|
|
def log(uid, source="auto_inject", count=2, query="how is the cache invalidated"):
|
|
return RetrievalLog(
|
|
user_id=uid, source=source, query=query, threshold=0.55,
|
|
limit_n=3, result_count=count, created_at=now - timedelta(hours=2),
|
|
result_ids=[{"id": 501, "score": 0.8, "rank": 0}],
|
|
)
|
|
|
|
rows = {
|
|
"a": log(me.id), "b": log(me.id),
|
|
"empty": log(me.id, count=0),
|
|
"other_arm": log(me.id, source="write_path"),
|
|
"theirs": log(other.id),
|
|
}
|
|
s.add_all(rows.values())
|
|
await s.commit()
|
|
ids = {k: int(v.id) for k, v in rows.items()}
|
|
ids |= {"me": me.id, "other": other.id}
|
|
yield ids
|
|
|
|
from sqlalchemy import delete
|
|
from grimoire.models.note_usage import NoteUsageEvent
|
|
from grimoire.models.retrieval_judgment import RetrievalJudgment
|
|
async with async_session() as s:
|
|
logs = [ids[k] for k in ("a", "b", "empty", "other_arm", "theirs")]
|
|
await s.execute(delete(RetrievalJudgment).where(
|
|
RetrievalJudgment.retrieval_log_id.in_(logs)))
|
|
await s.execute(delete(RetrievalLog).where(RetrievalLog.id.in_(logs)))
|
|
await s.execute(delete(NoteUsageEvent).where(
|
|
NoteUsageEvent.user_id == ids["me"], NoteUsageEvent.note_id.in_([501, 502])))
|
|
await s.commit()
|
|
|
|
|
|
def _lines(*specs):
|
|
"""A stand-in re-run: (record_id, rank, within_budget)."""
|
|
return AsyncMock(return_value=[
|
|
{"record_id": rid, "rank": rank, "score": 0.9 - rank / 10,
|
|
"within_budget": within, "changed_since_call": False,
|
|
"kind": "note", "name": f"n{rid}", "passage": "p"}
|
|
for rid, rank, within in specs
|
|
])
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("_dispose_engine")
|
|
async def test_the_sample_offers_only_the_reviewers_unjudged_reviewable_calls(logged_calls):
|
|
ids = logged_calls
|
|
with patch.object(review, "_rerun", _lines((501, 1, True))):
|
|
out = await review.menus_to_review(ids["me"], n=20, days=1)
|
|
assert {m["log_id"] for m in out["menus"]} == {ids["a"], ids["b"]}
|
|
assert out["remaining"] == 2
|
|
|
|
await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
|
|
{"record_id": 501, "verdict": "on_point", "reason": "names the cache"},
|
|
])
|
|
out = await review.menus_to_review(ids["me"], n=20, days=1)
|
|
assert {m["log_id"] for m in out["menus"]} == {ids["b"]}
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("_dispose_engine")
|
|
async def test_another_users_call_cannot_be_judged(logged_calls):
|
|
ids = logged_calls
|
|
with patch.object(review, "_rerun", _lines((501, 1, True))), \
|
|
pytest.raises(ValueError, match="not found"):
|
|
await review.judge_menu(ids["me"], log_id=ids["theirs"], verdicts=[
|
|
{"record_id": 501, "verdict": "unrelated", "reason": "r"},
|
|
])
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("_dispose_engine")
|
|
async def test_a_record_the_re_run_does_not_hold_is_refused(logged_calls):
|
|
ids = logged_calls
|
|
with patch.object(review, "_rerun", _lines((501, 1, True))), \
|
|
pytest.raises(ValueError, match=r"\[999\]"):
|
|
await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
|
|
{"record_id": 999, "verdict": "unrelated", "reason": "r"},
|
|
])
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("_dispose_engine")
|
|
async def test_verdicts_land_replace_and_read_back_by_rank(logged_calls):
|
|
from grimoire.models.note_usage import NoteUsageEvent
|
|
from grimoire.services.retrieval_telemetry import retrieval_summary
|
|
|
|
ids = logged_calls
|
|
# The reviewer's agent opened 501 an hour inside the call; nothing opened 502.
|
|
async with async_session() as s:
|
|
s.add(NoteUsageEvent(
|
|
user_id=ids["me"], note_id=501, event="pulled", source="mcp_get_note",
|
|
created_at=datetime.now(timezone.utc) - timedelta(hours=1, minutes=30),
|
|
))
|
|
await s.commit()
|
|
|
|
rerun = _lines((501, 1, True), (502, 4, False))
|
|
with patch.object(review, "_rerun", rerun):
|
|
await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
|
|
{"record_id": 501, "verdict": "adjacent", "reason": "first look"},
|
|
{"record_id": 502, "verdict": "on_point", "reason": "the exact fix"},
|
|
])
|
|
again = await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
|
|
{"record_id": 501, "verdict": "on_point", "reason": "on reflection"},
|
|
])
|
|
assert again == {"log_id": ids["a"], "recorded": 1, "replaced": 1}
|
|
|
|
block = (await retrieval_summary(ids["me"], days=1))["judged"]
|
|
assert "judged_failed" not in block, "the readout did not execute"
|
|
ai = block["auto_inject"]
|
|
assert ai["judged_calls"] == 1 and ai["judged_lines"] == 2
|
|
assert ai["within_budget"] == {
|
|
"on_point": 1, "adjacent": 0, "unrelated": 0, "opened": 1}
|
|
assert ai["beyond_budget"] == {
|
|
"on_point": 1, "adjacent": 0, "unrelated": 0, "opened": 0}
|
|
assert [r["rank"] for r in ai["by_rank"]] == [1, 4]
|
|
# Related and left closed: the number an open rate could never show.
|
|
assert ai["on_point_unopened"] == 1
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("_dispose_engine")
|
|
async def test_a_fresh_install_reads_an_empty_judged_block_not_a_failure():
|
|
from grimoire.services.retrieval_telemetry import retrieval_summary
|
|
|
|
assert (await retrieval_summary(990041, days=30))["judged"] == {}
|