ouroboros/tests/test_native_tool_timeout.py

"""A reviewer's inspection tool call runs under the loop's own timeout policy.

`review_native_episode` held the runtime's only unbounded `registry.execute`:
one wedged `query_code` kept a reviewer slot silent for 7 h 16 min while the
episode checked its clock only before a model send. The call is now bounded by
the SAME per-tool resolution every other tool call uses (`_get_tool_timeout`),
narrowed by whatever calendar/execution bound the reviewer inherited, and a
call that outlives its bound is ABANDONED: the reviewer is told in the standard
host text, the receipt is an error carrying `native_tool_abandoned`, and the
value the worker finally produces sources no receipt and no read coverage.
"""

import threading
from datetime import datetime

import pytest

import ouroboros.loop_tool_execution as loop_tool_execution
import ouroboros.review_native_episode as native_episode
from ouroboros.model_wait import execution_deadline_scope, monotonic_now
from ouroboros.review_execution import ReviewAssignment, ReviewRouteKind
from ouroboros.review_native_episode import NativeToolRoundReviewExecutor
from ouroboros.review_substrate import ReviewRequest, ReviewSlot
from tests.test_native_tool_round_executor import _VERDICT, _ScriptedLLM, _tool_call

_REAL_REGISTRY = native_episode.inspection_registry

# What the real reader stamps on the shared context after a read renders.
_STAMP = {
    "target": "/x/greeting.txt", "opened_path": "greeting.txt", "opened_root": "active_workspace",
    "first_line": 1, "end_line": 1, "total_lines": 1, "body_start": 0, "line_ends": [10],
}
_LATE_BODY = "late body from an abandoned read"


@pytest.fixture()
def repo(tmp_path):
    root = tmp_path / "subject"
    root.mkdir()
    (root / "greeting.txt").write_text("hello native reviewer\n", encoding="utf-8")
    (tmp_path / "custody").mkdir()
    return root


def _assignment(repo_dir):
    request = ReviewRequest(
        surface="multi_model_review", goal="review", task_id="t-native",
        session_root=str(repo_dir), session_task="Review the staged change; cite files.",
        policy={"output_contract": "JSON array of findings"}, no_proxy=True,
    )
    slot = ReviewSlot(slot_id="t1", model="openai/fake-reviewer", effort="low",
                      route=ReviewRouteKind.API_CHAT, subagent_id="api-critic")
    # A real custody root: the episode's own source store must succeed, or its
    # persistence gap would be the one this suite is reading.
    return ReviewAssignment(request=request, slot=slot, call_id="op-1",
                            custody_root=repo_dir.parent / "custody")


class _SlowRegistry:
    """The real inspection registry, with `read_file` held until released."""

    def __init__(self, inner, release, *, hold_sec):
        self._inner = inner
        self._release = release
        self._hold_sec = hold_sec
        self.returned = threading.Event()

    def __getattr__(self, name):
        return getattr(self._inner, name)

    def execute(self, name, args):
        if name != "read_file":
            return self._inner.execute(name, args)
        self._release.wait(self._hold_sec)
        # A late worker stamps the shared reader view exactly as the real
        # reader would: the episode must not credit that stamp to anything.
        self._inner._ctx.last_read_view = dict(_STAMP)
        self.returned.set()
        return _LATE_BODY


def _install_registry(monkeypatch, *, hold_sec):
    """Give the next episode a registry whose `read_file` hangs until released."""
    release, holder = threading.Event(), {}

    def factory(root, drive_root, task_id=""):
        registry, ctx, schemas = _REAL_REGISTRY(root, drive_root, task_id)
        holder["registry"], holder["ctx"] = _SlowRegistry(registry, release, hold_sec=hold_sec), ctx
        return holder["registry"], ctx, schemas

    monkeypatch.setattr(native_episode, "inspection_registry", factory)
    return release, holder


class _ReleasingLLM(_ScriptedLLM):
    """Frees the held tool the moment the model is asked a SECOND time, which can
    only happen after the first call was abandoned. An event, not a wall-clock
    guess: a timer that fired 0.3 s after the abandonment put the second call
    exactly on its own 0.3 s bound, and a slow CI runner lost that race."""

    def __init__(self, script, release):
        super().__init__(script)
        self._release = release

    def chat(self, **kwargs):
        if self.calls:
            self._release.set()
        return super().chat(**kwargs)


def _reading_script():
    return _ScriptedLLM([
        {"tool_calls": [_tool_call("read_file", {"path": "greeting.txt"})]},
        {"content": _VERDICT},
    ])


def _tool_messages(llm, call_index=1):
    return [m for m in llm.calls[call_index]["messages"] if m.get("role") == "tool"]


