diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 854645c0..15b097f7 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: "20260816-atlas-forge-workflow-v2" + ai.bstein.dev/config-rev: "20260816-cli-board-storage-recovery-v1" prometheus.io/scrape: "true" prometheus.io/path: /metrics prometheus.io/port: "9010" diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 1c9410f6..35ed3ba4 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -9,6 +9,7 @@ import os import re import selectors import signal +import sqlite3 import subprocess import sys import threading @@ -865,28 +866,38 @@ def _external(task: Any) -> bool: def _connect_healthy_board(kanban_db: Any, board: str) -> Any | None: - """Open one board without letting localized corruption stop other lanes.""" + """Open one board without letting localized storage faults stop other lanes.""" try: - conn = kanban_db.connect(board=board) - except getattr(kanban_db, "KanbanDbCorruptError", ()) as error: - detail = str(error) - if BOARD_CORRUPTION_ERRORS.get(board) != detail: - print( - f"quarantining corrupt Kanban board {board!r}: {detail}", - file=sys.stderr, - flush=True, - ) - BOARD_CORRUPTION_ERRORS[board] = detail + return kanban_db.connect(board=board) + except Exception as error: + _record_board_access_error(board, error) return None - BOARD_CORRUPTION_ERRORS.pop(board, None) - return conn + + +def _record_board_access_error(board: str, error: Exception) -> None: + """Report one board access failure once while allowing other lanes to run.""" + failure_kind = "storage" if isinstance(error, (OSError, sqlite3.Error)) else "access" + detail = f"{failure_kind} {type(error).__name__}: {error}" + if BOARD_CORRUPTION_ERRORS.get(board) == detail: + return + print( + f"temporarily skipping Kanban board {board!r}: {detail}", + file=sys.stderr, + flush=True, + ) + BOARD_CORRUPTION_ERRORS[board] = detail def recover_orphans() -> None: """Return external running tasks to ready after a runner/pod restart.""" from hermes_cli import kanban_db - for raw_board in kanban_db.list_boards(include_archived=False): + try: + boards = kanban_db.list_boards(include_archived=False) + except Exception as error: + _record_board_access_error("board-registry", error) + return + for raw_board in boards: board = _board_slug(raw_board) if not board: continue @@ -902,6 +913,9 @@ def recover_orphans() -> None: str(_task_value(task, "id")), reason="direct CLI lane restarted; provider session will resume", ) + BOARD_CORRUPTION_ERRORS.pop(board, None) + except Exception as error: + _record_board_access_error(board, error) finally: conn.close() @@ -923,7 +937,9 @@ def claim_ready(active: set[tuple[str, str]], limit: int) -> list[tuple[str, str continue try: kanban_db.recompute_ready(conn) - for task in kanban_db.list_tasks(conn): + tasks = kanban_db.list_tasks(conn) + BOARD_CORRUPTION_ERRORS.pop(board, None) + for task in tasks: task_id = str(_task_value(task, "id", "")) assignee = str(_task_value(task, "assignee", "") or "") if task_id and not assignee and str(_task_value(task, "status", "")) == "ready": @@ -950,6 +966,8 @@ def claim_ready(active: set[tuple[str, str]], limit: int) -> list[tuple[str, str claimed.append((board, task_id)) if len(claimed) >= limit: return claimed + except Exception as error: + _record_board_access_error(board, error) finally: conn.close() return claimed @@ -972,7 +990,13 @@ def main() -> int: print(f"worker future failed: {error}", file=sys.stderr, flush=True) del futures[future] active = set(futures.values()) - for board, task_id in claim_ready(active, workers - len(futures)): + try: + newly_claimed = claim_ready(active, workers - len(futures)) + BOARD_CORRUPTION_ERRORS.pop("board-registry", None) + except Exception as error: + _record_board_access_error("board-registry", error) + newly_claimed = [] + for board, task_id in newly_claimed: future = pool.submit(execute_claim, board, task_id) futures[future] = (board, task_id) time.sleep(5) diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index 164c04a0..eb118de6 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -453,7 +453,41 @@ def test_corrupt_board_is_quarantined_without_stopping_healthy_lanes(monkeypatch lanes.BOARD_CORRUPTION_ERRORS.clear() assert lanes.claim_ready(set(), 1) == [("healthy", "t_healthy")] - assert "quarantining corrupt Kanban board 'cassandra'" in capsys.readouterr().err + assert "temporarily skipping Kanban board 'cassandra'" in capsys.readouterr().err + + +def test_transient_board_scan_failure_does_not_stop_healthy_lanes(monkeypatch, capsys): + task = SimpleNamespace(id="t_healthy", assignee="cli-auto", status="ready") + + class Connection: + def __init__(self, board): + self.board = board + + def close(self): + return None + + def recompute_ready(connection): + if connection.board == "cassandra": + raise lanes.sqlite3.OperationalError("disk I/O error") + + fake_db = SimpleNamespace( + list_boards=lambda include_archived=False: [ + {"slug": "cassandra"}, + {"slug": "healthy"}, + ], + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(board), + recompute_ready=recompute_ready, + list_tasks=lambda _conn: [task], + claim_task=lambda _conn, _task_id, **_kwargs: task, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + lanes.BOARD_CORRUPTION_ERRORS.clear() + + assert lanes.claim_ready(set(), 1) == [("healthy", "t_healthy")] + error = capsys.readouterr().err + assert "temporarily skipping Kanban board 'cassandra'" in error + assert "storage OperationalError: disk I/O error" in error @pytest.mark.parametrize(