diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index fac35ea9..6ab44d2d 100644 --- a/services/hermes/agent-configmap.yaml +++ b/services/hermes/agent-configmap.yaml @@ -237,6 +237,13 @@ data: synthesize all child evidence in the foreground coordinator. Do not delegate a one-tool mechanical action merely to create an agent. + Only the foreground durable worker owns its Kanban task lifecycle. + Delegated children, including independent reviewers, must return findings + to that foreground worker and must never complete, block, unblock, reclaim, + or otherwise mutate the parent task. The foreground worker must evaluate + those findings, finish any required repair and verification, and emit the + task's final structured result itself. + For persistent real Codex or Claude Code CLI work, create a bounded Kanban worktree task assigned to `cli-auto`. The direct lane reserves the task atomically, sends every start/retry/continuation boundary through Switchyard and its diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index d1116626..d2984e8d 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -8,6 +8,7 @@ import json import os import re import selectors +import signal import subprocess import sys import threading @@ -340,6 +341,7 @@ def stream_process( stderr=subprocess.STDOUT, text=True, bufsize=1, + start_new_session=True, ) assert process.stdout is not None selector = selectors.DefaultSelector() @@ -356,13 +358,13 @@ def stream_process( while process.poll() is None: now = time.monotonic() if now - started > max_runtime: - process.terminate() + _terminate_worker_process(process) 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"): - process.terminate() + _terminate_worker_process(process) forced_failure = "Kanban lease was lost; provider process terminated" lines.append(forced_failure + "\n") break @@ -378,12 +380,7 @@ def stream_process( log.flush() parsed = _event_payload(provider, line, state, state_file) structured = parsed or structured - if process.poll() is None: - try: - process.wait(timeout=10) - except subprocess.TimeoutExpired: - process.kill() - process.wait(timeout=10) + _terminate_worker_process(process) remainder = process.stdout.read() if remainder: lines.append(remainder) @@ -405,6 +402,31 @@ 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: + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait(timeout=10) + + # 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 + + def _base_env() -> dict[str, str]: env = os.environ.copy() env.update( diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index a6272a8f..17eeecee 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -73,6 +73,15 @@ def test_chat_config_enables_real_research_compute_and_delegation(): assert "code_execution" not in toolsets +def test_agent_config_keeps_delegated_reviewers_from_owning_task_lifecycle(): + configmap = _documents(HERMES / "agent-configmap.yaml")[0] + instructions = configmap["data"]["AGENTS.md"] + + assert "Only the foreground durable worker owns its Kanban task lifecycle" in instructions + assert "must never complete, block, unblock, reclaim" in instructions + assert "task's final structured result itself" in instructions + + def test_sandbox_shares_only_the_tenant_workspace_without_credentials(): sandbox_docs = _documents(HERMES / "chat-sandbox.yaml") deployments = [doc for doc in sandbox_docs if doc["kind"] == "Deployment"] diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index cbe30fc5..d2ce3c64 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -5,6 +5,7 @@ from __future__ import annotations import importlib.util import json import os +import signal import sys from contextlib import nullcontext from pathlib import Path @@ -247,6 +248,47 @@ def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: assert result.capacity_failure is False +def test_worker_process_uses_isolated_process_session(monkeypatch, tmp_path: Path): + original_popen = lanes.subprocess.Popen + calls = [] + + def recording_popen(*args, **kwargs): + calls.append(kwargs) + return original_popen(*args, **kwargs) + + monkeypatch.setattr(lanes.subprocess, "Popen", recording_popen) + result = lanes.stream_process( + [sys.executable, "-c", "print('done')"], + provider="codex", + cwd=tmp_path, + env=dict(os.environ), + log_path=tmp_path / "worker.log", + state={}, + state_file=tmp_path / "state.json", + heartbeat=lambda _note: True, + max_runtime=60, + ) + + assert result.returncode == 0 + assert calls[0]["start_new_session"] is True + + +def test_worker_process_group_is_killed_after_leader_exits(monkeypatch): + class FinishedProcess: + pid = 4321 + + @staticmethod + def poll(): + return 0 + + signals = [] + monkeypatch.setattr(lanes.os, "killpg", lambda pid, sig: signals.append((pid, sig))) + + lanes._terminate_worker_process(FinishedProcess()) + + assert signals == [(4321, signal.SIGKILL)] + + def test_unassigned_ready_task_is_persistently_routed_to_auto_lane(monkeypatch): task = SimpleNamespace(id="t_auto", assignee=None, status="ready") assigned = []