hermes: survive transient Kanban storage faults

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"

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)

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(