From 012e5fc2bacef0403d260c0d92ce01978292b422 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 04:49:02 -0300 Subject: [PATCH] hermes: recover stale kanban status pages --- services/hermes/agent-deployment.yaml | 2 +- services/hermes/kustomization.yaml | 1 + .../hermes/scripts/kanban_status_recovery.py | 286 ++++++++++++++++++ .../hermes/scripts/repair_cassandra_kanban.py | 37 ++- testing/tests/test_hermes_kanban_recovery.py | 162 ++++++++++ 5 files changed, 477 insertions(+), 11 deletions(-) create mode 100644 services/hermes/scripts/kanban_status_recovery.py diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index e6b3483a..535aadf6 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-activity-lineage-vault6" + ai.bstein.dev/config-rev: "20260816-kanban-recovery-v7" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/role: hermes-agent vault.hashicorp.com/agent-inject-secret-agent-api-key: kv/data/atlas/hermes/agent-tokens diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 4f29a2c7..94d04027 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -73,6 +73,7 @@ configMapGenerator: - hermes_stt_client.py=scripts/hermes_stt_client.py - image_broker.py=scripts/image_broker.py - install_agent_tools.sh=scripts/install_agent_tools.sh + - kanban_status_recovery.py=scripts/kanban_status_recovery.py - migrate_herdr_state.py=scripts/migrate_herdr_state.py - migrate_api_session_lineage.py=scripts/migrate_api_session_lineage.py - patch_api_server_sessions.py=scripts/patch_api_server_sessions.py diff --git a/services/hermes/scripts/kanban_status_recovery.py b/services/hermes/scripts/kanban_status_recovery.py new file mode 100644 index 00000000..b07a543f --- /dev/null +++ b/services/hermes/scripts/kanban_status_recovery.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +"""Evidence-gated reconstruction for stale Kanban task and run status pages.""" + +from __future__ import annotations + +import json +import re +import sqlite3 +from typing import Iterable + + +TASKS_PK_INDEX = "sqlite_autoindex_tasks_1" +TASK_STATUS_INDEXES = ("idx_tasks_status", "idx_tasks_assignee_status") +RUN_STATUS_INDEX = "idx_runs_status" + + +class RecoveryRefused(RuntimeError): + """The database failure is not the reviewed, loss-bounded corruption.""" + + +def _quote(identifier: str) -> str: + """Quote one SQLite identifier from the inspected local schema.""" + return '"' + identifier.replace('"', '""') + '"' + + +def _table_columns(connection: sqlite3.Connection, table: str) -> list[str]: + """Return columns in their persisted insertion order.""" + return [ + str(row[1]) for row in connection.execute(f"PRAGMA table_info({_quote(table)})") + ] + + +def known_task_status_alias_corruption( + connection: sqlite3.Connection, errors: Iterable[str] +) -> bool: + """Recognize the reviewed tasks-page alias with stale status rows.""" + row = connection.execute( + "SELECT rootpage FROM sqlite_master " + "WHERE type = 'table' AND name = 'tasks'" + ).fetchone() + if row is None: + return False + root_page = int(row[0]) + task_indexes = { + str(index[0]) + for index in connection.execute( + "SELECT name FROM sqlite_master " + "WHERE type = 'index' AND tbl_name = 'tasks'" + ) + } + allowed_indexes = task_indexes | {RUN_STATUS_INDEX} + tree_alias = re.compile( + rf"Tree {root_page} page \d+ cell \d+: 2nd reference to page \d+" + ) + rowid_order = re.compile( + rf"Tree {root_page} page \d+ cell \d+: Rowid \d+ out of order" + ) + wrong_entries = re.compile(r"wrong # of entries in index ([A-Za-z0-9_]+)") + missing_entry = re.compile(r"row \d+ missing from index ([A-Za-z0-9_]+)") + lines = [ + line + for error in errors + for line in str(error).splitlines() + if line and line != "*** in database main ***" + ] + if not lines: + return False + saw_alias = False + saw_order = False + saw_task_index = False + saw_task_status = False + saw_run_status = False + for line in lines: + if tree_alias.fullmatch(line): + saw_alias = True + continue + if rowid_order.fullmatch(line): + saw_order = True + continue + match = wrong_entries.fullmatch(line) or missing_entry.fullmatch(line) + if match is None or match.group(1) not in allowed_indexes: + return False + index = match.group(1) + saw_task_index = saw_task_index or index in task_indexes + saw_task_status = saw_task_status or index in TASK_STATUS_INDEXES + saw_run_status = saw_run_status or index == RUN_STATUS_INDEX + return all( + (saw_alias, saw_order, saw_task_index, saw_task_status, saw_run_status) + ) + + +def _indexed_values( + connection: sqlite3.Connection, + table: str, + index: str, + columns: tuple[str, ...], +) -> dict[int, tuple[object, ...]]: + """Read one value per rowid through a named, independently checked index.""" + projection = ", ".join(_quote(column) for column in columns) + values: dict[int, tuple[object, ...]] = {} + for row in connection.execute( + f"SELECT rowid, {projection} FROM {_quote(table)} " + f"INDEXED BY {_quote(index)} ORDER BY rowid" + ): + rowid = int(row[0]) + if rowid in values: + raise RecoveryRefused(f"duplicate rowid in trusted index {index}") + values[rowid] = tuple(row[1:]) + return values + + +def _completed_event( + connection: sqlite3.Connection, task_id: str, run_id: int +) -> tuple[int, str]: + """Return the matching completion timestamp and retained summary evidence.""" + row = connection.execute( + "SELECT created_at, payload FROM task_events " + "WHERE task_id = ? AND run_id = ? AND kind = 'completed' " + "ORDER BY id DESC LIMIT 1", + (task_id, run_id), + ).fetchone() + if row is None: + raise RecoveryRefused( + f"status disagreement for {task_id} lacks a completed event" + ) + try: + payload = json.loads(str(row[1])) + except (TypeError, ValueError) as error: + raise RecoveryRefused( + f"completed event for {task_id} has invalid payload" + ) from error + summary = payload.get("summary") + if not isinstance(summary, str) or not summary.strip(): + raise RecoveryRefused( + f"completed event for {task_id} lacks retained summary evidence" + ) + return int(row[0]), summary + + +def _task_rows( + connection: sqlite3.Connection, task_columns: list[str] +) -> dict[int, list[object]]: + """Read the unique task row set through its primary-key index.""" + projection = ", ".join(_quote(column) for column in task_columns) + rows: dict[int, list[object]] = {} + task_ids: set[str] = set() + for row in connection.execute( + f"SELECT rowid, {projection} FROM tasks " + f"INDEXED BY {_quote(TASKS_PK_INDEX)} ORDER BY id" + ): + rowid = int(row[0]) + values = list(row[1:]) + task_id = str(values[task_columns.index("id")]) + if rowid in rows or task_id in task_ids: + raise RecoveryRefused("trusted tasks primary index is not unique") + rows[rowid] = values + task_ids.add(task_id) + return rows + + +def _reconcile_tasks( + connection: sqlite3.Connection, + task_columns: list[str], +) -> tuple[dict[int, list[object]], dict[int, tuple[int, str]]]: + """Reconcile stale task values when two status indexes and an event agree.""" + task_rows = _task_rows(connection, task_columns) + status_maps = [ + _indexed_values(connection, "tasks", index, ("status",)) + for index in TASK_STATUS_INDEXES + ] + if status_maps[0] != status_maps[1]: + raise RecoveryRefused("independent task status indexes disagree") + if set(task_rows) != set(status_maps[0]): + raise RecoveryRefused("task primary and status indexes disagree on row set") + status_column = task_columns.index("status") + task_id_column = task_columns.index("id") + run_column = task_columns.index("current_run_id") + completed_evidence: dict[int, tuple[int, str]] = {} + for rowid, values in task_rows.items(): + indexed_status = str(status_maps[0][rowid][0]) + if values[status_column] == indexed_status: + continue + if indexed_status != "done" or values[run_column] is None: + raise RecoveryRefused("task status disagreement is not a completed run") + task_id = str(values[task_id_column]) + run_id = int(values[run_column]) + completed_at, summary = _completed_event(connection, task_id, run_id) + completed_evidence[run_id] = (completed_at, summary) + replacements: dict[str, object] = { + "status": "done", + "result": summary, + "completed_at": completed_at, + "claim_lock": None, + "claim_expires": None, + "worker_pid": None, + "current_run_id": None, + "block_kind": None, + "block_recurrences": 0, + } + for column, replacement in replacements.items(): + values[task_columns.index(column)] = replacement + if not completed_evidence: + raise RecoveryRefused("no evidence-backed task status disagreement was found") + return task_rows, completed_evidence + + +def _reconcile_runs( + connection: sqlite3.Connection, + run_columns: list[str], + completed_evidence: dict[int, tuple[int, str]], +) -> dict[int, list[object]]: + """Reconcile stale run values against its index and completed event.""" + projection = ", ".join(_quote(column) for column in run_columns) + run_rows = { + int(row[0]): list(row[1:]) + for row in connection.execute( + f"SELECT rowid, {projection} FROM task_runs NOT INDEXED ORDER BY rowid" + ) + } + run_statuses = _indexed_values( + connection, "task_runs", RUN_STATUS_INDEX, ("status",) + ) + if set(run_rows) != set(run_statuses): + raise RecoveryRefused("run table and status index disagree on row set") + status_column = run_columns.index("status") + run_id_column = run_columns.index("id") + task_column = run_columns.index("task_id") + reconciled_runs: set[int] = set() + for rowid, values in run_rows.items(): + indexed_status = str(run_statuses[rowid][0]) + if values[status_column] == indexed_status: + continue + run_id = int(values[run_id_column]) + if run_id != rowid or indexed_status != "done": + raise RecoveryRefused("run status disagreement is not a completed run") + evidence = completed_evidence.get(run_id) + if evidence is None: + raise RecoveryRefused("run status disagreement lacks matching task evidence") + completed_at, summary = evidence + task_id = str(values[task_column]) + if _completed_event(connection, task_id, run_id) != evidence: + raise RecoveryRefused("task and run completion evidence disagree") + replacements = { + "status": "done", + "outcome": "completed", + "ended_at": completed_at, + "claim_lock": None, + "claim_expires": None, + "worker_pid": None, + "summary": summary, + "metadata": None, + "error": None, + } + for column, replacement in replacements.items(): + values[run_columns.index(column)] = replacement + reconciled_runs.add(run_id) + if reconciled_runs != set(completed_evidence): + raise RecoveryRefused("completed task and run disagreements do not match") + return run_rows + + +def reconciled_task_status_rows( + connection: sqlite3.Connection, +) -> dict[str, list[tuple[object, ...]]]: + """Reconstruct reviewed stale task/run pages from independent evidence.""" + task_columns = _table_columns(connection, "tasks") + run_columns = _table_columns(connection, "task_runs") + required_task_columns = { + "id", "status", "result", "completed_at", "claim_lock", + "claim_expires", "worker_pid", "current_run_id", "block_kind", + "block_recurrences", + } + required_run_columns = { + "id", "task_id", "status", "outcome", "ended_at", "claim_lock", + "claim_expires", "worker_pid", "summary", "metadata", "error", + } + if not required_task_columns <= set(task_columns): + raise RecoveryRefused("tasks schema lacks status-recovery columns") + if not required_run_columns <= set(run_columns): + raise RecoveryRefused("task_runs schema lacks status-recovery columns") + task_rows, completed_evidence = _reconcile_tasks(connection, task_columns) + run_rows = _reconcile_runs(connection, run_columns, completed_evidence) + return { + "tasks": [tuple(task_rows[rowid]) for rowid in sorted(task_rows)], + "task_runs": [tuple(run_rows[rowid]) for rowid in sorted(run_rows)], + } diff --git a/services/hermes/scripts/repair_cassandra_kanban.py b/services/hermes/scripts/repair_cassandra_kanban.py index ec646092..e5a66d9a 100644 --- a/services/hermes/scripts/repair_cassandra_kanban.py +++ b/services/hermes/scripts/repair_cassandra_kanban.py @@ -17,6 +17,12 @@ from itertools import islice from pathlib import Path from typing import Iterable +from kanban_status_recovery import ( + RecoveryRefused, + known_task_status_alias_corruption, + reconciled_task_status_rows, +) + DEFAULT_DATABASE = Path("/opt/data/kanban/boards/cassandra/kanban.db") EXPECTED_TABLES = { @@ -41,10 +47,6 @@ COMMENTS_INDEX = "idx_comments_task" REBUILDABLE_INDEXES = {"idx_events_run", "idx_events_task"} -class RecoveryRefused(RuntimeError): - """The database failure is not the reviewed, loss-bounded corruption.""" - - def _quote(identifier: str) -> str: """Quote one SQLite identifier from the inspected local schema.""" return '"' + identifier.replace('"', '""') + '"' @@ -153,10 +155,12 @@ def _table_columns(connection: sqlite3.Connection, table: str) -> list[str]: ] + def _copy_table( source: sqlite3.Connection, destination: sqlite3.Connection, table: str, + reconciled_rows: dict[str, list[tuple[object, ...]]] | None = None, ) -> int: """Copy one table, using the intact comments index as the trusted row set.""" columns = _table_columns(source, table) @@ -165,7 +169,9 @@ def _copy_table( projection = ", ".join(_quote(column) for column in columns) placeholders = ", ".join("?" for _ in columns) insert = f"INSERT INTO {_quote(table)} ({projection}) VALUES ({placeholders})" - if table == "task_comments": + if reconciled_rows and table in reconciled_rows: + rows = iter(reconciled_rows[table]) + elif table == "task_comments": # The intact index is the authority for which comments existed, but a # covering lookup follows the damaged table page and can abort. Walk # the table once and retain only rowids present in that intact index. @@ -196,7 +202,9 @@ def _copy_table( def _populate_replacement( - source: sqlite3.Connection, replacement: Path + source: sqlite3.Connection, + replacement: Path, + reconciled_rows: dict[str, list[tuple[object, ...]]] | None = None, ) -> dict[str, int]: """Build and verify a clean database using only reviewed source rows.""" entries = _schema_entries(source) @@ -210,7 +218,9 @@ def _populate_replacement( for _, _, _, statement in tables: destination.execute(statement) for table in COPY_ORDER: - counts[table] = _copy_table(source, destination, table) + counts[table] = _copy_table( + source, destination, table, reconciled_rows + ) if source.execute( "SELECT 1 FROM sqlite_master WHERE name = 'sqlite_sequence'" ).fetchone(): @@ -266,10 +276,15 @@ 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 ( + reconciled_rows = None + known_corruption = ( _known_comments_alias_corruption(source, errors) or _known_index_only_corruption(errors) - ): + ) + if known_task_status_alias_corruption(source, errors): + reconciled_rows = reconciled_task_status_rows(source) + known_corruption = True + if not known_corruption: raise RecoveryRefused( "integrity failure does not match the reviewed corruption" ) @@ -280,7 +295,9 @@ def recover_database(database: Path, errors: list[str]) -> dict[str, object]: replacement = Path(temporary_name) replacement.unlink() try: - counts = _populate_replacement(source, replacement) + counts = _populate_replacement( + source, replacement, reconciled_rows + ) except Exception: replacement.unlink(missing_ok=True) raise diff --git a/testing/tests/test_hermes_kanban_recovery.py b/testing/tests/test_hermes_kanban_recovery.py index 4ace7857..d0c1b536 100644 --- a/testing/tests/test_hermes_kanban_recovery.py +++ b/testing/tests/test_hermes_kanban_recovery.py @@ -4,6 +4,7 @@ from __future__ import annotations import importlib.util import sqlite3 +import sys from pathlib import Path import pytest @@ -11,6 +12,8 @@ import pytest ROOT = Path(__file__).resolve().parents[2] SCRIPT = ROOT / "services/hermes/scripts/repair_cassandra_kanban.py" +sys.path.insert(0, str(SCRIPT.parent)) +status_recovery = importlib.import_module("kanban_status_recovery") SPEC = importlib.util.spec_from_file_location("repair_cassandra_kanban", SCRIPT) assert SPEC and SPEC.loader recovery = importlib.util.module_from_spec(SPEC) @@ -77,6 +80,89 @@ def _known_errors(database: Path) -> list[str]: ] +def _task_status_alias_database(path: Path) -> Path: + """Create the production-shaped rows needed for status reconciliation.""" + connection = sqlite3.connect(path) + connection.executescript( + """ + CREATE TABLE tasks ( + id TEXT PRIMARY KEY, title TEXT NOT NULL, assignee TEXT, + status TEXT NOT NULL, result TEXT, completed_at INTEGER, + claim_lock TEXT, claim_expires INTEGER, worker_pid INTEGER, + current_run_id INTEGER, block_kind TEXT, + block_recurrences INTEGER NOT NULL DEFAULT 0, + tenant TEXT, idempotency_key TEXT, session_id TEXT + ); + CREATE TABLE task_links ( + parent_id TEXT NOT NULL, child_id TEXT NOT NULL, + PRIMARY KEY (parent_id, child_id) + ); + CREATE TABLE task_comments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, + author TEXT NOT NULL, body TEXT NOT NULL, created_at INTEGER NOT NULL + ); + CREATE TABLE task_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, + run_id INTEGER, kind TEXT NOT NULL, payload TEXT, created_at INTEGER NOT NULL + ); + CREATE TABLE task_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, + status TEXT NOT NULL, outcome TEXT, ended_at INTEGER, + claim_lock TEXT, claim_expires INTEGER, worker_pid INTEGER, + summary TEXT, metadata TEXT, error TEXT, started_at INTEGER NOT NULL + ); + CREATE TABLE task_attachments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL + ); + CREATE TABLE kanban_notify_subs ( + task_id TEXT NOT NULL, platform TEXT NOT NULL, + PRIMARY KEY (task_id, platform) + ); + CREATE INDEX idx_comments_task ON task_comments(task_id, created_at); + CREATE INDEX idx_tasks_status ON tasks(status); + CREATE INDEX idx_tasks_assignee_status ON tasks(assignee, status); + CREATE INDEX idx_tasks_idempotency ON tasks(idempotency_key); + CREATE INDEX idx_tasks_session_id ON tasks(session_id); + CREATE INDEX idx_tasks_tenant ON tasks(tenant); + CREATE INDEX idx_runs_status ON task_runs(status); + CREATE INDEX idx_runs_task ON task_runs(task_id, started_at); + INSERT INTO tasks( + id, title, assignee, status, claim_lock, claim_expires, + current_run_id, block_kind, block_recurrences + ) VALUES ( + 'task-1', 'Recover completion', 'worker', 'running', + 'stale-claim', 99, 1, 'capability', 2 + ); + INSERT INTO task_runs( + task_id, status, claim_lock, claim_expires, started_at + ) VALUES ('task-1', 'running', 'stale-claim', 99, 1); + INSERT INTO task_events(task_id, run_id, kind, payload, created_at) + VALUES ( + 'task-1', 1, 'completed', + '{"result_len": 18, "summary": "verified completion"}', 42 + ); + """ + ) + connection.commit() + connection.close() + return path + + +def _task_status_alias_errors(database: Path) -> list[str]: + with sqlite3.connect(database) as connection: + root = connection.execute( + "SELECT rootpage FROM sqlite_master WHERE name = 'tasks'" + ).fetchone()[0] + return [ + "*** in database main ***\n" + f"Tree {root} page 16 cell 1: Rowid 1 out of order\n" + f"Tree {root} page 16 cell 1: 2nd reference to page 17", + "wrong # of entries in index idx_tasks_status", + "row 1 missing from index idx_tasks_assignee_status", + "row 1 missing from index idx_runs_status", + ] + + def test_healthy_board_is_untouched(tmp_path: Path) -> None: database = _database(tmp_path / "kanban.db") before = database.read_bytes() @@ -142,6 +228,82 @@ def test_known_event_index_mismatch_rebuilds_rows(tmp_path: Path) -> None: ) +def test_known_task_status_alias_uses_independent_completion_evidence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + database = _task_status_alias_database(tmp_path / "kanban.db") + original_indexed_values = status_recovery._indexed_values + + def completed_index_values(connection, table, index, columns): + values = original_indexed_values(connection, table, index, columns) + if columns == ("status",) and index in { + *status_recovery.TASK_STATUS_INDEXES, + status_recovery.RUN_STATUS_INDEX, + }: + values[1] = ("done",) + return values + + monkeypatch.setattr( + status_recovery, "_indexed_values", completed_index_values + ) + + result = recovery.recover_database( + database, _task_status_alias_errors(database) + ) + + assert result["state"] == "recovered" + with sqlite3.connect(database) as connection: + assert connection.execute("PRAGMA integrity_check").fetchone() == ("ok",) + assert connection.execute( + "SELECT status, result, completed_at, claim_lock, current_run_id, " + "block_kind, block_recurrences FROM tasks WHERE id = 'task-1'" + ).fetchone() == ( + "done", + "verified completion", + 42, + None, + None, + None, + 0, + ) + assert connection.execute( + "SELECT status, outcome, ended_at, claim_lock, summary " + "FROM task_runs WHERE id = 1" + ).fetchone() == ( + "done", + "completed", + 42, + None, + "verified completion", + ) + + +def test_task_status_alias_refuses_disagreeing_status_indexes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + database = _task_status_alias_database(tmp_path / "kanban.db") + original_indexed_values = status_recovery._indexed_values + + def disagreeing_index_values(connection, table, index, columns): + values = original_indexed_values(connection, table, index, columns) + if ( + columns == ("status",) + and index == status_recovery.TASK_STATUS_INDEXES[0] + ): + values[1] = ("done",) + return values + + monkeypatch.setattr( + status_recovery, "_indexed_values", disagreeing_index_values + ) + + with pytest.raises( + recovery.RecoveryRefused, + match="independent task status indexes disagree", + ): + recovery.recover_database(database, _task_status_alias_errors(database)) + + @pytest.mark.parametrize( "errors", [