hermes: recover Kanban event indexes

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-model-resolution"
+        ai.bstein.dev/config-rev: "20260815-kanban-index-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

services/hermes/scripts/repair_cassandra_kanban.py

@@ -38,6 +38,7 @@ COPY_ORDER = (
     "kanban_notify_subs",
 )
 COMMENTS_INDEX = "idx_comments_task"
+REBUILDABLE_INDEXES = {"idx_events_run", "idx_events_task"}
 
 
 class RecoveryRefused(RuntimeError):
@@ -109,6 +110,26 @@ def _known_comments_alias_corruption(
     )
 
 
+def _known_index_only_corruption(errors: Iterable[str]) -> bool:
+    """Recognize stale entries confined to the reviewed event indexes."""
+    lines = [
+        line
+        for error in errors
+        for line in str(error).splitlines()
+        if line and line != "*** in database main ***"
+    ]
+    if not lines:
+        return False
+    pattern = re.compile(r"wrong # of entries in index ([A-Za-z0-9_]+)")
+    indexes: set[str] = set()
+    for line in lines:
+        match = pattern.fullmatch(line)
+        if match is None:
+            return False
+        indexes.add(match.group(1))
+    return bool(indexes) and indexes <= REBUILDABLE_INDEXES
+
+
 def _schema_entries(
     connection: sqlite3.Connection,
 ) -> list[tuple[str, str, str, str]]:
@@ -245,7 +266,10 @@ def recover_database(database: Path, errors: list[str]) -> dict[str, object]:
         ).fetchone()
         if index != ("task_comments",):
             raise RecoveryRefused("trusted task-comments index is unavailable")
-        if not _known_comments_alias_corruption(source, errors):
+        if not (
+            _known_comments_alias_corruption(source, errors)
+            or _known_index_only_corruption(errors)
+        ):
             raise RecoveryRefused(
                 "integrity failure does not match the reviewed corruption"
             )

testing/tests/test_hermes_kanban_recovery.py

@@ -34,7 +34,8 @@ def _database(path: Path) -> Path:
           created_at INTEGER NOT NULL
         );
         CREATE TABLE task_events (
-          id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL
+          id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL,
+          run_id INTEGER, created_at INTEGER NOT NULL DEFAULT 1
         );
         CREATE TABLE task_runs (
           id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL
@@ -47,6 +48,8 @@ def _database(path: Path) -> Path:
           PRIMARY KEY (task_id, platform)
         );
         CREATE INDEX idx_comments_task ON task_comments(task_id, created_at);
+        CREATE INDEX idx_events_task ON task_events(task_id, created_at);
+        CREATE INDEX idx_events_run ON task_events(run_id, id);
         INSERT INTO tasks VALUES ('task-1', 'Preserve me');
         INSERT INTO task_comments(task_id, author, body, created_at)
           VALUES ('task-1', 'worker', 'evidence', 1);
@@ -116,3 +119,43 @@ def test_unknown_corruption_is_preserved_and_refused(tmp_path: Path) -> None:
 
     assert database.read_bytes() == before
     assert not list(tmp_path.glob("*.corrupt.*"))
+
+
+def test_known_event_index_mismatch_rebuilds_rows(tmp_path: Path) -> None:
+    database = _database(tmp_path / "kanban.db")
+    before = database.read_bytes()
+
+    result = recovery.recover_database(
+        database,
+        [
+            "wrong # of entries in index idx_events_task",
+            "wrong # of entries in index idx_events_run",
+        ],
+    )
+
+    assert result["state"] == "recovered"
+    assert Path(result["backup"]).read_bytes() == before
+    with sqlite3.connect(database) as connection:
+        assert connection.execute("PRAGMA integrity_check").fetchone() == ("ok",)
+        assert connection.execute("SELECT task_id FROM task_events").fetchone() == (
+            "task-1",
+        )
+
+
+@pytest.mark.parametrize(
+    "errors",
+    [
+        ["wrong # of entries in index idx_comments_task"],
+        [
+            "wrong # of entries in index idx_events_task",
+            "freelist leaf count is too big",
+        ],
+    ],
+)
+def test_unreviewed_index_failures_remain_refused(
+    tmp_path: Path, errors: list[str]
+) -> None:
+    database = _database(tmp_path / "kanban.db")
+
+    with pytest.raises(recovery.RecoveryRefused, match="does not match"):
+        recovery.recover_database(database, errors)