From 29eeb1ebf88ea9400fb2b5ec6b9954ac3c98996f Mon Sep 17 00:00:00 2001 From: jenkins Date: Sat, 15 Aug 2026 07:07:21 -0300 Subject: [PATCH] hermes: recover workers after pod restart --- services/hermes/agent-configmap.yaml | 5 + services/hermes/agent-deployment.yaml | 20 +++- services/hermes/kustomization.yaml | 1 + .../scripts/recover_cassandra_workers.py | 59 ++++++++++ testing/tests/test_hermes_worker_recovery.py | 104 ++++++++++++++++++ 5 files changed, 188 insertions(+), 1 deletion(-) create mode 100755 services/hermes/scripts/recover_cassandra_workers.py create mode 100644 testing/tests/test_hermes_worker_recovery.py diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index 6ab44d2d..5719a163 100644 --- a/services/hermes/agent-configmap.yaml +++ b/services/hermes/agent-configmap.yaml @@ -244,6 +244,11 @@ data: those findings, finish any required repair and verification, and emit the task's final structured result itself. + For an implementation or verification task, "review-ready" is a completed + task outcome with evidence, not a reason to call `kanban_block`. Request a + block only for a genuine external decision or unavailable capability. A + later review or acceptance card owns any downstream approval gate. + 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/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index a7cabb0b..009b9ba4 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: "20260815-durable-worker-lifecycle" + ai.bstein.dev/config-rev: "20260815-worker-restart-recovery" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/role: hermes-agent vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens @@ -255,6 +255,24 @@ spec: resources: requests: {cpu: 25m, memory: 64Mi} limits: {cpu: 250m, memory: 256Mi} + - name: recover-cassandra-workers + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: + - /opt/hermes/.venv/bin/python + - /opt/coordinator/recover_cassandra_workers.py + securityContext: + allowPrivilegeEscalation: false + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: + type: RuntimeDefault + volumeMounts: + - {name: home, mountPath: /opt/data} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + resources: + requests: {cpu: 25m, memory: 64Mi} + limits: {cpu: 250m, memory: 256Mi} - name: patch-auth image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 imagePullPolicy: IfNotPresent diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 5afa5965..ce561ab6 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -81,6 +81,7 @@ configMapGenerator: - patch_tui_gateway.py=scripts/patch_tui_gateway.py - patch_ttyd_index.py=scripts/patch_ttyd_index.py - repair_cassandra_kanban.py=scripts/repair_cassandra_kanban.py + - recover_cassandra_workers.py=scripts/recover_cassandra_workers.py - routing_catalog.py=scripts/routing_catalog.py - telegram_media_server.py=scripts/telegram_media_server.py options: diff --git a/services/hermes/scripts/recover_cassandra_workers.py b/services/hermes/scripts/recover_cassandra_workers.py new file mode 100755 index 00000000..ed0c12da --- /dev/null +++ b/services/hermes/scripts/recover_cassandra_workers.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""Requeue Cassandra workers that were interrupted by a Hermes pod restart.""" + +from __future__ import annotations + +import os +from typing import Any + + +BOARD = "cassandra" +REASON = "Hermes pod restarted; resume the durable task from its saved context" + + +def recover_running_tasks(kanban_db: Any) -> list[str]: + """Reclaim every running Cassandra task and return the recovered task IDs.""" + if not kanban_db.board_exists(BOARD): + return [] + + recovered: list[str] = [] + with kanban_db.scoped_current_board(BOARD): + connection = kanban_db.connect(board=BOARD) + try: + for task in kanban_db.list_tasks(connection): + if str(getattr(task, "status", "")) != "running": + continue + task_id = str(getattr(task, "id", "")) + if not task_id or not kanban_db.reclaim_task( + connection, + task_id, + reason=REASON, + ): + continue + kanban_db.add_comment( + connection, + task_id, + "pod-recovery", + "Requeued after the Hermes pod restart; the replacement pod will resume this task.", + ) + recovered.append(task_id) + finally: + connection.close() + return recovered + + +def main() -> int: + """Load Hermes's Kanban API and reclaim interrupted Cassandra work.""" + os.environ.setdefault("HERMES_HOME", "/opt/data") + from hermes_cli import kanban_db + + recovered = recover_running_tasks(kanban_db) + if recovered: + print(f"Requeued {len(recovered)} interrupted Cassandra task(s): {', '.join(recovered)}") + else: + print("No interrupted Cassandra tasks needed recovery") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/testing/tests/test_hermes_worker_recovery.py b/testing/tests/test_hermes_worker_recovery.py new file mode 100644 index 00000000..77307bd3 --- /dev/null +++ b/testing/tests/test_hermes_worker_recovery.py @@ -0,0 +1,104 @@ +"""Tests for restart recovery of durable Cassandra workers.""" + +from __future__ import annotations + +import importlib.util +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace + + +MODULE_PATH = ( + Path(__file__).parents[2] + / "services" + / "hermes" + / "scripts" + / "recover_cassandra_workers.py" +) + + +def _load_module(): + spec = importlib.util.spec_from_file_location("recover_cassandra_workers", MODULE_PATH) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class FakeConnection: + """Minimal connection that records closure.""" + + def __init__(self) -> None: + self.closed = False + + def close(self) -> None: + self.closed = True + + +class FakeKanban: + """Small fake implementing the Kanban operations used by recovery.""" + + def __init__(self, tasks, *, board_exists: bool = True) -> None: + self.tasks = tasks + self.has_board = board_exists + self.connection = FakeConnection() + self.reclaimed: list[tuple[str, str]] = [] + self.comments: list[tuple[str, str, str]] = [] + + def board_exists(self, board: str) -> bool: + assert board == "cassandra" + return self.has_board + + def scoped_current_board(self, board: str): + assert board == "cassandra" + return nullcontext() + + def connect(self, *, board: str): + assert board == "cassandra" + return self.connection + + def list_tasks(self, connection): + assert connection is self.connection + return self.tasks + + def reclaim_task(self, connection, task_id: str, *, reason: str) -> bool: + assert connection is self.connection + self.reclaimed.append((task_id, reason)) + return task_id != "t_race" + + def add_comment(self, connection, task_id: str, author: str, body: str) -> None: + assert connection is self.connection + self.comments.append((task_id, author, body)) + + +def test_recover_running_tasks_requeues_only_claimed_workers() -> None: + module = _load_module() + kanban = FakeKanban( + [ + SimpleNamespace(id="t_running", status="running"), + SimpleNamespace(id="t_done", status="done"), + SimpleNamespace(id="t_race", status="running"), + ] + ) + + assert module.recover_running_tasks(kanban) == ["t_running"] + assert kanban.reclaimed == [ + ("t_running", module.REASON), + ("t_race", module.REASON), + ] + assert kanban.comments == [ + ( + "t_running", + "pod-recovery", + "Requeued after the Hermes pod restart; the replacement pod will resume this task.", + ) + ] + assert kanban.connection.closed + + +def test_recover_running_tasks_is_a_noop_before_board_bootstrap() -> None: + module = _load_module() + kanban = FakeKanban([], board_exists=False) + + assert module.recover_running_tasks(kanban) == [] + assert not kanban.connection.closed