8036101f0b
# Conflicts: # scripts/tests/test_dashboards_render_atlas_drilldowns.py # scripts/tests/test_dashboards_render_jobs.py # services/hermes/scm-common/scripts/scm_broker.py # services/hermes/scripts/cli_lane_dispatch.py # services/hermes/scripts/cli_lane_execution.py # testing/quality_contract.json # testing/tests/test_hermes_agent_access.py # testing/tests/test_hermes_agent_security.py # testing/tests/test_hermes_chat_config.py # testing/tests/test_hermes_chat_images.py # testing/tests/test_hermes_chat_provider_auth.py # testing/tests/test_hermes_chat_quality.py # testing/tests/test_hermes_chat_support.py # testing/tests/test_hermes_chat_voice.py # testing/tests/test_hermes_cli_finalization_edges.py # testing/tests/test_hermes_cli_foundation_coverage.py # testing/tests/test_hermes_cli_lanes_configuration.py # testing/tests/test_hermes_cli_recovery_edges.py # testing/tests/test_hermes_cli_retention_edges.py # testing/tests/test_hermes_coordinator.py # testing/tests/test_hermes_coordinator_boards.py # testing/tests/test_hermes_coordinator_support.py
404 lines
12 KiB
Python
404 lines
12 KiB
Python
"""Process lifecycle and provider artifact edge coverage for CLI workers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from testing.tests.test_hermes_cli_support import _completed_result, lanes
|
|
|
|
|
|
def test_stream_process_enforces_runtime_and_lease_boundaries(
|
|
tmp_path: Path,
|
|
):
|
|
timed_out = lanes.stream_process(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"],
|
|
provider="codex",
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
log_path=tmp_path / "timeout.log",
|
|
state={},
|
|
state_file=tmp_path / "timeout-state.json",
|
|
heartbeat=lambda _note: True,
|
|
max_runtime=-1,
|
|
)
|
|
assert "maximum runtime" in timed_out.output
|
|
assert timed_out.returncode != 0
|
|
|
|
lease_lost = lanes.stream_process(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"],
|
|
provider="claude",
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
log_path=tmp_path / "lease.log",
|
|
state={},
|
|
state_file=tmp_path / "lease-state.json",
|
|
heartbeat=lambda _note: False,
|
|
max_runtime=60,
|
|
)
|
|
assert "lease was lost" in lease_lost.output
|
|
assert lease_lost.returncode != 0
|
|
|
|
|
|
def test_stream_process_bounds_retained_line_history(tmp_path: Path):
|
|
result = lanes.stream_process(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"import sys; [sys.stdout.write(f'{n}\\n') for n in range(4010)]",
|
|
],
|
|
provider="codex",
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
log_path=tmp_path / "bounded.log",
|
|
state={},
|
|
state_file=tmp_path / "state.json",
|
|
heartbeat=lambda _note: True,
|
|
max_runtime=60,
|
|
)
|
|
assert result.returncode == 0
|
|
assert "4009" in result.output
|
|
|
|
|
|
def test_process_identity_and_signal_failures_are_bounded(monkeypatch):
|
|
assert lanes._process_record(999999999) is None
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"_process_identity_matches",
|
|
lambda _pid, _start: True,
|
|
)
|
|
monkeypatch.setattr(
|
|
lanes.os,
|
|
"killpg",
|
|
lambda *_args: (_ for _ in ()).throw(ProcessLookupError()),
|
|
)
|
|
monkeypatch.setattr(
|
|
lanes.os,
|
|
"kill",
|
|
lambda *_args: (_ for _ in ()).throw(ProcessLookupError()),
|
|
)
|
|
lanes._signal_worker_tree(10, {11: (12, 13)}, signal.SIGTERM)
|
|
|
|
signals = []
|
|
monkeypatch.setattr(lanes, "_process_identity_matches", lambda *_args: False)
|
|
monkeypatch.setattr(lanes.os, "killpg", lambda pid, sig: signals.append((pid, sig)))
|
|
monkeypatch.setattr(
|
|
lanes.os,
|
|
"kill",
|
|
lambda *_args: (_ for _ in ()).throw(AssertionError("stale PID signaled")),
|
|
)
|
|
lanes._signal_worker_tree(20, {21: (22, 23)}, signal.SIGKILL)
|
|
assert signals == [(20, signal.SIGKILL)]
|
|
|
|
|
|
def test_descendant_snapshot_follows_multiple_generations(monkeypatch):
|
|
entries = [
|
|
Path("/proc/self"),
|
|
Path("/proc/10"),
|
|
Path("/proc/11"),
|
|
Path("/proc/12"),
|
|
Path("/proc/13"),
|
|
]
|
|
monkeypatch.setattr(lanes.Path, "iterdir", lambda _path: iter(entries))
|
|
records = {
|
|
10: (1, 10, 100),
|
|
11: (10, 11, 101),
|
|
12: (11, 12, 102),
|
|
}
|
|
monkeypatch.setattr(lanes, "_process_record", lambda pid: records.get(pid))
|
|
assert lanes._descendant_processes(10) == {11: (11, 101), 12: (12, 102)}
|
|
|
|
|
|
def test_claude_missing_unstarted_session_does_not_mark_started(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
state = {"claude_session_id": "synthetic-session"}
|
|
state_file = tmp_path / "state.json"
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"stream_process",
|
|
lambda *_args, **_kwargs: lanes.ProcessResult(
|
|
1, lanes.NO_CLAUDE_SESSION, None, False
|
|
),
|
|
)
|
|
route = lanes.Route("claude", "model", "high", "profile", "test", "test", 1, ())
|
|
result = lanes.run_provider(
|
|
route,
|
|
"work",
|
|
tmp_path,
|
|
state,
|
|
state_file,
|
|
tmp_path / "log",
|
|
lambda _note: True,
|
|
60,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "claude_started" not in state
|
|
|
|
|
|
def test_terminate_escalates_after_term_timeout(monkeypatch):
|
|
class Process:
|
|
pid = 321
|
|
waits = 0
|
|
|
|
@classmethod
|
|
def poll(cls):
|
|
return None
|
|
|
|
@classmethod
|
|
def wait(cls, timeout):
|
|
cls.waits += 1
|
|
if cls.waits == 1:
|
|
raise subprocess.TimeoutExpired("worker", timeout)
|
|
return 0
|
|
|
|
signals = []
|
|
monkeypatch.setattr(lanes, "_descendant_processes", lambda _pid: {})
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"_signal_worker_tree",
|
|
lambda _pid, _descendants, sig: signals.append(sig),
|
|
)
|
|
lanes._terminate_worker_process(Process())
|
|
assert signals == [signal.SIGTERM, signal.SIGKILL]
|
|
assert Process.waits == 2
|
|
|
|
|
|
def test_codex_result_collision_and_chmod_failure_preserve_result(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
state = {"run_id": 7, "result_sequence": 0}
|
|
state_file = tmp_path / "task.json"
|
|
first = lanes._result_path(state_file, 7, 1)
|
|
first.write_text("prior", encoding="utf-8")
|
|
result_paths = []
|
|
|
|
def stream(command, **_kwargs):
|
|
result_path = Path(command[command.index("-o") + 1])
|
|
result_path.write_text(json.dumps(_completed_result("new result")), encoding="utf-8")
|
|
result_paths.append(result_path)
|
|
return lanes.ProcessResult(0, "", None, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", stream)
|
|
monkeypatch.setattr(
|
|
lanes.Path,
|
|
"chmod",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("readonly")),
|
|
)
|
|
route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ())
|
|
result = lanes.run_provider(
|
|
route,
|
|
"work",
|
|
tmp_path,
|
|
state,
|
|
state_file,
|
|
tmp_path / "log",
|
|
lambda _note: True,
|
|
60,
|
|
)
|
|
assert result.structured["summary"] == "new result"
|
|
assert result_paths[0].name.endswith("provider-2.result.json")
|
|
|
|
|
|
def test_codex_missing_thread_skips_colliding_restart_result(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
state = {"run_id": 8, "codex_thread_id": "missing"}
|
|
state_file = tmp_path / "task.json"
|
|
second = lanes._result_path(state_file, 8, 2)
|
|
second.write_text("collision", encoding="utf-8")
|
|
calls = []
|
|
|
|
def stream(command, **_kwargs):
|
|
calls.append(command)
|
|
if len(calls) == 1:
|
|
return lanes.ProcessResult(1, lanes.NO_CODEX_THREAD, None, False)
|
|
return lanes.ProcessResult(0, "", None, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", stream)
|
|
route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ())
|
|
lanes.run_provider(
|
|
route,
|
|
"work",
|
|
tmp_path,
|
|
state,
|
|
state_file,
|
|
tmp_path / "log",
|
|
lambda _note: True,
|
|
60,
|
|
)
|
|
restart = Path(calls[1][calls[1].index("-o") + 1])
|
|
assert restart.name.endswith("provider-3.result.json")
|
|
|
|
|
|
def test_stream_process_trims_the_retained_line_window(tmp_path: Path, monkeypatch):
|
|
"""Streamed chatter keeps only the newest bounded output window."""
|
|
monkeypatch.setattr(lanes, "_descendant_processes", lambda _pid: {})
|
|
monkeypatch.setattr(lanes, "_terminate_worker_process", lambda *_a, **_k: None)
|
|
read_fd, write_fd = os.pipe()
|
|
os.write(write_fd, "".join(f"{index}\n" for index in range(4100)).encode())
|
|
os.close(write_fd)
|
|
polls = []
|
|
|
|
class ChatteringProcess:
|
|
pid = 4243
|
|
returncode = 0
|
|
stdout = os.fdopen(read_fd, "r")
|
|
|
|
def poll(self):
|
|
polls.append(True)
|
|
return None if len(polls) < 4300 else 0
|
|
|
|
monkeypatch.setattr(
|
|
lanes.subprocess,
|
|
"Popen",
|
|
lambda *_args, **_kwargs: ChatteringProcess(),
|
|
)
|
|
heartbeats = []
|
|
|
|
result = lanes.stream_process(
|
|
["ignored"],
|
|
provider="codex",
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
log_path=tmp_path / "chatter.log",
|
|
state={},
|
|
state_file=tmp_path / "chatter-state.json",
|
|
heartbeat=lambda note: heartbeats.append(note) or True,
|
|
max_runtime=60,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert heartbeats
|
|
published = result.output.splitlines()
|
|
assert len(published) == 4000
|
|
assert published[0] == "100"
|
|
assert published[-1] == "4099"
|
|
|
|
|
|
def test_stream_process_parses_output_left_after_worker_exit(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""Buffered final output from an already-exited worker is still parsed."""
|
|
payload = json.dumps({"result": json.dumps(_completed_result("buffered"))})
|
|
read_fd, write_fd = os.pipe()
|
|
os.write(write_fd, f"{payload}\n".encode())
|
|
os.close(write_fd)
|
|
fake = type(
|
|
"ExitedProcess",
|
|
(),
|
|
{
|
|
"pid": 4242,
|
|
"returncode": 0,
|
|
"poll": lambda self: 0,
|
|
"stdout": os.fdopen(read_fd, "r"),
|
|
},
|
|
)()
|
|
monkeypatch.setattr(lanes.subprocess, "Popen", lambda *_args, **_kwargs: fake)
|
|
monkeypatch.setattr(lanes, "_terminate_worker_process", lambda *_a, **_k: None)
|
|
|
|
result = lanes.stream_process(
|
|
["ignored"],
|
|
provider="codex",
|
|
cwd=tmp_path,
|
|
env={},
|
|
log_path=tmp_path / "buffered.log",
|
|
state={},
|
|
state_file=tmp_path / "buffered-state.json",
|
|
heartbeat=lambda _note: True,
|
|
max_runtime=60,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert result.structured and result.structured["summary"] == "buffered"
|
|
assert payload in (tmp_path / "buffered.log").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_descendant_snapshot_ignores_processes_that_vanish(monkeypatch):
|
|
entries = [Path("/proc/10"), Path("/proc/11"), Path("/proc/13")]
|
|
monkeypatch.setattr(lanes.Path, "iterdir", lambda _path: iter(entries))
|
|
records = {10: (1, 10, 100), 11: (10, 11, 101)}
|
|
monkeypatch.setattr(lanes, "_process_record", lambda pid: records.get(pid))
|
|
|
|
assert lanes._descendant_processes(10) == {11: (11, 101)}
|
|
|
|
|
|
def test_signal_tree_targets_only_identity_verified_descendants(monkeypatch):
|
|
"""Reused PIDs are never signaled; verified groups and PIDs both are."""
|
|
records = {10: (1, 20, 100)}
|
|
monkeypatch.setattr(lanes, "_process_record", lambda pid: records.get(pid))
|
|
group_signals = []
|
|
pid_signals = []
|
|
monkeypatch.setattr(
|
|
lanes.os,
|
|
"killpg",
|
|
lambda group, sig: group_signals.append((group, sig)),
|
|
)
|
|
monkeypatch.setattr(
|
|
lanes.os,
|
|
"kill",
|
|
lambda pid, sig: pid_signals.append((pid, sig)),
|
|
)
|
|
|
|
lanes._signal_worker_tree(
|
|
5,
|
|
{10: (20, 100), 11: (21, 999)},
|
|
signal.SIGTERM,
|
|
)
|
|
|
|
assert sorted(group_signals) == [(5, signal.SIGTERM), (20, signal.SIGTERM)]
|
|
assert pid_signals == [(10, signal.SIGTERM)]
|
|
|
|
|
|
def test_claude_worker_without_any_recoverable_session_stays_unstarted(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""A session lost on resume and restart never marks the lane as started."""
|
|
state = {"claude_started": True, "claude_session_id": "session-lost"}
|
|
state_file = tmp_path / "state.json"
|
|
commands = []
|
|
|
|
def stream(command, **_kwargs):
|
|
commands.append(command)
|
|
return lanes.ProcessResult(
|
|
1,
|
|
f"{lanes.NO_CLAUDE_SESSION} session-lost",
|
|
None,
|
|
False,
|
|
)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", stream)
|
|
persisted = []
|
|
monkeypatch.setattr(
|
|
sys.modules["cli_lane_provider"],
|
|
"atomic_json",
|
|
lambda path, value: persisted.append(dict(value)),
|
|
)
|
|
route = lanes.Route("claude", "claude-fable-5", "high", "p", "c", "r", 1, ())
|
|
|
|
result = lanes.run_provider(
|
|
route,
|
|
"prompt",
|
|
tmp_path,
|
|
state,
|
|
state_file,
|
|
tmp_path / "worker.log",
|
|
lambda _note: True,
|
|
60,
|
|
)
|
|
|
|
assert result.returncode == 1
|
|
assert len(commands) == 2
|
|
assert "--resume" in commands[0]
|
|
assert "--session-id" in commands[1]
|
|
assert len(persisted) == 1
|