hermes: recover workers after pod restart

This commit is contained in:
jenkins 2026-08-15 07:07:21 -03:00
parent b4634fa275
commit 29eeb1ebf8
5 changed files with 188 additions and 1 deletions

View File

@ -244,6 +244,11 @@ data:
those findings, finish any required repair and verification, and emit the those findings, finish any required repair and verification, and emit the
task's final structured result itself. 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 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, worktree task assigned to `cli-auto`. The direct lane reserves the task atomically,
sends every start/retry/continuation boundary through Switchyard and its sends every start/retry/continuation boundary through Switchyard and its

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/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/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/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/agent-inject: "true"
vault.hashicorp.com/role: hermes-agent vault.hashicorp.com/role: hermes-agent
vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens
@ -255,6 +255,24 @@ spec:
resources: resources:
requests: {cpu: 25m, memory: 64Mi} requests: {cpu: 25m, memory: 64Mi}
limits: {cpu: 250m, memory: 256Mi} 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 - name: patch-auth
image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107
imagePullPolicy: IfNotPresent imagePullPolicy: IfNotPresent

View File

@ -81,6 +81,7 @@ configMapGenerator:
- patch_tui_gateway.py=scripts/patch_tui_gateway.py - patch_tui_gateway.py=scripts/patch_tui_gateway.py
- patch_ttyd_index.py=scripts/patch_ttyd_index.py - patch_ttyd_index.py=scripts/patch_ttyd_index.py
- repair_cassandra_kanban.py=scripts/repair_cassandra_kanban.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 - routing_catalog.py=scripts/routing_catalog.py
- telegram_media_server.py=scripts/telegram_media_server.py - telegram_media_server.py=scripts/telegram_media_server.py
options: options:

View File

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

View File

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