diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 30e4d7c6..33feb7a1 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-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 diff --git a/services/hermes/scripts/repair_cassandra_kanban.py b/services/hermes/scripts/repair_cassandra_kanban.py index 79e97e5d..ec646092 100644 --- a/services/hermes/scripts/repair_cassandra_kanban.py +++ b/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" ) diff --git a/testing/tests/test_hermes_kanban_recovery.py b/testing/tests/test_hermes_kanban_recovery.py index 09dbe8e1..4ace7857 100644 --- a/testing/tests/test_hermes_kanban_recovery.py +++ b/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)