atlas-iac/testing/tests/test_hermes_execution_pool_worker_execute.py

"""Worker execution, fallback, terminal handoff, and exception contracts."""

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest


ROOT = Path(__file__).parents[2]
SCRIPTS = ROOT / "services/hermes/scripts"
SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts"
sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)]

from cli_lane_config import ProcessResult, Route  # noqa: E402
import execution_pool_protocol as protocol  # noqa: E402
import execution_pool_worker as worker  # noqa: E402
import gitea_api_policy as gitea_policy  # noqa: E402
from testing.tests.test_hermes_execution_pool_worker_v2 import assignment  # noqa: E402


def route(provider="codex", effort="high"):
    return Route(
        provider=provider,
        model=f"{provider}-model",
        effort=effort,
        profile="agent",
        classifier="test",
        reason="unit",
        latency_ms=1,
        fallback_chain=(provider,),
    )


def completed_result(**changes):
    structured = {
        "status": "completed",
        "summary": "Completed safely.",
        "changed_files": ["safe.py"],
        "tests_run": ["pytest"],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    }
    value = ProcessResult(
        returncode=0,
        output="done",
        structured=structured,
        capacity_failure=False,
    )
    for name, replacement in changes.items():
        setattr(value, name, replacement)
    return value


def prepare_execute(tmp_path, monkeypatch, *, item=None, results=None, git_status=""):
    worker_root = tmp_path / "worker"
    workspace = worker_root / "runs/metis/t_deadbeef/23"
    workspace.mkdir(parents=True)
    exact = item or assignment(
        workspace=str(workspace),
        payload={
            "context": "safe objective",
            "assignee": "cli-auto",
            "deadline_unix": 10_000_000_000,
            "max_runtime_seconds": 3600,
        },
    )
    monkeypatch.setattr(worker, "ROOT", worker_root)
    monkeypatch.setattr(worker, "ORDINAL", 0)
    monkeypatch.setattr(worker, "NODE", "titan-05")
    monkeypatch.setattr(worker, "_bind_provider_sessions", lambda _a: None)
    monkeypatch.setattr(worker.cli_lane_runner, "load_json", lambda _path: {})
    routes = [route("codex"), route("claude")]
    route_calls = []

    def select_route(*arguments, **keywords):
        route_calls.append((arguments, keywords))
        return routes[min(len(route_calls) - 1, len(routes) - 1)]

    monkeypatch.setattr(worker.cli_lane_runner, "select_route", select_route)
    monkeypatch.setattr(worker.cli_lane_runner, "fresh_unavailable_provider", lambda: "claude")
    monkeypatch.setattr(worker.cli_lane_runner, "git_handoff", lambda *_a: "\nhandoff")
    outcomes = list(results or [completed_result()])
    provider_calls = []

    def run_provider(*arguments):
        provider_calls.append(arguments)
        return outcomes.pop(0)

    monkeypatch.setattr(worker.cli_lane_runner, "run_provider", run_provider)
    client_calls = []

    def client(operation, **values):
        client_calls.append((operation, values))
        return {"ack": {"accepted": True}}

    monkeypatch.setattr(worker, "_client", client)
    monkeypatch.setattr(worker, "_git", lambda *_a: git_status)
    refreshed = []
    monkeypatch.setattr(worker, "_refresh_assignment", lambda exact: refreshed.append(exact))
    return exact, workspace, client_calls, provider_calls, route_calls, refreshed


def test_execute_completed_clean_result_refreshes_and_finishes_exact_run(
    tmp_path, monkeypatch
):
    exact, _workspace, calls, providers, routes, refreshed = prepare_execute(
        tmp_path, monkeypatch
    )
    worker.execute(exact)
    assert len(providers) == 1 and len(routes) == 1
    assert refreshed == [worker._binding(exact)]
    operations = [name for name, _values in calls]
    assert operations == ["heartbeat", "finish"]
    finish = calls[-1][1]["payload"]
    assert finish["structured"]["status"] == "completed"
    assert finish["node"] == "titan-05"
    state = json.loads(worker._state_path(exact).read_text())
    assert state["terminal_at"] > 0 and state["baseline_sha"] == "a" * 40


@pytest.mark.parametrize("summary", ["x" * 300, "界" * 300, "Ж" * 300])
def test_execute_titles_are_accepted_by_gitea_draft_policy(tmp_path, monkeypatch, summary):
    """Ordinary worker publication preserves Gitea's character and byte limits."""
    _exact, _workspace, calls, _providers, _routes, _refreshed = prepare_execute(
        tmp_path, monkeypatch, results=[completed_result(structured={
            "status": "completed", "summary": summary, "changed_files": [],
            "tests_run": [], "artifacts": [], "findings": [], "blockers": [],
        })]
    )

    worker.execute(_exact)

    title = calls[-1][1]["title"]
    assert len(title) <= 240 and len(title.encode()) <= 507
    assert gitea_policy._draft_title(title) == "WIP: " + title


