test(retrieval): two source guards read the pipeline where the rule arms now live (milestone 456 step 2, #4904)
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / integration (push) Successful in 49s
CI & Build / TypeScript typecheck (push) Successful in 52s
CI & Build / Python tests (push) Successful in 1m45s
CI & Build / Build & push image (push) Successful in 30s

CI run 8137 had two failures, both source-scanning guards whose property
moved to the pipeline. The property itself still held:

- test_every_surface_name_is_a_real_telemetry_source looked for
  source="<name>" literals. The rule arms now record under their spec field,
  so the spec is the join key, read from RULE_ARMS.
- test_every_hook_rule_search_says_which_project_it_is_for counted 4 direct
  searches in plugin_context. It now expects 0 there, so a copied arm fails
  the count. It also asserts that the pipeline has exactly one search and
  that its keyword literal carries project_id and never everywhere.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-05 08:07:26 -04:00
co-authored by Claude Opus 5.5
parent f6b824b214
commit b72c9a92e2
2 changed files with 37 additions and 2 deletions
+29 -1
View File
@@ -1698,7 +1698,10 @@ def test_every_hook_rule_search_says_which_project_it_is_for():
`everywhere` is not an acceptable answer in a hook, which speaks unasked.
"""
sources = {
"src/scribe/services/plugin_context.py": 4,
# Since milestone 456 the hook arms search through the pipeline's one
# ranked-search stage, checked below — so a direct search appearing
# here again is a copy of the arm coming back, and fails the count.
"src/scribe/services/plugin_context.py": 0,
"src/scribe/services/reply_preferences.py": 1,
}
for path, expected in sources.items():
@@ -1724,6 +1727,31 @@ def test_every_hook_rule_search_says_which_project_it_is_for():
f"{path}:{call.lineno} searches every project's rules from a hook"
)
# The pipeline: exactly one search, in `_ranked`, whose keyword set is
# built as a dict literal — so the keys are read from that literal.
pipeline = ast.parse(Path("src/scribe/services/retrieval_pipeline.py").read_text())
searches = [
n for n in ast.walk(pipeline)
if isinstance(n, ast.Call) and getattr(n.func, "attr", None) == "search"
]
assert len(searches) == 1, (
f"the pipeline has {len(searches)} rule searches, expected 1 — every "
f"arm is meant to reach the ranker through `_ranked`"
)
ranked = next(n for n in ast.walk(pipeline)
if isinstance(n, ast.AsyncFunctionDef) and n.name == "_ranked")
keys = {
k.value for d in ast.walk(ranked) if isinstance(d, ast.Dict)
for k in d.keys if isinstance(k, ast.Constant)
}
assert "project_id" in keys, (
"the pipeline's rule search no longer passes project_id, so every hook "
"arm gets global rules only and never its own project's"
)
assert "everywhere" not in keys and not any(
isinstance(n, ast.keyword) and n.arg == "everywhere" for n in ast.walk(ranked)
), "the pipeline searches every project's rules from a hook"
@pytest.mark.asyncio
@pytest.mark.parametrize("bound, scope", [(7, 7), (0, None)])