diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 15b097f7..ad2175fa 100644 --- a/services/hermes/agent-deployment.yaml +++ b/services/hermes/agent-deployment.yaml @@ -25,7 +25,7 @@ spec: ai.bstein.dev/execution: Hermes Kanban with durable direct Codex and Claude Code CLI workers ai.bstein.dev/model-policy: Jetson-assisted AUTO routing, low through xhigh, cross-provider fallback ai.bstein.dev/placement: rpi5 preferred; Jetson deferred until state storage is available - ai.bstein.dev/config-rev: "20260816-cli-board-storage-recovery-v1" + ai.bstein.dev/config-rev: "20260816-cli-board-fresh-connections-v2" prometheus.io/scrape: "true" prometheus.io/path: /metrics prometheus.io/port: "9010" diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 35ed3ba4..768702a0 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -39,6 +39,7 @@ DEFAULT_CLAIM_TTL = 7 * 24 * 60 * 60 DEFAULT_MAX_RUNTIME = 12 * 60 * 60 HEARTBEAT_SECONDS = 20 PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60 +KANBAN_STORAGE_ATTEMPTS = 5 PROVIDER_HEALTH_PATHS = { "codex": DATA_ROOT / "provider-health/codex.json", "claude": DATA_ROOT / "provider-health/claude.json", @@ -394,6 +395,7 @@ def stream_process( selector.register(process.stdout, selectors.EVENT_READ) started = time.monotonic() last_heartbeat = 0.0 + known_descendants: dict[int, tuple[int, int]] = {} lines: list[str] = [] structured: dict[str, Any] | None = None log_path.parent.mkdir(parents=True, exist_ok=True) @@ -402,15 +404,16 @@ def stream_process( log.write(f"\n[{utc_now()}] starting {provider} worker\n") log.flush() while process.poll() is None: + known_descendants.update(_descendant_processes(process.pid)) now = time.monotonic() if now - started > max_runtime: - _terminate_worker_process(process) + _terminate_worker_process(process, known_descendants) forced_failure = "worker exceeded its maximum runtime" lines.append(forced_failure + "\n") break if now - last_heartbeat >= HEARTBEAT_SECONDS: if not heartbeat(f"{provider} worker active for {round(now - started)}s"): - _terminate_worker_process(process) + _terminate_worker_process(process, known_descendants) forced_failure = "Kanban lease was lost; provider process terminated" lines.append(forced_failure + "\n") break @@ -426,7 +429,7 @@ def stream_process( log.flush() parsed = _event_payload(provider, line, state, state_file) structured = parsed or structured - _terminate_worker_process(process) + _terminate_worker_process(process, known_descendants) remainder = process.stdout.read() if remainder: lines.append(remainder) @@ -448,29 +451,90 @@ def stream_process( ) -def _terminate_worker_process(process: subprocess.Popen[str]) -> None: - """Reap a worker and every subprocess in its isolated process group.""" - if process.poll() is None: +def _process_record(pid: int) -> tuple[int, int, int] | None: + """Return one Linux process's parent, group, and start-time identity.""" + try: + raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8") + fields = raw[raw.rfind(")") + 2 :].split() + return int(fields[1]), int(fields[2]), int(fields[19]) + except (IndexError, OSError, ValueError): + return None + + +def _descendant_processes(root_pid: int) -> dict[int, tuple[int, int]]: + """Snapshot descendants as pid -> (process group, start-time identity).""" + records: dict[int, tuple[int, int, int]] = {} + for entry in Path("/proc").iterdir(): + if not entry.name.isdigit(): + continue + pid = int(entry.name) + record = _process_record(pid) + if record is not None: + records[pid] = record + family = {root_pid} + changed = True + while changed: + changed = False + for pid, (parent, _, _) in records.items(): + if pid not in family and parent in family: + family.add(pid) + changed = True + return { + pid: (records[pid][1], records[pid][2]) + for pid in family + if pid != root_pid and pid in records + } + + +def _process_identity_matches(pid: int, start_time: int) -> bool: + record = _process_record(pid) + return record is not None and record[2] == start_time + + +def _signal_worker_tree( + root_pid: int, + descendants: dict[int, tuple[int, int]], + sig: signal.Signals, +) -> None: + """Signal captured descendants across terminal-created process groups.""" + groups = {root_pid} + for pid, (process_group, start_time) in descendants.items(): + if _process_identity_matches(pid, start_time): + groups.add(process_group) + for process_group in groups: try: - os.killpg(process.pid, signal.SIGTERM) + os.killpg(process_group, sig) except ProcessLookupError: pass + for pid, (_, start_time) in descendants.items(): + if not _process_identity_matches(pid, start_time): + continue + try: + os.kill(pid, sig) + except ProcessLookupError: + pass + + +def _terminate_worker_process( + process: subprocess.Popen[str], + known_descendants: dict[int, tuple[int, int]] | None = None, +) -> None: + """Reap a worker and terminal descendants that created their own groups.""" + descendants = dict(known_descendants or {}) + descendants.update(_descendant_processes(process.pid)) + if process.poll() is None: + _signal_worker_tree(process.pid, descendants, signal.SIGTERM) try: process.wait(timeout=10) except subprocess.TimeoutExpired: - try: - os.killpg(process.pid, signal.SIGKILL) - except ProcessLookupError: - pass - process.wait(timeout=10) + pass - # A provider can exit before one of its terminal or language-server children. - # The process group remains addressable after its leader exits, so reap those - # children as well before a retry is allowed to own the same worktree. - try: - os.killpg(process.pid, signal.SIGKILL) - except ProcessLookupError: - pass + # Provider terminal tools create their own sessions, so killing only the + # native CLI's process group leaves test/build subprocesses orphaned. The + # identity check prevents a reused PID from targeting an unrelated worker. + _signal_worker_tree(process.pid, descendants, signal.SIGKILL) + if process.poll() is None: + process.wait(timeout=10) def _base_env() -> dict[str, str]: @@ -676,18 +740,46 @@ def _resolve_workspace(kanban_db: Any, conn: Any, task: Any, board: str) -> Path return Path(value).resolve() +def _board_call( + kanban_db: Any, + board: str, + operation: Callable[[Any], Any], +) -> Any: + """Run one bounded Kanban operation on a fresh, promptly closed connection.""" + last_storage_error: Exception | None = None + for attempt in range(KANBAN_STORAGE_ATTEMPTS): + conn = None + try: + with kanban_db.scoped_current_board(board): + conn = kanban_db.connect(board=board) + result = operation(conn) + BOARD_CORRUPTION_ERRORS.pop(board, None) + return result + except (OSError, sqlite3.Error) as error: + last_storage_error = error + _record_board_access_error(board, error) + if attempt + 1 < KANBAN_STORAGE_ATTEMPTS: + time.sleep(0.2 * (attempt + 1)) + finally: + if conn is not None: + conn.close() + assert last_storage_error is not None + raise last_storage_error + + def execute_claim(board: str, task_id: str) -> None: """Execute one already-claimed task and commit its outcome to Kanban.""" from hermes_cli import kanban_db + run_id = None with kanban_db.scoped_current_board(board): conn = kanban_db.connect(board=board) - task = kanban_db.get_task(conn, task_id) - if task is None: - return - run_id = _task_value(task, "current_run_id", None) - assignee = str(_task_value(task, "assignee", "cli-auto") or "cli-auto") try: + task = kanban_db.get_task(conn, task_id) + if task is None: + return + run_id = _task_value(task, "current_run_id", None) + assignee = str(_task_value(task, "assignee", "cli-auto") or "cli-auto") state_file = state_path(board, task_id) state = load_json(state_file) state.update( @@ -704,25 +796,61 @@ def execute_claim(board: str, task_id: str) -> None: kanban_db.set_workspace_path(conn, task_id, str(workspace)) context = _task_context(kanban_db, conn, task_id) except Exception as error: - kanban_db.block_task( - conn, - task_id, - reason=f"Direct CLI lane preparation failed: {type(error).__name__}: {error}", - kind="capability", - expected_run_id=run_id, + failure_reason = ( + f"Direct CLI lane preparation failed: {type(error).__name__}: {error}" ) conn.close() + _board_call( + kanban_db, + board, + lambda fresh: kanban_db.block_task( + fresh, + task_id, + reason=failure_reason, + kind="capability", + expected_run_id=run_id, + ), + ) return + finally: + conn.close() def heartbeat(note: str) -> bool: - return bool( - kanban_db.heartbeat_worker( - conn, - task_id, - note=note, - expected_run_id=run_id, + try: + return bool( + _board_call( + kanban_db, + board, + lambda fresh: kanban_db.heartbeat_worker( + fresh, + task_id, + note=note, + expected_run_id=run_id, + ), + ) ) - ) + except (OSError, sqlite3.Error): + # The claim TTL is deliberately long. Keep the expensive + # provider process alive during a transient volume stall and + # retry on the next heartbeat instead of losing its work. + return True + + def comment(body: str) -> None: + try: + _board_call( + kanban_db, + board, + lambda fresh: kanban_db.add_comment( + fresh, + task_id, + "cli-lane-runner", + body, + ), + ) + except (OSError, sqlite3.Error): + # Route state is also written to the durable lane-state file; + # a later heartbeat or terminal result remains authoritative. + pass try: previous_route = state.get("current_route") @@ -738,16 +866,10 @@ def execute_claim(board: str, task_id: str) -> None: else None, ) if excluded_provider: - kanban_db.add_comment( - conn, - task_id, - "cli-lane-runner", + comment( f"Provider health guard excluded {excluded_provider} before automatic routing.", ) - kanban_db.add_comment( - conn, - task_id, - "cli-lane-runner", + comment( f"CLI route: {route.provider}/{route.model} at {route.effort}; classifier={route.classifier}; {route.reason}", ) resume_handoff = "" @@ -761,10 +883,7 @@ def execute_claim(board: str, task_id: str) -> None: except OSError: previous_output = "Previous provider log was unavailable after restart." resume_handoff = git_handoff(workspace, previous_output) - kanban_db.add_comment( - conn, - task_id, - "cli-lane-runner", + comment( f"Restart-time provider change: {previous_route.get('provider')} -> {route.provider}; explicit workspace handoff attached.", ) prompt = build_prompt(context, workspace, resume_handoff) @@ -783,10 +902,7 @@ def execute_claim(board: str, task_id: str) -> None: retry_context, f"cli-{alternate}-{route.effort}", ) - kanban_db.add_comment( - conn, - task_id, - "cli-lane-runner", + comment( f"Provider fallback: {route.provider} -> {fallback.provider}; Jetson reclassified the retry boundary.", ) result = run_provider( @@ -822,13 +938,17 @@ def execute_claim(board: str, task_id: str) -> None: value = structured.get(key) metadata[key] = value if isinstance(value, list) else [] if structured and structured.get("status") == "completed" and result.returncode == 0: - kanban_db.complete_task( - conn, - task_id, - result=json.dumps(structured, sort_keys=True), - summary=str(structured.get("summary") or "Completed"), - metadata=metadata, - expected_run_id=run_id, + _board_call( + kanban_db, + board, + lambda fresh: kanban_db.complete_task( + fresh, + task_id, + result=json.dumps(structured, sort_keys=True), + summary=str(structured.get("summary") or "Completed"), + metadata=metadata, + expected_run_id=run_id, + ), ) else: reason = ( @@ -836,23 +956,30 @@ def execute_claim(board: str, task_id: str) -> None: if structured else result.output[-4000:] ) - kanban_db.block_task( - conn, - task_id, - reason=reason or f"{route.provider} worker failed with exit {result.returncode}", - kind="transient" if result.capacity_failure else "capability", - expected_run_id=run_id, + _board_call( + kanban_db, + board, + lambda fresh: kanban_db.block_task( + fresh, + task_id, + reason=reason or f"{route.provider} worker failed with exit {result.returncode}", + kind="transient" if result.capacity_failure else "capability", + expected_run_id=run_id, + ), ) except Exception as error: - kanban_db.block_task( - conn, - task_id, - reason=f"Direct CLI lane failed: {type(error).__name__}: {error}", - kind="capability", - expected_run_id=run_id, + failure_reason = f"Direct CLI lane failed: {type(error).__name__}: {error}" + _board_call( + kanban_db, + board, + lambda fresh: kanban_db.block_task( + fresh, + task_id, + reason=failure_reason, + kind="capability", + expected_run_id=run_id, + ), ) - finally: - conn.close() def _board_slug(board: Any) -> str: diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index eb118de6..7e015b02 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -393,6 +393,51 @@ def test_worker_process_group_is_killed_after_leader_exits(monkeypatch): assert signals == [(4321, signal.SIGKILL)] +def test_worker_terminal_descendants_in_separate_groups_are_killed(monkeypatch): + class RunningProcess: + pid = 4321 + running = True + + @classmethod + def poll(cls): + return None if cls.running else 0 + + @classmethod + def wait(cls, timeout): + assert timeout == 10 + cls.running = False + return 0 + + group_signals = [] + process_signals = [] + monkeypatch.setattr( + lanes, + "_descendant_processes", + lambda _pid: {5000: (5000, 12345)}, + ) + monkeypatch.setattr(lanes, "_process_identity_matches", lambda _pid, _start: True) + monkeypatch.setattr( + lanes.os, + "killpg", + lambda process_group, sig: group_signals.append((process_group, sig)), + ) + monkeypatch.setattr( + lanes.os, + "kill", + lambda pid, sig: process_signals.append((pid, sig)), + ) + + lanes._terminate_worker_process(RunningProcess()) + + assert set(group_signals) == { + (4321, signal.SIGTERM), + (5000, signal.SIGTERM), + (4321, signal.SIGKILL), + (5000, signal.SIGKILL), + } + assert process_signals == [(5000, signal.SIGTERM), (5000, signal.SIGKILL)] + + def test_unassigned_ready_task_is_persistently_routed_to_auto_lane(monkeypatch): task = SimpleNamespace(id="t_auto", assignee=None, status="ready") assigned = [] @@ -490,6 +535,41 @@ def test_transient_board_scan_failure_does_not_stop_healthy_lanes(monkeypatch, c assert "storage OperationalError: disk I/O error" in error +def test_board_call_retries_storage_faults_on_fresh_connections(): + connections = [] + + class Connection: + def __init__(self): + self.closed = False + + def close(self): + self.closed = True + + def connect(*, board): + assert board == "cassandra" + connection = Connection() + connections.append(connection) + return connection + + attempts = [] + + def operation(_connection): + attempts.append(1) + if len(attempts) < 3: + raise lanes.sqlite3.OperationalError("disk I/O error") + return "healthy" + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=connect, + ) + lanes.BOARD_CORRUPTION_ERRORS.clear() + + assert lanes._board_call(fake_db, "cassandra", operation) == "healthy" + assert len(connections) == 3 + assert all(connection.closed for connection in connections) + + @pytest.mark.parametrize( ("result", "expected_action"), [ @@ -526,17 +606,27 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts( ) calls = [] heartbeats = [] + connections = [] artifact = tmp_path / "reports/result.json" artifact.parent.mkdir() artifact.write_text("{}\n", encoding="utf-8") class Connection: + def __init__(self): + self.closed = False + def close(self): - return None + self.closed = True + + def connect(*, board): + assert board == "cassandra" + connection = Connection() + connections.append(connection) + return connection fake_db = SimpleNamespace( scoped_current_board=lambda _board: nullcontext(), - connect=lambda board: Connection(), + connect=connect, get_task=lambda _conn, _task_id: task, worker_log_path=lambda _task_id, board: tmp_path / "worker.log", _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_worker"), @@ -560,7 +650,11 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts( ), ) def run_provider(*args, **_kwargs): + assert connections[0].closed + before_heartbeat = len(connections) assert args[6]("working") is True + assert len(connections) == before_heartbeat + 1 + assert connections[-1].closed return result monkeypatch.setattr(lanes, "run_provider", run_provider) @@ -574,6 +668,7 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts( assert calls[0][1]["metadata"]["tests_run"] == ["pytest -q"] else: assert calls[0][1]["kind"] == "capability" + assert all(connection.closed for connection in connections) def test_workspace_preparation_failure_durably_blocks_the_claim(tmp_path: Path, monkeypatch):