def test_execute_capacity_fallback_changes_provider_and_preserves_handoff(
    tmp_path, monkeypatch
):
    first = completed_result(capacity_failure=True, output="capacity")
    second = completed_result()
    exact, _workspace, calls, providers, routes, _refreshed = prepare_execute(
        tmp_path, monkeypatch, results=[first, second]
    )
    worker.execute(exact)
    assert len(providers) == 2 and len(routes) == 2
    assert providers[1][0].provider == "claude"
    assert providers[1][1].endswith("handoff")
    heartbeats = [value for operation, value in calls if operation == "heartbeat"]
    assert any("fallback=codex->claude" in value["payload"]["note"] for value in heartbeats)


def test_execute_dirty_workspace_downgrades_completed_result(tmp_path, monkeypatch):
    exact, _workspace, calls, _providers, _routes, refreshed = prepare_execute(
        tmp_path, monkeypatch, git_status="?? untracked"
    )
    worker.execute(exact)
    structured = calls[-1][1]["payload"]["structured"]
    assert structured["status"] == "incomplete"
    assert "uncommitted or untracked" in structured["blockers"][0]
    assert refreshed == []


def test_execute_fills_missing_lists_for_failed_provider_result(tmp_path, monkeypatch):
    failed = ProcessResult(
        returncode=1,
        output="failed",
        structured={"status": "blocked", "summary": "Provider failed."},
        capacity_failure=False,
    )
    exact, _workspace, calls, _providers, _routes, _refreshed = prepare_execute(
        tmp_path, monkeypatch, results=[failed]
    )
    worker.execute(exact)
    structured = calls[-1][1]["payload"]["structured"]
    for name in ("changed_files", "tests_run", "artifacts", "findings", "blockers"):
        assert structured[name] == []


def test_execute_rejects_payload_lease_baseline_and_terminal_ack_failures(
    tmp_path, monkeypatch
):
    exact, *_ = prepare_execute(tmp_path, monkeypatch)
    with pytest.raises(protocol.ProtocolError, match="payload"):
        worker.execute({**exact, "payload": []})

    monkeypatch.setattr(worker, "_client", lambda *_a, **_k: {"ack": {"accepted": False}})
    with pytest.raises(protocol.ProtocolError, match="lease"):
        worker.execute(exact)

    exact, *_ = prepare_execute(tmp_path / "baseline", monkeypatch)
    exact["baseline_sha"] = "invalid"
    with pytest.raises(RuntimeError, match="baseline"):
        worker.execute(exact)

    exact, *_ = prepare_execute(tmp_path / "ack", monkeypatch)
    calls = []

    def client(operation, **_values):
        calls.append(operation)
        return {"ack": {"accepted": operation != "finish"}}

    monkeypatch.setattr(worker, "_client", client)
    with pytest.raises(protocol.ProtocolError, match="terminal"):
        worker.execute(exact)


def test_execute_heartbeat_transport_failure_loses_lease(tmp_path, monkeypatch):
    exact, *_ = prepare_execute(tmp_path, monkeypatch)
    monkeypatch.setattr(
        worker, "_client", lambda *_a, **_k: (_ for _ in ()).throw(ValueError("offline"))
    )
    with pytest.raises(protocol.ProtocolError, match="lease"):
        worker.execute(exact)


def test_report_exception_is_transient_exact_and_transport_safe(monkeypatch):
    calls = []

    def client(operation, **values):
        calls.append((operation, values))
        return {"ack": {"accepted": True}}

    monkeypatch.setattr(worker, "_client", client)
    assert worker.report_exception(assignment(), RuntimeError("boom"))
    payload = calls[0][1]["payload"]
    assert payload["capacity_failure"] is True
    assert payload["structured"]["status"] == "blocked"
    assert payload["structured"]["blockers"] == ["RuntimeError: boom"]
    monkeypatch.setattr(
        worker, "_client", lambda *_a, **_k: (_ for _ in ()).throw(OSError("offline"))
    )
    assert worker.report_exception(assignment(), OSError("boom")) is False


def test_main_idle_and_exception_paths_do_not_spin_silently(monkeypatch, capsys):
    monkeypatch.setattr(worker, "readiness", lambda: None)
    monkeypatch.setattr(worker, "garbage_collect", lambda: 0)
    monkeypatch.setattr(worker, "_poll", lambda: None)
    monkeypatch.setattr(
        worker.time, "sleep", lambda _seconds: (_ for _ in ()).throw(SystemExit("stop"))
    )
    with pytest.raises(SystemExit, match="stop"):
        worker.main()

    exact = assignment()
    monkeypatch.setattr(worker, "_poll", lambda: exact)
    monkeypatch.setattr(worker, "execute", lambda _a: (_ for _ in ()).throw(RuntimeError("boom")))
    monkeypatch.setattr(worker, "report_exception", lambda _a, _e: True)
    with pytest.raises(SystemExit, match="stop"):
        worker.main()
    assert "surfaced" in capsys.readouterr().out

    monkeypatch.setattr(
        worker, "garbage_collect", lambda: (_ for _ in ()).throw(RuntimeError("gc"))
    )
    monkeypatch.setattr(worker, "report_exception", lambda *_a: False)
    with pytest.raises(SystemExit, match="stop"):
        worker.main()
    assert "deferred" in capsys.readouterr().out