titan/atlas-iac
3
0
Fork 0
Code Issues Pull Requests 9 Actions Packages Projects Releases Wiki Activity

hermes: isolate worker database lifetimes

Browse Source
This commit is contained in:
jenkins 2026-08-16 10:30:58 -03:00
parent e758ee1059
commit 2a94c7c74f
3 changed files with 299 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
services/hermes/agent-deployment.yaml
View File

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

275
services/hermes/scripts/cli_lane_runner.py
View File

@ -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:

99
testing/tests/test_hermes_cli_lanes.py
View File

@ -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):