FESADev/tests/test_harness_timeouts.py

import io
import json
import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest

from scripts import execute
from scripts.hooks import stop_validation
from scripts.msvc_harness.models import (
    CommandSpec,
    ProjectKind,
    ValidationPlan,
    ValidationStep,
)
from scripts.msvc_harness.process import ValidationFailure, execute_plan


def WriteExecutor(tmp_path, monkeypatch):
    phase_dir = tmp_path / "phases" / "timeout-phase"
    phase_dir.mkdir(parents=True)
    (phase_dir.parent / "index.json").write_text(
        json.dumps({"phases": [{"dir": "timeout-phase", "status": "pending"}]}),
        encoding="utf-8",
    )
    (phase_dir / "index.json").write_text(
        json.dumps(
            {
                "project": "Harness Tests",
                "phase": "timeout-phase",
                "steps": [{"step": 0, "name": "timeout", "status": "pending"}],
            }
        ),
        encoding="utf-8",
    )
    (phase_dir / "step0.md").write_text("timeout", encoding="utf-8")
    profile = tmp_path / ".codex" / "agents" / "implementation-agent.toml"
    profile.parent.mkdir(parents=True)
    profile.write_text(
        'model = "gpt-5.6-terra"\n'
        'model_reasoning_effort = "xhigh"\n'
        'developer_instructions = """test"""\n',
        encoding="utf-8",
    )
    monkeypatch.setattr(execute, "ROOT", tmp_path)
    return execute.StepExecutor("timeout-phase")


def TestClock(values):
    iterator = iter(values)
    return lambda: next(iterator)


def test_executor_uses_finite_default_for_actual_timeout(tmp_path, monkeypatch):
    executor = WriteExecutor(tmp_path, monkeypatch)
    timeouts = []

    def timeout_run(command, **kwargs):
        timeouts.append(kwargs["timeout"])
        raise subprocess.TimeoutExpired(command, kwargs["timeout"])

    monkeypatch.setattr(execute.subprocess, "run", timeout_run)
    output = executor._invoke_codex({"step": 0, "name": "timeout"}, "")

    assert execute.StepExecutor.CODEX_TIMEOUT_SECONDS == 14400
    assert timeouts == [14400]
    assert output["transportFailure"] == "timeout"
    assert "Codex timed out after 14400 seconds." in output["stderr"]


def test_stop_passes_one_shared_deadline_to_root_toolchain_and_plan(
    tmp_path, monkeypatch
):
    root = tmp_path
    selection = SimpleNamespace(kind=stop_validation.ProjectKind.CMAKE)
    calls = {}

    monkeypatch.setattr(stop_validation, "load_config", lambda _: object())
    monkeypatch.setattr(
        stop_validation,
        "discover_project",
        lambda _, __: SimpleNamespace(selection=selection),
    )

    def discover_toolchain(kind, *, deadline, clock):
        calls["toolchain"] = (kind, deadline, clock())
        return object()

    class Adapter:
        def create_plan(self, *args):
            return object()

    def capture_plan(plan, plan_root, *, env, deadline, clock):
        calls["plan"] = (plan, plan_root, env, deadline, clock())

    monkeypatch.setattr(stop_validation, "discover_toolchain", discover_toolchain)
    monkeypatch.setattr(stop_validation, "CMakeAdapter", Adapter)
    monkeypatch.setattr(stop_validation, "MSBuildAdapter", Adapter)
    monkeypatch.setattr(stop_validation, "execute_plan", capture_plan)

    clock = lambda: 100.0
    assert stop_validation.run_validations(root, clock=clock) == (True, "")
    assert stop_validation.TOTAL_TIMEOUT_SECONDS == 7200
    assert calls["toolchain"][0].value == ProjectKind.CMAKE.value
    assert calls["toolchain"][1] == 7300.0
    assert calls["plan"][1] == root
    assert calls["plan"][3] == 7300.0
    assert calls["plan"][2][stop_validation.REENTRY_ENV] == "1"

    observed = {}
    payload = {"cwd": str(root), "stop_hook_active": False}
    monkeypatch.setattr(stop_validation.sys, "stdin", io.StringIO(json.dumps(payload)))
    monkeypatch.setattr(stop_validation, "is_document_only_turn", lambda *_: False)
    monkeypatch.setattr(
        stop_validation,
        "_project_root",
        lambda *args, **kwargs: observed.setdefault("root", kwargs["deadline"]) and root,
    )
    monkeypatch.setattr(
        stop_validation,
        "run_validations",
        lambda *args, **kwargs: observed.setdefault("validation", kwargs["deadline"])
        and (True, ""),
    )
    assert stop_validation.main(clock=clock) == 0
    assert observed == {"root": 7300.0, "validation": 7300.0}


def test_command_defaults_and_shared_remaining_cap_are_preserved(tmp_path):
    default_command = CommandSpec(("default",), tmp_path, "default")
    assert default_command.timeout_seconds == 7200
    plan = ValidationPlan(
        ProjectKind.CMAKE,
        (
            ValidationStep(default_command),
            ValidationStep(CommandSpec(("remaining",), tmp_path, "remaining")),
            ValidationStep(CommandSpec(("explicit",), tmp_path, "explicit", 5)),
        ),
    )
    observed = []

    def run(command, **kwargs):
        observed.append(kwargs["timeout"])
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    execute_plan(
        plan,
        tmp_path,
        run=run,
        clock=TestClock((0.0, 0.0, 2000.0, 2000.0)),
    )
    assert observed == [7200, 5200.0, 5]

    expired_plan = ValidationPlan(
        ProjectKind.CMAKE,
        (ValidationStep(default_command),),
    )
    with pytest.raises(ValidationFailure, match="timed out before it could start"):
        execute_plan(expired_plan, tmp_path, deadline=0.0, clock=lambda: 0.0)

    def nonzero(command, **kwargs):
        return subprocess.CompletedProcess(command, 1, stdout="", stderr="failed")

    with pytest.raises(ValidationFailure, match="failed with exit code 1"):
        execute_plan(expired_plan, tmp_path, run=nonzero, clock=TestClock((0.0, 0.0)))


def test_registered_hook_and_one_time_invocation_contract(monkeypatch):
    root = Path(__file__).resolve().parents[1]
    hooks = json.loads((root / ".codex" / "hooks.json").read_text(encoding="utf-8"))
    pre_tool = hooks["hooks"]["PreToolUse"][0]["hooks"][0]
    stop = hooks["hooks"]["Stop"][0]["hooks"][0]

    assert stop["timeout"] == 7500
    assert pre_tool["timeout"] == 30
    assert execute.StepExecutor.CODEX_TIMEOUT_SECONDS > stop["timeout"]
    assert stop["timeout"] > stop_validation.TOTAL_TIMEOUT_SECONDS
    assert all(
        timeout > 0
        for timeout in (
            execute.StepExecutor.CODEX_TIMEOUT_SECONDS,
            stop["timeout"],
            stop_validation.TOTAL_TIMEOUT_SECONDS,
            CommandSpec(("test",), root, "test").timeout_seconds,
        )
    )

    main_calls = []
    monkeypatch.setattr(execute.StepExecutor, "CODEX_TIMEOUT_SECONDS", 1800)
    monkeypatch.setattr(
        execute,
        "main",
        lambda: main_calls.append(execute.StepExecutor.CODEX_TIMEOUT_SECONDS),
    )
    exec(
        "execute.StepExecutor.CODEX_TIMEOUT_SECONDS = 14400; execute.main()",
        {"execute": execute},
    )
    assert main_calls == [14400]