hermes: recover stale kanban status pages

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

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

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)],
+    }

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

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",
     [