diff --git a/tests/test_deep_review_slot.py b/tests/test_deep_review_slot.py index 3e53c8c4a..131299927 100644 --- a/tests/test_deep_review_slot.py +++ b/tests/test_deep_review_slot.py @@ -802,7 +802,11 @@ def test_packed_incomplete_follows_the_provider_finish_reason(review_repo, revie llm.chat.return_value = ({"content": "Cut repo"}, usage_in) with mock.patch.object(deep_self_review, "build_review_pack", return_value=(pack, stats)): text, usage = run_deep_self_review(review_repo, review_drive, llm, lambda _m: None, slot=_row()) - assert f"incomplete={expected}" in text.split("\n")[0], (finish, text.split("\n")[0]) + comment = text.split("\n")[0] + assert f"incomplete={expected}" in comment, (finish, comment) + assert ("INCOMPLETE (output_reserve" in text) == (expected == "output_reserve") + assert usage["deep_review_memory"] == stats["memory"] and "memory=1/7" in comment + assert "execution_status" not in usage # a cut report is a product, disclosed — not a failure # The direct-Anthropic lane (the shipped `anthropic::` deep default) sets # NO usage finish reason; its cut marker is the message's `stop_reason`. for message, expected in ( @@ -819,16 +823,6 @@ def test_packed_incomplete_follows_the_provider_finish_reason(review_repo, revie text, usage = run_deep_self_review(review_repo, review_drive, llm, lambda _m: None, slot=_row()) assert f"incomplete={expected}" in text.split("\n")[0], (message, text.split("\n")[0]) assert ("INCOMPLETE (output_reserve" in text) == (expected == "output_reserve") - for finish, expected in (("length", "output_reserve"),): - usage_in = {"cost": 0.0, "response_finish_reason": finish} - llm.chat.return_value = ({"content": "Cut repo"}, usage_in) - with mock.patch.object(deep_self_review, "build_review_pack", return_value=(pack, stats)): - text, usage = run_deep_self_review(review_repo, review_drive, llm, lambda _m: None, slot=_row()) - comment = text.split("\n")[0] - assert f"incomplete={expected}" in comment, (finish, comment) - assert ("INCOMPLETE (output_reserve" in text) == (expected == "output_reserve") - assert usage["deep_review_memory"] == stats["memory"] and "memory=1/7" in comment - assert "execution_status" not in usage # a cut report is a product, disclosed — not a failure def test_memory_dispositions_are_disclosed_per_whitelisted_entry(review_repo, tmp_path, monkeypatch): @@ -938,7 +932,9 @@ def test_a_registry_refused_read_never_inherits_the_previous_reads_extent(review its typed delta; after a real PARTIAL read of BIBLE.md the traversal shapes never lift it to `read`.""" monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test") - shapes = ("a/../BIBLE.md", "BIBLE.md/../BIBLE.md", "../../../etc/passwd") + # ONE traversal shape at the coverage level; the refusal trio itself is the + # executor suite's receipt-level pin (test_read_file_receipts_carry_the_delivered_extent). + shapes = ("a/../BIBLE.md",) llm = _ScriptedLLM([ {"tool_calls": [_tool_call("read_file", {"path": "docs/ARCHITECTURE.md"}, "c1")] + [_tool_call("read_file", {"path": p}, f"c{i}") for i, p in enumerate(shapes, 2)]}, diff --git a/tests/test_delivery_retrieves.py b/tests/test_delivery_retrieves.py index c0a5892fe..615b3ee4b 100644 --- a/tests/test_delivery_retrieves.py +++ b/tests/test_delivery_retrieves.py @@ -21,24 +21,40 @@ def test_predicate_accepts_route_kind_or_wire_string(): assert delivery_retrieves(None, "") is False -def test_owner_deadline_reaches_the_triad_and_scope_requests(): +def test_owner_deadline_reaches_the_triad_and_scope_requests(monkeypatch, tmp_path): """R23: the owner deadline is a bound of every retrieving episode, so the commit triad and scope requests carry the task's deadline_at exactly as - the advisory does; a context without one yields ''.""" + the advisory does; a context without one yields ''. Behavioural: the + request each surface hands the substrate carries the context's deadline — + the stub captures it and refuses before any send.""" + import asyncio from types import SimpleNamespace + import ouroboros.review_substrate as substrate + from ouroboros.tools import review as review_mod, scope_review as scope_mod from ouroboros.tools.review import _owner_deadline_at ctx = SimpleNamespace(task_metadata={"deadline_at": "2030-01-01T00:00:00Z"}) assert _owner_deadline_at(ctx) == "2030-01-01T00:00:00Z" assert _owner_deadline_at(SimpleNamespace(task_metadata={})) == "" assert _owner_deadline_at(SimpleNamespace()) == "" and _owner_deadline_at(None) == "" - import inspect - from ouroboros.tools import review as review_mod, scope_review as scope_mod + seen = [] - assert "deadline_at=_owner_deadline_at(ctx)" in inspect.getsource(review_mod._query_model) - assert "deadline_at=_owner_deadline_at(ctx)" in inspect.getsource(scope_mod._call_scope_llm) + def _capture(request, **_kwargs): + seen.append((request.surface, request.deadline_at)) + raise RuntimeError("captured before any send") + + monkeypatch.setattr(substrate, "run_review_request", _capture) + monkeypatch.setattr(scope_mod, "LLMClient", lambda: object()) + monkeypatch.setattr(scope_mod, "_scope_window", lambda model, **_k: SimpleNamespace(sizing_window=lambda floor: 200_000)) + ctx = SimpleNamespace(task_metadata={"deadline_at": "2030-01-01T00:00:00Z"}, task_id="t-deadline", + drive_root=str(tmp_path), pending_events=[], event_queue=None) + _, payload, _ = asyncio.run(review_mod._query_model(object(), "openai/fake-reviewer", [], asyncio.Semaphore(1), ctx=ctx)) + assert "captured before any send" in payload["error"] + _, _, error = scope_mod._call_scope_llm("scope prompt", scope_model="openai/fake-scope", ctx=ctx) + assert "captured before any send" in error + assert seen == [("multi_model_review", "2030-01-01T00:00:00Z"), ("scope_review", "2030-01-01T00:00:00Z")] def test_slot_properties_and_plan_review_facade_share_the_predicate(): diff --git a/tests/test_native_tool_round_executor.py b/tests/test_native_tool_round_executor.py index 03fdfb6d6..2ae15aeb8 100644 --- a/tests/test_native_tool_round_executor.py +++ b/tests/test_native_tool_round_executor.py @@ -298,44 +298,34 @@ def test_escaping_inflation_cannot_jump_the_landing_notice(subject_repo, monkeyp assert "RESULT TRUNCATED" in tool_msg["content"] and "\\n" not in tool_msg["content"] # real text, escaped only on the wire -def test_first_send_is_measured_on_the_wire(subject_repo, monkeypatch): - """The initial system+task messages are charged as the serialized objects - the send carries (escape-heavy text inflates), so a bound the raw text - would pass but the wire would not is refused before any send.""" +@pytest.mark.parametrize("wire_inflated", [False, True], ids=["room-below-landing", "escape-heavy-first-send"]) +def test_bound_below_the_first_send_is_a_typed_refusal_before_any_send(subject_repo, monkeypatch, wire_inflated): + """A bound that leaves no room to read anything must not make the landing + notice the first thing the reviewer hears (an obedient `[]` would then be a + strict clean verdict with zero reads): the refusal is typed and precedes + any send. The first send is charged as the serialized objects the wire + carries, so escape-heavy task text inflates it: a bound the raw text would + pass but the wire would not is refused the same way.""" import ouroboros.review_native_episode as native_episode - task = ('line "quoted" \\ back\n' * 2_000) # ~44K raw, far more on the wire llm = _ScriptedLLM([{"content": _VERDICT}]) - executor = NativeToolRoundReviewExecutor(_assignment(subject_repo, llm, session_task=task), llm=llm) - raw = len(executor.episode_prompt) + len(native_episode._NATIVE_REVIEW_INSTRUCTIONS) - wire = len(json.dumps([{"role": "system", "content": native_episode._NATIVE_REVIEW_INSTRUCTIONS}, - {"role": "user", "content": executor.episode_prompt}], ensure_ascii=False)) - assert wire > raw + 2_000 - monkeypatch.setattr(native_episode, "review_native_transcript_bound", lambda *a, **k: raw + 12_000) + if wire_inflated: + task = 'line "quoted" \\ back\n' * 2_000 # ~44K raw, far more on the wire + executor = NativeToolRoundReviewExecutor(_assignment(subject_repo, llm, session_task=task), llm=llm) + raw = len(executor.episode_prompt) + len(native_episode._NATIVE_REVIEW_INSTRUCTIONS) + wire = len(json.dumps([{"role": "system", "content": native_episode._NATIVE_REVIEW_INSTRUCTIONS}, + {"role": "user", "content": executor.episode_prompt}], ensure_ascii=False)) + assert wire > raw + 2_000 + bound = raw + 12_000 + else: + executor = NativeToolRoundReviewExecutor(_assignment(subject_repo, llm), llm=llm) + bound = _first_send_chars(subject_repo) + 200 # landing_at <= first send + monkeypatch.setattr(native_episode, "review_native_transcript_bound", lambda *a, **k: bound) with pytest.raises(ReviewRouteUnavailable) as exc: executor.execute() assert exc.value.code == "native_bound_below_first_send" and not llm.calls -def test_bound_below_the_first_send_is_a_typed_refusal_before_any_send(subject_repo, monkeypatch): - """A bound that leaves no room to read anything must not make the landing - notice the first thing the reviewer hears (an obedient `[]` would then be a - strict clean verdict with zero reads).""" - import ouroboros.review_native_episode as native_episode - - llm = _ScriptedLLM([{"content": _VERDICT}]) - executor = NativeToolRoundReviewExecutor(_assignment(subject_repo, llm), llm=llm) - first_send = _first_send_chars(subject_repo) - monkeypatch.setattr(native_episode, "review_native_transcript_bound", - lambda *a, **k: first_send + 200) # landing_at <= first send - llm2 = _ScriptedLLM([{"content": _VERDICT}]) - executor = NativeToolRoundReviewExecutor(_assignment(subject_repo, llm2), llm=llm2) - with pytest.raises(ReviewRouteUnavailable) as exc: - executor.execute() - assert exc.value.code == "native_bound_below_first_send" - assert not llm2.calls - - def test_pre_send_refusal_never_projects_a_native_execution(subject_repo, tmp_path, monkeypatch): """The receipt keys (resolved model/provider) are filled only from a real send: a refusal before the first send hands the error actor its facts via @@ -937,24 +927,6 @@ def test_round_without_progress_is_a_typed_malformed_end(subject_repo, container assert len(llm.calls) == 1 and llm.script -def test_assistant_envelope_is_counted_into_the_transcript(subject_repo): - """The whole assistant message (content + tool-call objects) rides every - later send, so the counter grows by at least its serialized size.""" - llm = _ScriptedLLM([ - {"content": "note " * 100, "tool_calls": [_tool_call("read_file", {"path": "greeting.txt"}, "c1")]}, - {"content": _VERDICT}, - ]) - executor = NativeToolRoundReviewExecutor(_assignment(subject_repo, llm), llm=llm) - usage = executor.execute().usage - first_send = _first_send_chars(subject_repo) - envelope = json.dumps({**llm_first(llm), "role": "assistant"}, ensure_ascii=False) - assert usage["native_transcript_chars"] >= first_send + len(envelope) + len("hello native reviewer\n") - - -def llm_first(llm): - return {"content": "note " * 100, "tool_calls": [_tool_call("read_file", {"path": "greeting.txt"}, "c1")]} - - def test_report_shape_delivers_the_collected_draft_marked_incomplete(subject_repo, monkeypatch, tmp_path): """A report is a product, not a verdict: when the bound lands before the final answer, the reviewer's last draft is delivered with a typed @@ -1374,104 +1346,73 @@ def test_read_extent_counts_only_complete_delivered_lines_from_the_stamp(subject def _last_read_view_sites(sources): """Every WRITE site of `last_read_view` in `{relative file: source}` as - `(file, scope, lineno, col, kind)`: `assign` for an Attribute store target - (Assign/AnnAssign/AugAssign/for/with targets, recursing Tuple/List/Starred; - module, class, nested-function and lambda scopes alike), `setattr` for a - `setattr(, "last_read_view", …)` / `__setattr__` call, `subscript` - for a `["last_read_view"] = …` (`__dict__`/`vars()`) store.""" + `(file, enclosing def, lineno, kind)`: `assign` for an Attribute target of + an Assign/AnnAssign/AugAssign statement (every target of a chained + assignment), `setattr` for a `setattr(, "last_read_view", …)` / + `__setattr__` call, `subscript` for a `["last_read_view"] = …` store + (`__dict__` / `vars()`). Residual porosity, disclosed (owner R45) — writes + this scanner does NOT see: an unpacking target (`ctx.last_read_view, x = + …`), `for` / `with` / comprehension targets, a plain-Name target (a + class-body default `last_read_view = None` would be a declaration, not a + write, and is skipped by design), an attribute name assembled at run time + (`setattr(ctx, "last_" + …)`, an aliased `__setattr__`) and dict mutators + (`ctx.__dict__.update(…)`, a `vars(ctx)` store under a non-constant key).""" import ast + def enclosing(tree, lineno): + defs = [n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) + and n.lineno <= lineno <= n.end_lineno] + return min(defs, key=lambda n: n.end_lineno - n.lineno).name if defs else "" + sites = set() - - def flatten(target): - if isinstance(target, (ast.Tuple, ast.List)): - for elt in target.elts: - yield from flatten(elt) - elif isinstance(target, ast.Starred): - yield from flatten(target.value) - else: - yield target - - def targets(node): - if isinstance(node, ast.Assign): - return node.targets - if isinstance(node, (ast.AnnAssign, ast.AugAssign, ast.For, ast.AsyncFor)): - return [node.target] - if isinstance(node, (ast.With, ast.AsyncWith)): - return [item.optional_vars for item in node.items if item.optional_vars is not None] - return [] - - def visit(node, rel, scope): - for child in ast.iter_child_nodes(node): - for target in (t for raw in targets(child) for t in flatten(raw)): - if isinstance(target, ast.Attribute) and target.attr == "last_read_view": - sites.add((rel, scope, child.lineno, child.col_offset, "assign")) - elif (isinstance(target, ast.Subscript) and isinstance(target.slice, ast.Constant) - and target.slice.value == "last_read_view"): - sites.add((rel, scope, child.lineno, child.col_offset, "subscript")) - if isinstance(child, ast.Call): - fn = child.func - name = fn.id if isinstance(fn, ast.Name) else (fn.attr if isinstance(fn, ast.Attribute) else "") - if name in ("setattr", "__setattr__") and any( - isinstance(a, ast.Constant) and a.value == "last_read_view" for a in child.args[:2]): - sites.add((rel, scope, child.lineno, child.col_offset, "setattr")) - if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)): - name = getattr(child, "name", "") - visit(child, rel, name if scope == "" else f"{scope}.{name}") - else: - visit(child, rel, scope) - for rel, text in sources.items(): try: - visit(ast.parse(text), rel, "") + tree = ast.parse(text) except SyntaxError: continue + for node in ast.walk(tree): + targets = (node.targets if isinstance(node, ast.Assign) + else [node.target] if isinstance(node, (ast.AnnAssign, ast.AugAssign)) else []) + for target in targets: + if isinstance(target, ast.Attribute) and target.attr == "last_read_view": + sites.add((rel, enclosing(tree, node.lineno), node.lineno, "assign")) + elif (isinstance(target, ast.Subscript) and isinstance(target.slice, ast.Constant) + and target.slice.value == "last_read_view"): + sites.add((rel, enclosing(tree, node.lineno), node.lineno, "subscript")) + if isinstance(node, ast.Call): + fn = node.func + name = fn.id if isinstance(fn, ast.Name) else getattr(fn, "attr", "") + if name in ("setattr", "__setattr__") and any( + isinstance(a, ast.Constant) and a.value == "last_read_view" for a in node.args[:2]): + sites.add((rel, enclosing(tree, node.lineno), node.lineno, "setattr")) return sites _CORE, _EPISODE = "ouroboros/tools/core.py", "ouroboros/review_native_episode.py" -_LAST_READ_VIEW_WRITERS = { # exact (file, scope, kind); line/col come from the live tree, the site COUNT is pinned at three +_LAST_READ_VIEW_WRITERS = { # exact (file, enclosing def, kind); the site COUNT is pinned at three (_CORE, "_stamp_read_view", "assign"), (_CORE, "_read_file", "assign"), - (_EPISODE, "NativeToolRoundReviewExecutor._execute_inspection_call", "assign"), + (_EPISODE, "_execute_inspection_call", "assign"), } +_RESET = " ctx.last_read_view = None\n" # the reader's entry reset inside _read_file -def _inject_after(text, anchor, lines): - rows = text.splitlines(keepends=True) - idx = next(i for i, row in enumerate(rows) if row.rstrip("\n") == anchor) - return "".join(rows[:idx + 1] + [line + "\n" for line in lines] + rows[idx + 1:]), idx + 2 # 1-based first injected line +def _assert_three_writers(sites): + assert {(f, d, k) for f, d, _l, k in sites} == _LAST_READ_VIEW_WRITERS and len(sites) == 3, sorted(sites) -_RESET = " ctx.last_read_view = None" # the entry reset inside _read_file: every in-function probe lands right after it -_FOURTH_WRITERS = { # label -> mutate(core.py source) -> (mutated source, the ONE extra site the scanner must report) - "fourth-in-_read_file": lambda t: (lambda m: (m[0], (_CORE, "_read_file", m[1], 4, "assign")))( - _inject_after(t, _RESET, [" ctx.last_read_view = None # injected"])), - "module-scope": lambda t: (t + "\nToolContext.last_read_view = None\n", (_CORE, "", t.count("\n") + 2, 0, "assign")), - "class-scope": lambda t: (t + "\nclass _Probe:\n ToolContext.last_read_view = None\n", (_CORE, "_Probe", t.count("\n") + 3, 4, "assign")), - "tuple-target-in-nested-def": lambda t: (lambda m: (m[0], (_CORE, "_read_file._inner", m[1] + 1, 8, "assign")))( - _inject_after(t, _RESET, [" def _inner():", " ctx.last_read_view, _unused = None, 0"])), - "setattr": lambda t: (lambda m: (m[0], (_CORE, "_read_file", m[1], 4, "setattr")))( - _inject_after(t, _RESET, [' setattr(ctx, "last_read_view", None)'])), - "__dict__-store": lambda t: (lambda m: (m[0], (_CORE, "_read_file", m[1], 4, "subscript")))( - _inject_after(t, _RESET, [' ctx.__dict__["last_read_view"] = None'])), -} - - -@pytest.mark.parametrize("label", ["live", *_FOURTH_WRITERS], ids=str) -def test_last_read_view_has_exactly_three_writers(label): +def test_last_read_view_has_exactly_three_writers(): """Writer-set invariant behind the structural stamp binding: `last_read_view` - is written at exactly three SITES — (file, scope, line, col, kind), asserted - as the exact (file, scope, kind) set with a site count of three — the reader's - entry reset and its stamp (`tools/core.py::_read_file` / `_stamp_read_view`) - and the episode's clear-before-dispatch - (`review_native_episode.py::NativeToolRoundReviewExecutor._execute_inspection_call`); no `setattr` / - `__setattr__` call or `__dict__` store names it anywhere in the runtime. - Sites, not (file, function) members: a fourth assignment inside one of the - three functions, at module or class scope, in a nested def, or through a - tuple target is a distinct site and fails. The negative cases run the SAME - scanner on mutated in-memory copies of the live source and prove each such - fourth writer is detected — the pin's own claim is tested, not assumed.""" + is written at exactly three SITES — the reader's entry reset and its stamp + (`tools/core.py::_read_file` / `_stamp_read_view`) and the episode's + clear-before-dispatch (`review_native_episode.py::_execute_inspection_call`) + — and no `setattr` / `__setattr__` call or `__dict__` store names it + anywhere in the runtime. Sites, not (file, def) members: a fourth + assignment inside one of the three functions is a distinct site. The one + negative case runs the SAME scanner on a mutated in-memory copy of the live + source with one more reset injected into `_read_file` and proves the pin + fails on it — the pin's own claim is tested, not assumed. What the scanner + cannot see is listed on `_last_read_view_sites`.""" import pathlib import ouroboros @@ -1486,12 +1427,15 @@ def test_last_read_view_has_exactly_three_writers(label): sources[rel.as_posix()] = py.read_text(encoding="utf-8") except UnicodeDecodeError: continue - if label == "live": - sites = _last_read_view_sites(sources) - assert {(f, sc, k) for f, sc, _l, _c, k in sites} == _LAST_READ_VIEW_WRITERS and len(sites) == 3, sorted(sites) - return - live = {rel: sources[rel] for rel in (_CORE, _EPISODE)} - base = _last_read_view_sites(live) # the three sites all live in these two files - assert {(f, sc, k) for f, sc, _l, _c, k in base} == _LAST_READ_VIEW_WRITERS and len(base) == 3, sorted(base) - live[_CORE], extra = _FOURTH_WRITERS[label](live[_CORE]) - assert _last_read_view_sites(live) == base | {extra}, label + _assert_three_writers(_last_read_view_sites(sources)) + live = {rel: sources[rel] for rel in (_CORE, _EPISODE)} # the three sites all live in these two files + base = _last_read_view_sites(live) + _assert_three_writers(base) + core = live[_CORE] + assert core.count(_RESET) == 1 + injected_at = core[:core.index(_RESET)].count("\n") + 2 # 1-based line of the injected reset + live[_CORE] = core.replace(_RESET, _RESET + _RESET.rstrip("\n") + " # injected\n") + fourth = _last_read_view_sites(live) + assert fourth == base | {(_CORE, "_read_file", injected_at, "assign")} + with pytest.raises(AssertionError): + _assert_three_writers(fourth) diff --git a/tests/test_project_routing_v664.py b/tests/test_project_routing_v664.py index b246982f6..272eec70e 100644 --- a/tests/test_project_routing_v664.py +++ b/tests/test_project_routing_v664.py @@ -504,33 +504,6 @@ def test_project_room_decision_turn_carries_last_task_result_ground_truth(tmp_pa assert "result" not in last and "RAW TEXT" not in str(last) -def test_project_last_task_result_breaks_an_mtime_tie_by_the_durable_ts(tmp_path, monkeypatch): - """Results finalized within one clock tick share a file mtime, so mtime order - alone is arbitrary among them: inside the tie group the durable `ts` decides. - The adverse listing order (the OLDER result first) is forced through the - directory listing, because a real filesystem's order is not controllable.""" - import os - - import server - from ouroboros import task_results as task_results_module - from ouroboros.projects_registry import create_project - from ouroboros.task_results import task_results_dir, write_task_result - - create_project(tmp_path, "racer", name="Racer") - write_task_result(tmp_path, "old1", "completed", project_id="racer", objective="a", ts="2026-08-10T00:00:01Z") - write_task_result(tmp_path, "new1", "completed", project_id="racer", objective="b", ts="2026-08-10T00:00:02Z") - write_task_result(tmp_path, "other1", "completed", project_id="boat", objective="c", ts="2026-08-10T00:00:03Z") - real_dir = task_results_dir(tmp_path, create=False) - tick = 1_700_000_000 * 10**9 - for path in real_dir.glob("*.json"): - os.utime(path, ns=(tick, tick)) # one shared mtime: the tie the scan must resolve by `ts` - adverse = [real_dir / "old1.json", real_dir / "new1.json", real_dir / "other1.json"] - assert all(path.exists() for path in adverse) - monkeypatch.setattr(task_results_module, "task_results_dir", - lambda root, create=True: types.SimpleNamespace(glob=lambda pattern: iter(adverse))) - assert server._latest_project_task_result(_ctx(tmp_path), "racer")["task_id"] == "new1" - - @pytest.mark.parametrize("shape", ["group_crosses_the_window", "group_in_the_self_heal_tail", "equal_ts", "empty_ts"]) def test_project_last_task_result_tie_order_is_total(tmp_path, monkeypatch, shape): """The tie order is TOTAL: an equal-mtime group that crosses the 64-entry diff --git a/tests/test_reviewer_slot_config.py b/tests/test_reviewer_slot_config.py index fa0ec4007..87a502797 100644 --- a/tests/test_reviewer_slot_config.py +++ b/tests/test_reviewer_slot_config.py @@ -854,40 +854,25 @@ def test_all_delegated_triad_writes_no_fallback_record_and_reaches_acceptance(mo reviewer_slot_save_check(json.dumps({**payload, "triad": []})) -# The acceptance API-pin apparatus retired with owner R2/R12 (2026-09-01): a -# name that survives anywhere — code, docstring, comment, test prose — is a -# hook for the fallback to grow back. Scanned as TEXT over every file under the -# roots the phase touched; this file is the assertion and is excluded. +# The acceptance API-pin apparatus retired with owner R2/R12 (2026-09-01): its +# helpers lived in `reviewer_slot_config` and their one importer was +# `claudexor_daemon` (both cleared at a3599ecd; the fallback record +# `reviewer_slot_api_fallback.json` had no writer but +# `_record_api_fallback_substitution`). A surviving module attribute is the +# hook a fallback would grow back on. _RETIRED_API_PIN_NAMES = ( "_fallback_warning_text", "_record_api_fallback_substitution", "api_fallback_disclosure", "reviewer_slot_api_fallback_warning", - "reviewer_slot_api_fallback.json", ) -_RETIRED_NAME_SCAN_ROOTS = ("ouroboros", "web/modules", "web/tests", "tests") -def test_the_retired_acceptance_api_pin_names_are_gone_repository_wide(): - """Pure file reads, no globals, no imports of the scanned modules — safe - under xdist. Binary fixtures (undecodable) carry no prose and are skipped.""" - import pathlib +def test_the_retired_acceptance_api_pin_apparatus_is_gone(): + from ouroboros import claudexor_daemon, reviewer_slot_config - here = pathlib.Path(__file__).resolve() - repo = here.parents[1] - hits = [] - for root in _RETIRED_NAME_SCAN_ROOTS: - for path in sorted((repo / root).rglob("*")): - if not path.is_file() or path.resolve() == here: - continue - if "__pycache__" in path.parts or "node_modules" in path.parts: - continue - try: - text = path.read_text(encoding="utf-8") - except (UnicodeDecodeError, OSError): - continue - hits.extend(f"{path.relative_to(repo)}: {name}" for name in _RETIRED_API_PIN_NAMES if name in text) - assert hits == [], hits + assert [(module.__name__, name) for module in (reviewer_slot_config, claudexor_daemon) + for name in _RETIRED_API_PIN_NAMES if hasattr(module, name)] == [] def test_mixed_triad_reaches_acceptance_in_row_order(monkeypatch):