hermes: isolate durable worker lifecycle

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

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(

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"]

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 = []