def test_wedged_reviewer_tool_is_abandoned_and_the_episode_continues(repo, monkeypatch):
    """The blocked case: a tool that outlives its bound never wedges the episode."""
    release, holder = _install_registry(monkeypatch, hold_sec=5.0)
    monkeypatch.setattr(loop_tool_execution, "_get_tool_timeout", lambda *_a, **_k: 0.3)
    llm = _reading_script()
    result = NativeToolRoundReviewExecutor(_assignment(repo), llm=llm).execute()

    assert result.raw_text == _VERDICT  # the episode CONTINUED past the wedged call
    usage = result.usage
    assert usage["native_rounds"] == 2
    receipt = usage["native_tool_receipts"][0]
    assert receipt["tool"] == "read_file"
    assert receipt["outcome"] == "error"
    assert receipt["source_gap"] == "native_tool_abandoned"
    assert "start_line" not in receipt and "opened_path" not in receipt
    assert usage["native_source_gap"] == "native_tool_abandoned"
    # The reviewer got the standard host text, not silence and not a verdict.
    assert _tool_messages(llm)[0]["content"].startswith("⚠️ TOOL_TIMEOUT (read_file)")

    release.set()
    assert holder["registry"].returned.wait(5), "the abandoned worker never settled"
    # The late value sources nothing: the receipts the episode published are final.
    assert usage["native_tool_receipts"][0]["outcome"] == "error"
    assert "text_sha256" not in usage["native_tool_receipts"][0]
    assert _LATE_BODY not in str(llm.calls[1]["messages"])


def test_fast_reviewer_tool_is_untouched_and_its_receipt_carries_duration(repo, monkeypatch):
    """The working case: the same bound leaves a returning read fully credited."""
    release, _holder = _install_registry(monkeypatch, hold_sec=5.0)
    release.set()  # the read returns at once
    monkeypatch.setattr(loop_tool_execution, "_get_tool_timeout", lambda *_a, **_k: 30)
    llm = _reading_script()
    result = NativeToolRoundReviewExecutor(_assignment(repo), llm=llm).execute()

    receipt = result.usage["native_tool_receipts"][0]
    assert receipt["outcome"] == "executed"
    assert "source_gap" not in receipt
    assert result.usage["native_source_gap"] == ""
    # One clock domain: an ISO wall stamp and the seconds elapsed on that clock.
    assert datetime.fromisoformat(receipt["started_at"]).tzinfo is not None
    assert isinstance(receipt["duration_sec"], float) and receipt["duration_sec"] >= 0.0
    assert _LATE_BODY in _tool_messages(llm)[0]["content"]


def test_tool_bound_is_the_loop_policy_for_this_registry_and_call(repo, monkeypatch):
    """The number comes from the main loop's own per-tool policy, resolved for
    THIS registry and THIS call's arguments — the episode mints none of its own."""
    release, holder = _install_registry(monkeypatch, hold_sec=5.0)
    release.set()
    seen = []
    real = loop_tool_execution._get_tool_timeout

    def spy(tools, tool_name, tool_args=None):
        seen.append((tools, tool_name, tool_args))
        return real(tools, tool_name, tool_args)

    monkeypatch.setattr(loop_tool_execution, "_get_tool_timeout", spy)
    NativeToolRoundReviewExecutor(_assignment(repo), llm=_reading_script()).execute()

    assert [(name, args) for _tools, name, args in seen] == [("read_file", {"path": "greeting.txt"})]
    assert seen[0][0] is holder["registry"]


def test_inherited_dispatch_deadline_narrows_the_tool_wait(repo, monkeypatch):
    """An inherited execution bound narrows the wait below the tool's own
    ceiling; without it the same call on the same tool finishes untouched."""
    release, _holder = _install_registry(monkeypatch, hold_sec=3.0)
    llm = _reading_script()
    started = monotonic_now()
    with execution_deadline_scope(monotonic_now() + 1.0):
        result = NativeToolRoundReviewExecutor(_assignment(repo), llm=llm).execute()
    elapsed = monotonic_now() - started

    assert elapsed < 3.0  # the 600s tool ceiling did not decide this wait
    receipt = result.usage["native_tool_receipts"][0]
    assert receipt["outcome"] == "error" and receipt["source_gap"] == "native_tool_abandoned"
    bound = float(_tool_messages(llm)[0]["content"].split("exceeded ")[1].split("s limit")[0])
    assert 0 < bound <= 1.0 + 1e-9 + 1e-9
    release.set()

    release_again, _holder_again = _install_registry(monkeypatch, hold_sec=3.0)
    release_again.set()
    again = NativeToolRoundReviewExecutor(_assignment(repo), llm=_reading_script()).execute()
    assert again.usage["native_tool_receipts"][0]["outcome"] == "executed"


def test_abandoned_call_stops_the_episode_crediting_read_extents(repo, monkeypatch):
    """A worker loose on the shared reader view makes attribution unprovable:
    after an abandonment no further read extent is credited, so a late stamp
    can never become this episode's coverage."""
    release, _holder = _install_registry(monkeypatch, hold_sec=5.0)
    monkeypatch.setattr(loop_tool_execution, "_get_tool_timeout", lambda *_a, **_k: 0.3)
    llm = _ReleasingLLM([
        {"tool_calls": [_tool_call("read_file", {"path": "greeting.txt"}, "c1")]},
        {"tool_calls": [_tool_call("read_file", {"path": "greeting.txt"}, "c2")]},
        {"content": _VERDICT},
    ], release)
    result = NativeToolRoundReviewExecutor(_assignment(repo), llm=llm).execute()

    receipts = result.usage["native_tool_receipts"]
    assert receipts[0]["outcome"] == "error"
    assert receipts[1]["outcome"] == "executed"  # the second call DID return
    assert all("start_line" not in receipt for receipt in receipts)
    assert result.usage["native_source_gap"] == "native_tool_abandoned"