tests: hygiene pass on the agentic-review suites (owner R45 pin form)

Test-only; no runtime change (every ouroboros/*.py AST is identical modulo
docstrings before and after).

- test_native_tool_round_executor.py: the `last_read_view` writer-set pin
  takes owner R45's form - one scanner over assignment sites
  (Assign/AnnAssign/AugAssign attribute targets, setattr/__setattr__ calls,
  constant-key subscript stores), ONE live pin and ONE negative case (a
  mutated in-memory copy with a fourth reset injected into `_read_file`
  must fail the pin); the scanner's residual porosity is listed on its
  docstring. The two first-send admission tests fold into one parametrized
  test; the assistant-envelope inequality test goes (implied by the
  strict-equality wire-measure pin).
- test_deep_review_slot.py: the duplicate ("length" -> output_reserve) loop
  folds into the first finish-reason loop; the coverage-level traversal case
  keeps ONE shape (the refusal trio stays the executor suite's receipt-level
  pin).
- test_delivery_retrieves.py: the two `inspect.getsource` call-site pins
  become a behavioural assertion - the request each surface hands the
  substrate carries the context's deadline_at.
- test_reviewer_slot_config.py: the repository-wide retired-name text scan
  becomes module-attribute absence asserts on the modules that carried the
  retired API-pin apparatus.
- test_project_routing_v664.py: the mtime-tie test is dropped (implied by
  the total-order parametrized test).
This commit is contained in:
Ouroboros 2026-09-03 20:12:57 +00:00
parent 902047baff
commit bbbf4dc952
5 changed files with 120 additions and 206 deletions

View file

@ -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)]},

View file

@ -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():

View file

@ -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(<expr>, "last_read_view", …)` / `__setattr__` call, `subscript`
for a `<expr>["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(<expr>, "last_read_view", …)` /
`__setattr__` call, `subscript` for a `<expr>["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 "<module>"
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", "<lambda>")
visit(child, rel, name if scope == "<module>" else f"{scope}.{name}")
else:
visit(child, rel, scope)
for rel, text in sources.items():
try:
visit(ast.parse(text), rel, "<module>")
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, "<module>", 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)

View file

@ -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

View file

@ -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):