82cc28b383
Some checks failed
Tests / Declarative: Post Actions failed: 42, skipped: 89, passed: 3938
306 lines
11 KiB
Python
306 lines
11 KiB
Python
"""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("assignee", ["cli-codex-high", "CLI-CODEX-FRONTIER-XHIGH"])
|
|
def test_execute_rejects_an_explicitly_disabled_provider_before_any_worker_action(
|
|
tmp_path, monkeypatch, assignee
|
|
):
|
|
"""An unavailable native lane cannot be selected by a manual assignee."""
|
|
exact = assignment(payload={
|
|
"context": "safe objective", "assignee": assignee,
|
|
"deadline_unix": 10_000_000_000, "max_runtime_seconds": 3600,
|
|
})
|
|
exact, _workspace, calls, providers, routes, _refreshed = prepare_execute(
|
|
tmp_path, monkeypatch, item=exact
|
|
)
|
|
monkeypatch.setattr(worker, "DISABLED_PROVIDER", "codex")
|
|
monkeypatch.setattr(
|
|
worker, "_bind_provider_sessions", lambda _a: pytest.fail("staged provider state")
|
|
)
|
|
|
|
with pytest.raises(protocol.ProtocolError, match="disabled provider"):
|
|
worker.execute(exact)
|
|
|
|
assert calls == [] and providers == [] and routes == []
|
|
|
|
|
|
def test_execute_normalizes_auto_before_excluding_disabled_provider(tmp_path, monkeypatch):
|
|
"""Whitespace and case cannot bypass the automatic disabled-provider guard."""
|
|
exact = assignment(payload={
|
|
"context": "safe objective", "assignee": " CLI-AUTO ",
|
|
"deadline_unix": 10_000_000_000, "max_runtime_seconds": 3600,
|
|
})
|
|
exact, workspace, _calls, _providers, routes, _refreshed = prepare_execute(
|
|
tmp_path, monkeypatch, item=exact
|
|
)
|
|
exact["workspace"] = str(workspace)
|
|
monkeypatch.setattr(worker, "DISABLED_PROVIDER", "codex")
|
|
|
|
worker.execute(exact)
|
|
|
|
assert routes[0][1]["exclude_provider"] == "codex"
|
|
|
|
|
|
@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
|