From 87520c6fa3507f90ddd7b6afed31a22e5d475fa3 Mon Sep 17 00:00:00 2001 From: Ryan Didur <224447970+ryandidurlabs@users.noreply.github.com> Date: Sat, 23 May 2026 06:53:16 -0700 Subject: [PATCH 1/2] fix: harden kanban sqlite corruption handling Add dispatcher lease coordination, corrupt DB detection, safer reconciliation, and Kanban regression coverage.\n\nTask: t_a0e3aea4 --- gateway/run.py | 241 ++++++-- hermes_cli/config.py | 16 +- hermes_cli/kanban.py | 182 +++++- hermes_cli/kanban_db.py | 565 ++++++++++++++++-- tests/hermes_cli/test_kanban_cli.py | 135 +++++ .../test_kanban_core_functionality.py | 299 ++++++++- tests/hermes_cli/test_kanban_db.py | 367 +++++++++++- 7 files changed, 1673 insertions(+), 132 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 9ca87452f978..caf72f13cad0 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -29,6 +29,7 @@ import inspect import json import logging +import math import os import re import shlex @@ -67,8 +68,110 @@ _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT = 5.0 _TELEGRAM_COMMAND_MENTION_RE = re.compile(r"(? Optional[int]: + """Return a safe live worker cap from ``kanban.max_spawn``. + + ``None`` means explicitly unlimited. Invalid, zero, or negative values + fall back to the provided safe default rather than silently disabling the + anti-stampede cap. + """ + if raw_value is None: + return None + try: + parsed = int(raw_value) + except (TypeError, ValueError, OverflowError): + logger.warning( + "kanban dispatcher: invalid kanban.max_spawn=%r; using default %r", + raw_value, + default, + ) + return default + if parsed < 1: + logger.warning( + "kanban dispatcher: kanban.max_spawn=%r is below 1; using default %r", + raw_value, + default, + ) + return default + return parsed + + +def _parse_kanban_failure_limit(raw_value: Any, default: int) -> int: + """Return a safe circuit-breaker threshold from ``kanban.failure_limit``.""" + try: + parsed = int(raw_value) + except (TypeError, ValueError, OverflowError): + logger.warning( + "kanban dispatcher: invalid kanban.failure_limit=%r; using default %d", + raw_value, + default, + ) + return default + if parsed < 1: + logger.warning( + "kanban dispatcher: kanban.failure_limit=%r is below 1; using default %d", + raw_value, + default, + ) + return default + return parsed + + +def _parse_positive_int_or_default(raw_value: Any, default: int, *, label: str) -> int: + """Parse a positive integer config value with a warning fallback.""" + try: + parsed = int(raw_value) + except (TypeError, ValueError, OverflowError): + logger.warning("kanban dispatcher: invalid %s=%r; using default %d", label, raw_value, default) + return default + if parsed < 1: + logger.warning("kanban dispatcher: %s=%r is below 1; using default %d", label, raw_value, default) + return default + return parsed + + +def _parse_positive_float_or_default(raw_value: Any, default: float, *, label: str) -> float: + """Parse a positive numeric config value with a warning fallback.""" + try: + parsed = float(raw_value) + except (TypeError, ValueError, OverflowError): + logger.warning("kanban dispatcher: invalid %s=%r; using default %s", label, raw_value, default) + return default + if not math.isfinite(parsed): + logger.warning("kanban dispatcher: %s=%r is not finite; using default %s", label, raw_value, default) + return default + if parsed < 1.0: + logger.warning("kanban dispatcher: %s=%r is below 1; using default %s", label, raw_value, default) + return default + return parsed + + +def _is_corrupt_kanban_board_db_error(exc: Exception) -> bool: + """Return whether ``exc`` means a Kanban board DB should be quarantined. + + The DB layer performs lightweight SQLite header validation before connect. + Invalid headers, malformed files, and truncated files are all persistent + board-file problems rather than transient dispatcher tick failures, so the + gateway should disable dispatch for that board fingerprint until the file + changes or the process restarts. + """ + if not isinstance(exc, sqlite3.DatabaseError): + return False + msg = str(exc).lower() + return ( + "file is not a database" in msg + or "database disk image is malformed" in msg + or "truncated sqlite file" in msg + ) + + +_GATEWAY_NOISY_STATUS_RE = re.compile( + r"(" # transient/auxiliary status that should stay in logs, not gateway chats r"auxiliary\s+.+\s+failed" r"|compression\s+summary\s+failed" r"|fallback\s+context\s+marker" @@ -76,6 +179,8 @@ r"|no\s+auxiliary\s+llm\s+provider\s+configured" r"|auto-lowered\s+compression\s+threshold" r"|preflight\s+compression" + r"|compacting\s+context" + r"|summarizing\s+earlier\s+conversation" r"|rate\s+limited\.\s+waiting\s+\d" r"|retrying\s+in\s+\d" r"|max\s+retries\s+\(\d+\).*(?:trying\s+fallback|exhausted|invalid\s+responses)" @@ -232,7 +337,7 @@ def _prepare_gateway_status_message(platform: Any, event_type: str, message: str return text text = _redact_gateway_user_facing_secrets(text) - if _TELEGRAM_NOISY_STATUS_RE.search(text): + if _GATEWAY_NOISY_STATUS_RE.search(text): return None if _looks_like_gateway_provider_error(text): return _gateway_provider_error_reply(text) @@ -5128,13 +5233,24 @@ async def _kanban_dispatcher_watcher(self) -> None: logger.warning("kanban dispatcher: kanban_db not importable; dispatcher disabled") return - interval = float(kanban_cfg.get("dispatch_interval_seconds", 60) or 60) - interval = max(interval, 1.0) # sanity floor — tighter than this is a footgun + interval = _parse_positive_float_or_default( + kanban_cfg.get( + "dispatch_interval_seconds", + _KANBAN_DEFAULT_DISPATCH_INTERVAL_SECONDS, + ), + _KANBAN_DEFAULT_DISPATCH_INTERVAL_SECONDS, + label="kanban.dispatch_interval_seconds", + ) - # Read max_spawn config to limit concurrent kanban tasks - max_spawn = kanban_cfg.get("max_spawn", None) + # Read max_spawn config to limit concurrent kanban tasks. + # Keep parsing defensive: a malformed config should not kill the + # gateway-hosted dispatcher loop or disable the anti-stampede cap. + max_spawn = _parse_kanban_max_spawn( + kanban_cfg.get("max_spawn", _KANBAN_DEFAULT_MAX_SPAWN), + default=_KANBAN_DEFAULT_MAX_SPAWN, + ) if max_spawn is not None: - logger.info(f"kanban dispatcher: max_spawn={max_spawn}") + logger.info("kanban dispatcher: max_spawn=%s", max_spawn) # Cap the number of simultaneously running tasks so slow workers # (local LLMs, resource-constrained hosts) don't pile up and time @@ -5145,7 +5261,7 @@ async def _kanban_dispatcher_watcher(self) -> None: if raw_max_in_progress is not None: try: max_in_progress = int(raw_max_in_progress) - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError): logger.warning( "kanban dispatcher: invalid kanban.max_in_progress=%r; ignoring", raw_max_in_progress, @@ -5162,28 +5278,26 @@ async def _kanban_dispatcher_watcher(self) -> None: logger.info(f"kanban dispatcher: max_in_progress={max_in_progress}") raw_failure_limit = kanban_cfg.get("failure_limit", _kb.DEFAULT_FAILURE_LIMIT) - try: - failure_limit = int(raw_failure_limit) - except (TypeError, ValueError): - logger.warning( - "kanban dispatcher: invalid kanban.failure_limit=%r; using default %d", - raw_failure_limit, - _kb.DEFAULT_FAILURE_LIMIT, - ) - failure_limit = _kb.DEFAULT_FAILURE_LIMIT - if failure_limit < 1: - logger.warning( - "kanban dispatcher: kanban.failure_limit=%r is below 1; using default %d", - raw_failure_limit, - _kb.DEFAULT_FAILURE_LIMIT, - ) - failure_limit = _kb.DEFAULT_FAILURE_LIMIT + failure_limit = _parse_kanban_failure_limit( + raw_failure_limit, + _kb.DEFAULT_FAILURE_LIMIT, + ) + + # A short DB-backed lease keeps multiple profile gateways from all + # running dispatcher ticks against the same shared board. The TTL must + # exceed one tick so the current owner can renew before failover. + dispatcher_owner = f"gateway:{_kb._claimer_id()}" + dispatcher_lease_ttl_seconds = _parse_positive_int_or_default( + kanban_cfg.get("dispatcher_lease_ttl_seconds", int(max(interval * 2, 30))), + int(max(interval * 2, 30)), + label="kanban.dispatcher_lease_ttl_seconds", + ) # Read stale_timeout_seconds — 0 disables stale detection. raw_stale = kanban_cfg.get("dispatch_stale_timeout_seconds", 0) try: stale_timeout_seconds = int(raw_stale or 0) - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError): logger.warning( "kanban dispatcher: invalid kanban.dispatch_stale_timeout_seconds=%r; " "disabling stale detection", @@ -5216,15 +5330,6 @@ def _board_db_fingerprint(slug: str) -> tuple[str, int | None, int | None]: return (resolved, None, None) return (resolved, stat.st_mtime_ns, stat.st_size) - def _is_corrupt_board_db_error(exc: Exception) -> bool: - if not isinstance(exc, sqlite3.DatabaseError): - return False - msg = str(exc).lower() - return ( - "file is not a database" in msg - or "database disk image is malformed" in msg - ) - def _tick_once_for_board(slug: str) -> "Optional[object]": """Run one dispatch_once for a specific board. @@ -5253,6 +5358,17 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": # re-ran the migration on a second connection, racing # the first. See the matching comment in # `_kanban_notifier_watcher` and issue #21378. + if not _kb.acquire_dispatcher_lease( + conn, + board=slug, + owner=dispatcher_owner, + ttl_seconds=dispatcher_lease_ttl_seconds, + ): + logger.debug( + "kanban dispatcher: board %s lease held by another gateway; skipping tick", + slug, + ) + return None return _kb.dispatch_once( conn, board=slug, @@ -5262,7 +5378,7 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": stale_timeout_seconds=stale_timeout_seconds, ) except sqlite3.DatabaseError as exc: - if _is_corrupt_board_db_error(exc): + if _is_corrupt_kanban_board_db_error(exc): disabled_corrupt_boards[slug] = fingerprint logger.error( "kanban dispatcher: board %s database %s is not a valid " @@ -5324,6 +5440,12 @@ def _ready_nonempty() -> bool: conn = None try: conn = _kb.connect(board=slug) + if not _kb.owns_dispatcher_lease( + conn, + board=slug, + owner=dispatcher_owner, + ): + continue if _kb.has_spawnable_ready(conn): return True if _kb.has_spawnable_review(conn): @@ -5349,7 +5471,7 @@ def _ready_nonempty() -> bool: auto_decompose_per_tick = int( kanban_cfg.get("auto_decompose_per_tick", 3) or 3 ) - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError): auto_decompose_per_tick = 3 if auto_decompose_per_tick < 1: auto_decompose_per_tick = 1 @@ -5376,6 +5498,51 @@ def _auto_decompose_tick() -> int: slug = b.get("slug") or _kb.DEFAULT_BOARD if attempted >= auto_decompose_per_tick: break + fingerprint = _board_db_fingerprint(slug) + disabled_fingerprint = disabled_corrupt_boards.get(slug) + if disabled_fingerprint == fingerprint: + continue + if disabled_fingerprint is not None: + disabled_corrupt_boards.pop(slug, None) + lease_conn = None + try: + lease_conn = _kb.connect(board=slug) + if not _kb.acquire_dispatcher_lease( + lease_conn, + board=slug, + owner=dispatcher_owner, + ttl_seconds=dispatcher_lease_ttl_seconds, + ): + logger.debug( + "kanban auto-decompose: board %s lease held by another gateway; skipping", + slug, + ) + continue + except sqlite3.DatabaseError as exc: + if _is_corrupt_kanban_board_db_error(exc): + disabled_corrupt_boards[slug] = fingerprint + logger.error( + "kanban dispatcher: board %s database %s is not a valid " + "SQLite database; disabling dispatch for this board " + "until the file changes or the gateway restarts. Move " + "or restore the file, then run `hermes kanban init` if " + "you need a fresh board.", + slug, + fingerprint[0], + ) + else: + logger.debug( + "kanban auto-decompose: lease check failed on board %s (%s)", + slug, + exc, + ) + continue + finally: + if lease_conn is not None: + try: + lease_conn.close() + except Exception: + pass # Pin this board for the duration of the call — same # pattern as the dashboard specify endpoint. The # decomposer module connects with no board kwarg and diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 715fd7eb76ff..856abcd2f43d 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1512,10 +1512,11 @@ def _ensure_hermes_home_managed(home: Path): # Kanban multi-agent coordination — controls the dispatcher loop that # spawns workers for ready tasks. The dispatcher ticks every N seconds - # (default 60), reclaims stale claims, promotes dependency-satisfied + # (default 120), reclaims stale claims, promotes dependency-satisfied # todos to ready, and fires `hermes -p chat -q ...` for # each claimable ready task. One dispatcher per profile is sufficient; - # running more than one on the same kanban.db will race for claims. + # running more than one on the same kanban.db is coordinated by a + # short dispatcher lease so workers do not stampede. "kanban": { # Run the dispatcher inside the gateway process. On by default — # the cost is ~300µs every `dispatch_interval_seconds` when idle, @@ -1524,12 +1525,17 @@ def _ensure_hermes_home_managed(home: Path): # don't want the gateway to spawn workers. "dispatch_in_gateway": True, # Seconds between dispatcher ticks (idle or not). Lower = snappier - # pickup of newly-ready tasks; higher = less SQL pressure. - "dispatch_interval_seconds": 60, + # pickup of newly-ready tasks; higher = less SQL pressure and gentler + # retry cadence when workers crash repeatedly. + "dispatch_interval_seconds": 120, + # Live worker concurrency cap across the board. Keeps a large + # crash-only retry wave from stampeding the host after an operator + # unblocks/requeues many tasks at once. None means unlimited. + "max_spawn": 3, # Auto-block after this many consecutive non-success attempts for the # same task/profile (spawn_failed, timed_out, or crashed). Reassignment # resets the streak for the new profile. - "failure_limit": 2, + "failure_limit": 4, # Worker stdout/stderr logs rotate at spawn time. Defaults preserve # the historical 2 MiB + one-backup behavior; long-running workers can # raise these to keep more early failure evidence. diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 4e975bb3e8d7..70d9741240fd 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -477,6 +477,37 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu help="Emit JSON (structured) instead of the default human table", ) + # --- reconcile-completions --- + p_reconcile = sub.add_parser( + "reconcile-completions", + help="Dry-run/apply completed-work reconciliation after malformed DB incidents", + ) + p_reconcile.add_argument( + "task_ids", + nargs="*", + help="Task ids to inspect. Omit to scan active running/ready/blocked tasks.", + ) + p_reconcile.add_argument( + "--output", + default=None, + help="Write the dry-run/apply JSON result to this path", + ) + p_reconcile.add_argument( + "--json", + action="store_true", + help="Print JSON instead of a concise human summary", + ) + p_reconcile.add_argument( + "--apply", + action="store_true", + help="Apply a previously inspected manifest. Requires --manifest and makes a timestamped DB backup first.", + ) + p_reconcile.add_argument( + "--manifest", + default=None, + help="Path to a dry-run manifest JSON to apply with --apply", + ) + # --- link / unlink --- p_link = sub.add_parser("link", help="Add a parent->child dependency") p_link.add_argument("parent_id") @@ -586,8 +617,8 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu "daemon", help="DEPRECATED — dispatcher now runs in the gateway. Use `hermes gateway start`.", ) - p_daemon.add_argument("--interval", type=float, default=60.0, - help="Seconds between dispatch ticks (default: 60)") + p_daemon.add_argument("--interval", type=float, default=120.0, + help="Seconds between dispatch ticks (default: 120)") p_daemon.add_argument("--max", type=int, default=None, help="Cap number of spawns per tick") p_daemon.add_argument("--failure-limit", type=int, @@ -890,6 +921,7 @@ def _restore_board_env() -> None: "reassign": _cmd_reassign, "diagnostics": _cmd_diagnostics, "diag": _cmd_diagnostics, + "reconcile-completions": _cmd_reconcile_completions, "link": _cmd_link, "unlink": _cmd_unlink, "claim": _cmd_claim, @@ -1222,7 +1254,7 @@ def _cmd_init(args: argparse.Namespace) -> int: print(" hermes gateway start") print() print( - "The gateway hosts an embedded dispatcher that ticks every 60 seconds\n" + "The gateway hosts an embedded dispatcher that ticks every 120 seconds\n" "by default (config: kanban.dispatch_interval_seconds). Without a\n" "running gateway, tasks stay in 'ready' forever." ) @@ -1755,6 +1787,143 @@ def _cmd_diagnostics(args: argparse.Namespace) -> int: return 0 +def _cmd_reconcile_completions(args: argparse.Namespace) -> int: + """Dry-run or apply completed-work reconciliation for malformed-DB incidents.""" + if getattr(args, "apply", False): + manifest_path = getattr(args, "manifest", None) + if not manifest_path: + print( + "kanban reconcile-completions: --apply requires --manifest from a prior dry-run", + file=sys.stderr, + ) + return 2 + if getattr(args, "task_ids", None): + print( + "kanban reconcile-completions: task ids are ignored in --apply mode; edit the manifest instead", + file=sys.stderr, + ) + return 2 + try: + manifest = json.loads(Path(manifest_path).expanduser().read_text(encoding="utf-8")) + if not isinstance(manifest, dict): + raise ValueError("manifest root must be a JSON object") + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(f"kanban reconcile-completions: cannot read manifest: {exc}", file=sys.stderr) + return 1 + if manifest.get("mode") != "dry-run" or not isinstance(manifest.get("items"), list): + print( + "kanban reconcile-completions: --manifest must be a dry-run JSON object with an items list", + file=sys.stderr, + ) + return 2 + if any(not isinstance(item, dict) for item in manifest.get("items", [])): + print( + "kanban reconcile-completions: --manifest items must be JSON objects", + file=sys.stderr, + ) + return 2 + current_board = kb.get_current_board() + current_db_path = kb.kanban_db_path().expanduser().resolve(strict=False) + manifest_board = manifest.get("board") + manifest_db = manifest.get("db_path") + try: + manifest_db_path = Path(str(manifest_db)).expanduser().resolve(strict=False) + except (OSError, TypeError, ValueError): + manifest_db_path = None + if manifest_board != current_board or manifest_db_path != current_db_path: + print( + "kanban reconcile-completions: manifest board/db_path does not match the current board; " + "rerun dry-run for this board before applying", + file=sys.stderr, + ) + return 2 + with kb.connect() as conn: + integrity_before = str(conn.execute("PRAGMA integrity_check").fetchone()[0]) + if integrity_before != "ok": + print( + f"kanban reconcile-completions: refusing to apply; integrity_check={integrity_before!r}", + file=sys.stderr, + ) + return 1 + import sqlite3 + db_path = kb.kanban_db_path() + backup_path = db_path.with_name( + f"{db_path.stem}.backup-{time.strftime('%Y%m%d-%H%M%S', time.localtime())}.{os.getpid()}.db" + ) + backup_path.parent.mkdir(parents=True, exist_ok=True) + dst = sqlite3.connect(str(backup_path)) + try: + conn.backup(dst) + finally: + dst.close() + result = kb.apply_completion_reconciliation_manifest(conn, manifest) + integrity_after = str(conn.execute("PRAGMA integrity_check").fetchone()[0]) + payload = { + "mode": "apply", + "manifest_path": str(Path(manifest_path).expanduser()), + "backup_path": str(backup_path), + "integrity_before": integrity_before, + "integrity_after": integrity_after, + "result": result, + } + if getattr(args, "output", None): + out_path = Path(args.output).expanduser() + out_path.parent.mkdir(parents=True, exist_ok=True) + payload["output_path"] = str(out_path) + out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + if getattr(args, "json", False): + print(json.dumps(payload, indent=2, ensure_ascii=False)) + else: + print(f"Backup: {backup_path}") + print(f"Integrity: before={integrity_before} after={integrity_after}") + print( + "Applied reconciliation: " + f"completed={len(result.get('completed', []))} " + f"skipped={len(result.get('skipped', []))} " + f"failed={len(result.get('failed', []))}" + ) + if getattr(args, "output", None): + print(f"Output: {payload['output_path']}") + return 0 if integrity_after == "ok" and not result.get("failed") else 1 + + with kb.connect() as conn: + manifest = kb.build_completion_reconciliation_manifest( + conn, + list(getattr(args, "task_ids", None) or []) or None, + ) + if getattr(args, "output", None): + out_path = Path(args.output).expanduser() + out_path.parent.mkdir(parents=True, exist_ok=True) + manifest["output_path"] = str(out_path) + out_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + if getattr(args, "json", False): + print(json.dumps(manifest, indent=2, ensure_ascii=False)) + return 0 + + counts: dict[str, int] = {} + for item in manifest.get("items", []): + action = str(item.get("proposed_action", "unknown")) if isinstance(item, dict) else "unknown" + counts[action] = counts.get(action, 0) + 1 + print("Completion reconciliation dry-run") + print(f" Board: {manifest.get('board')}") + print(f" DB: {manifest.get('db_path')}") + print(f" Integrity: {manifest.get('integrity_check')}") + print(" Proposed: " + (", ".join(f"{k}={v}" for k, v in sorted(counts.items())) or "none")) + for item in manifest.get("items", []): + if not isinstance(item, dict): + continue + print( + f" {item.get('task_id')} {item.get('current_status')} " + f"{item.get('evidence_status')} -> {item.get('proposed_action')}" + ) + print(f" {item.get('reason')}") + if getattr(args, "output", None): + print(f" Output: {manifest['output_path']}") + if counts.get("complete"): + print("Review the manifest, then apply with: hermes kanban reconcile-completions --apply --manifest ") + return 0 + + def _cmd_link(args: argparse.Namespace) -> int: with kb.connect() as conn: kb.link_tasks(conn, args.parent_id, args.child_id) @@ -2076,12 +2245,13 @@ def _cmd_daemon(args: argparse.Namespace) -> int: " hermes gateway start # starts the gateway + embedded dispatcher\n" "\n" "Ready tasks will be picked up on the next dispatcher tick\n" - "(default: every 60 seconds). Configure via config.yaml:\n" + "(default: every 120 seconds). Configure via config.yaml:\n" "\n" " kanban:\n" " dispatch_in_gateway: true # default\n" - " dispatch_interval_seconds: 60\n" - " failure_limit: 2 # consecutive non-success attempts before auto-block\n" + " dispatch_interval_seconds: 120\n" + " max_spawn: 3 # live worker concurrency cap\n" + " failure_limit: 4 # consecutive non-success attempts before auto-block\n" "\n" "Running both the gateway AND this standalone daemon will\n" "race for claims. If you truly need the old standalone\n" diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 33de8945ff54..a0677d96cebd 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -935,6 +935,17 @@ class Event: PRIMARY KEY (task_id, platform, chat_id, thread_id) ); +-- Per-board dispatcher lease. Multiple profile gateways may point at the +-- same shared board; the lease elects one gateway process to run the +-- dispatcher tick for a short TTL while allowing automatic failover if that +-- process exits or stops renewing. +CREATE TABLE IF NOT EXISTS dispatcher_leases ( + board TEXT PRIMARY KEY, + owner TEXT NOT NULL, + expires_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); + CREATE INDEX IF NOT EXISTS idx_tasks_assignee_status ON tasks(assignee, status); CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status); CREATE INDEX IF NOT EXISTS idx_links_child ON task_links(child_id); @@ -991,10 +1002,40 @@ def _validate_sqlite_header(path: Path) -> None: return try: with path.open("rb") as handle: - head = handle.read(64) + head = handle.read(100) except OSError: return if head.startswith(_SQLITE_HEADER): + if stat.st_size < 100: + raise sqlite3.DatabaseError( + "truncated SQLite file for " + f"{path}: size_bytes={stat.st_size} shorter than 100-byte header" + ) + raw_page_size = int.from_bytes(head[16:18], "big") + page_size = 65536 if raw_page_size == 1 else raw_page_size + valid_page_size = page_size in {512, 1024, 2048, 4096, 8192, 16384, 32768, 65536} + if not valid_page_size: + raise sqlite3.DatabaseError( + "file is not a database: invalid SQLite page size for " + f"{path}; page_size={raw_page_size}; first_32={head[:32].hex(' ')}" + ) + actual_pages, remainder = divmod(stat.st_size, page_size) + if remainder: + raise sqlite3.DatabaseError( + "truncated SQLite file for " + f"{path}: size_bytes={stat.st_size} is not a multiple of " + f"page_size={page_size}; remainder={remainder}" + ) + header_pages = int.from_bytes(head[28:32], "big") + change_counter = int.from_bytes(head[24:28], "big") + version_valid_for = int.from_bytes(head[92:96], "big") + if header_pages and change_counter == version_valid_for and header_pages > actual_pages: + raise sqlite3.DatabaseError( + "truncated SQLite file for " + f"{path}: header_pages={header_pages} actual_pages={actual_pages} " + f"missing_pages={header_pages - actual_pages} page_size={page_size} " + f"size_bytes={stat.st_size}" + ) return signature = "" if head.startswith(b"SQLit") and _looks_like_tls_record_at(head, 5): @@ -1007,9 +1048,14 @@ def _validate_sqlite_header(path: Path) -> None: ) -class KanbanDbCorruptError(RuntimeError): +class KanbanDbCorruptError(sqlite3.DatabaseError): """Raised when an existing kanban DB file fails integrity checks. + Subclasses :class:`sqlite3.DatabaseError` so callers that only care about + SQLite open failures can handle it with the standard sqlite exception tree, + while kanban-specific recovery paths can still inspect ``db_path`` and + ``backup_path``. + Fail-closed guard against silent recreation of a corrupt board file, which would otherwise destroy the user's tasks. Carries both the original path and the timestamped backup we made before refusing. @@ -1161,8 +1207,19 @@ def connect( path = kanban_db_path(board=board) path.parent.mkdir(parents=True, exist_ok=True) # Cheap byte-level check first — catches the #29507 TLS-overwrite shape - # and other invalid-header cases without opening a sqlite connection. - _validate_sqlite_header(path) + # and other invalid-header cases without opening a sqlite connection. Wrap + # any byte-level corruption in the same fail-closed backup/error type as the + # PRAGMA integrity probe so callers never see a raw sqlite3.DatabaseError. + try: + _validate_sqlite_header(path) + except sqlite3.DatabaseError as exc: + resolved_path = path.resolve() + backup = _backup_corrupt_db(resolved_path) + raise KanbanDbCorruptError( + resolved_path, + backup, + f"sqlite refused to open file: {exc}", + ) from exc # Full integrity probe — catches corruption past the header (malformed # pages, broken internal metadata). Cached per-path after first success # via _INITIALIZED_PATHS so it only runs once per process per path. @@ -1511,6 +1568,92 @@ def _claimer_id() -> str: return f"{host}:{os.getpid()}" +def _dispatcher_board_key(board: Optional[str]) -> str: + """Return a stable board key for the dispatcher lease table.""" + return _normalize_board_slug(board) or get_current_board() + + +def acquire_dispatcher_lease( + conn: sqlite3.Connection, + *, + board: Optional[str] = None, + owner: Optional[str] = None, + ttl_seconds: int = 120, +) -> bool: + """Acquire or renew the short-lived dispatcher lease for ``board``. + + Multiple profile gateways can point at the same shared board. Task-level + CAS still prevents duplicate claims, but without a process-level lease each + gateway independently scans/reaps/spawns on its own interval, which creates + retry waves after crashes or operator unblocks. The lease elects one + dispatcher owner at a time while allowing failover after ``ttl_seconds``. + """ + lease_board = _dispatcher_board_key(board) + lease_owner = owner or _claimer_id() + ttl = max(1, int(ttl_seconds or 1)) + now = int(time.time()) + expires_at = now + ttl + acquired = False + with write_txn(conn): + row = conn.execute( + "SELECT owner, expires_at FROM dispatcher_leases WHERE board = ?", + (lease_board,), + ).fetchone() + if row is None: + conn.execute( + "INSERT INTO dispatcher_leases(board, owner, expires_at, updated_at) " + "VALUES (?, ?, ?, ?)", + (lease_board, lease_owner, expires_at, now), + ) + acquired = True + elif row["owner"] == lease_owner or int(row["expires_at"] or 0) <= now: + conn.execute( + "UPDATE dispatcher_leases SET owner = ?, expires_at = ?, updated_at = ? " + "WHERE board = ?", + (lease_owner, expires_at, now, lease_board), + ) + acquired = True + return acquired + + +def release_dispatcher_lease( + conn: sqlite3.Connection, + *, + board: Optional[str] = None, + owner: Optional[str] = None, +) -> bool: + """Release ``owner``'s dispatcher lease for ``board`` if it still owns it.""" + lease_board = _dispatcher_board_key(board) + lease_owner = owner or _claimer_id() + with write_txn(conn): + cur = conn.execute( + "DELETE FROM dispatcher_leases WHERE board = ? AND owner = ?", + (lease_board, lease_owner), + ) + return cur.rowcount > 0 + + +def owns_dispatcher_lease( + conn: sqlite3.Connection, + *, + board: Optional[str] = None, + owner: Optional[str] = None, +) -> bool: + """Return whether ``owner`` currently holds an unexpired dispatcher lease.""" + lease_board = _dispatcher_board_key(board) + lease_owner = owner or _claimer_id() + now = int(time.time()) + row = conn.execute( + "SELECT owner, expires_at FROM dispatcher_leases WHERE board = ?", + (lease_board,), + ).fetchone() + return bool( + row + and row["owner"] == lease_owner + and int(row["expires_at"] or 0) > now + ) + + # --------------------------------------------------------------------------- # Task creation / mutation # --------------------------------------------------------------------------- @@ -3770,7 +3913,7 @@ def schedule_task( # dispatcher stops retrying and parks the task in ``blocked`` with a reason so # a human can investigate. Prevents retry storms when a worker repeatedly times # out, crashes, or cannot spawn. -DEFAULT_FAILURE_LIMIT = 2 +DEFAULT_FAILURE_LIMIT = 4 # Legacy alias — callers / tests still reference the old name. DEFAULT_SPAWN_FAILURE_LIMIT = DEFAULT_FAILURE_LIMIT @@ -4091,6 +4234,7 @@ def enforce_max_runtime( conn: sqlite3.Connection, *, signal_fn=None, + failure_limit: Optional[int] = None, ) -> list[str]: """Terminate workers whose per-task ``max_runtime_seconds`` has elapsed. @@ -4102,10 +4246,13 @@ def enforce_max_runtime( Runs host-local: only tasks claimed by this host are candidates (same reasoning as ``detect_crashed_workers``). ``signal_fn`` is a - test hook; defaults to ``os.kill`` on POSIX. + test hook; defaults to ``os.kill`` on POSIX. ``failure_limit`` threads + the dispatcher/gateway retry budget into the unified failure counter for + timed-out workers; when omitted, the default Kanban retry budget applies. """ import signal timed_out: list[str] = [] + auto_blocked: list[str] = [] now = int(time.time()) host_prefix = f"{_claimer_id().split(':', 1)[0]}:" @@ -4189,14 +4336,18 @@ def enforce_max_runtime( # emits a ``gave_up`` event on top of the ``timed_out`` we # already emitted. if cur.rowcount == 1: - _record_task_failure( + tripped = _record_task_failure( conn, tid, error=f"elapsed {int(elapsed)}s > limit {int(row['max_runtime_seconds'])}s", outcome="timed_out", + failure_limit=failure_limit, release_claim=False, end_run=False, event_payload_extra={"pid": pid, "sigkill": killed}, ) + if tripped: + auto_blocked.append(tid) + enforce_max_runtime._last_auto_blocked = auto_blocked # type: ignore[attr-defined] return timed_out @@ -4342,18 +4493,11 @@ def set_max_runtime( return cur.rowcount == 1 -def _error_fingerprint(error_text: str) -> str: - """Normalize an error message for grouping identical failures. - - Strips host-specific details (PIDs, timestamps) so that errors - with the same root cause produce the same fingerprint. - """ - fp = re.sub(r'\bpid \d+\b', 'pid N', error_text[:80]) - fp = re.sub(r'\b\d{10,}\b', '', fp) - return fp.lower().strip() - - -def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: +def detect_crashed_workers( + conn: sqlite3.Connection, + *, + failure_limit: Optional[int] = None, +) -> list[str]: """Reclaim ``running`` tasks whose worker PID is no longer alive. Appends a ``crashed`` event and drops the task back to ``ready``. @@ -4456,27 +4600,19 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: # # Protocol-violation crashes force an immediate trip (failure_limit=1) # because clean-exit-without-transition is deterministic: the next - # respawn will do exactly the same thing. Better to surface to a - # human with a clear reason than to loop ``DEFAULT_FAILURE_LIMIT`` - # times first. + # respawn will do exactly the same thing. Ordinary crashes — including + # a batch of ``pid not alive`` reports after a gateway/host incident — + # must use the configured retry budget so transient infrastructure + # failures do not auto-block every affected task at effective_limit=1. auto_blocked: list[str] = [] if crash_details: - # Fingerprint errors to detect systemic failures. - _fp_counts: dict[str, int] = {} - for _, _, _, _, err_text in crash_details: - fp = _error_fingerprint(err_text) - _fp_counts[fp] = _fp_counts.get(fp, 0) + 1 for tid, pid, claimer, protocol_violation, error_text in crash_details: - fp = _error_fingerprint(error_text) - is_systemic = ( - not protocol_violation - and _fp_counts.get(fp, 0) >= 3 - ) tripped = _record_task_failure( conn, tid, error=error_text, outcome="crashed", - failure_limit=1 if (protocol_violation or is_systemic) else None, + failure_limit=1 if protocol_violation else failure_limit, + ignore_task_max_retries=protocol_violation, release_claim=False, end_run=False, event_payload_extra={"pid": pid, "claimer": claimer}, @@ -4497,7 +4633,8 @@ def _record_task_failure( error: str, *, outcome: str, - failure_limit: int = None, + failure_limit: Optional[int] = None, + ignore_task_max_retries: bool = False, release_claim: bool = False, end_run: bool = False, event_payload_extra: Optional[dict] = None, @@ -4530,11 +4667,16 @@ def _record_task_failure( when the breaker trips, so callers can include outcome-specific context (e.g. pid on crash, elapsed on timeout). + ``ignore_task_max_retries`` is reserved for deterministic protocol + violations where the caller needs an immediate trip even if a task has a + larger per-task retry budget. + Resolution order for the effective threshold: - 1. per-task ``max_retries`` if set (nothing else overrides) - 2. caller-supplied ``failure_limit`` (gateway passes the config + 1. caller-supplied ``failure_limit`` when ``ignore_task_max_retries`` is set + 2. per-task ``max_retries`` if set (normal non-protocol failure path) + 3. caller-supplied ``failure_limit`` (gateway passes the config value from ``kanban.failure_limit``; tests pass fixed values) - 3. ``DEFAULT_FAILURE_LIMIT`` + 4. ``DEFAULT_FAILURE_LIMIT`` """ if failure_limit is None: failure_limit = DEFAULT_FAILURE_LIMIT @@ -4549,10 +4691,13 @@ def _record_task_failure( failures = int(row["consecutive_failures"]) + 1 cur_status = row["status"] - # Per-task override wins over both caller-supplied and default - # thresholds. None (the common case) falls through. + # Per-task override usually wins over both caller-supplied and default + # thresholds. Protocol violations are the exception: a clean worker + # exit without kanban_complete/kanban_block is deterministic looping, + # so callers can force the immediate dispatcher threshold. task_override = ( - row["max_retries"] if "max_retries" in row.keys() else None + None if ignore_task_max_retries + else row["max_retries"] if "max_retries" in row.keys() else None ) if task_override is not None: effective_limit = int(task_override) @@ -4910,7 +5055,7 @@ def dispatch_once( result.stale = detect_stale_running( conn, stale_timeout_seconds=stale_timeout_seconds, ) - result.crashed = detect_crashed_workers(conn) + result.crashed = detect_crashed_workers(conn, failure_limit=failure_limit) # detect_crashed_workers stashes protocol-violation auto-blocks on # itself so the public list-return stays stable. Pull them into the # DispatchResult here so telemetry / tests see the trip. @@ -4919,7 +5064,12 @@ def dispatch_once( ) if _crash_auto_blocked: result.auto_blocked.extend(_crash_auto_blocked) - result.timed_out = enforce_max_runtime(conn) + result.timed_out = enforce_max_runtime(conn, failure_limit=failure_limit) + _timeout_auto_blocked = getattr( + enforce_max_runtime, "_last_auto_blocked", [] + ) + if _timeout_auto_blocked: + result.auto_blocked.extend(_timeout_auto_blocked) result.promoted = recompute_ready(conn) # Count tasks already running so max_spawn enforces concurrency rather @@ -4930,32 +5080,34 @@ def dispatch_once( # they sit in status='running' until the worker calls # kanban_complete/kanban_block (or the dispatcher TTL-reclaims them). running_count = 0 - if max_spawn is not None: + if max_spawn is not None or max_in_progress is not None: running_count = int( conn.execute( "SELECT COUNT(*) FROM tasks WHERE status = 'running'" ).fetchone()[0] ) + # Honour kanban.max_in_progress before either ready or review dispatch: if + # the board already has enough running tasks, skip spawning this tick so + # slow workers (local LLMs, resource-constrained hosts) can finish what they + # have before more tasks pile up and time out. + if max_in_progress is not None: + if running_count >= max_in_progress: + return result + # ``max_spawn`` is a live concurrency cap, so the loops below compare + # ``running_count + spawned`` against the cap. ``max_in_progress`` is a + # second live concurrency cap; combine them by lowering the cap itself, + # not by replacing it with the remaining slots. Otherwise a board with + # 2 running tasks, max_spawn=3, max_in_progress=3 would set max_spawn=1 + # and then immediately stop because running_count + spawned >= 1. + if max_spawn is None or max_spawn > max_in_progress: + max_spawn = max_in_progress + ready_rows = conn.execute( "SELECT id, assignee FROM tasks " "WHERE status = 'ready' AND claim_lock IS NULL " "ORDER BY priority DESC, created_at ASC" ).fetchall() - # Honour kanban.max_in_progress: if the board already has enough running - # tasks, skip spawning this tick so slow workers (local LLMs, - # resource-constrained hosts) can finish what they have before more tasks - # pile up and time out. - if max_in_progress is not None and ready_rows: - in_progress = conn.execute( - "SELECT COUNT(*) FROM tasks WHERE status = 'running'" - ).fetchone()[0] - if in_progress >= max_in_progress: - return result - # Only spawn enough to reach the cap, respecting max_spawn too. - remaining = max_in_progress - in_progress - if max_spawn is None or max_spawn > remaining: - max_spawn = remaining spawned = 0 for row in ready_rows: if max_spawn is not None and running_count + spawned >= max_spawn: @@ -5570,7 +5722,7 @@ def _default_spawn( def run_daemon( *, - interval: float = 60.0, + interval: float = 120.0, max_spawn: Optional[int] = None, failure_limit: int = DEFAULT_SPAWN_FAILURE_LIMIT, stop_event=None, @@ -6221,6 +6373,303 @@ def read_worker_log( return None +# --------------------------------------------------------------------------- +# Completion reconciliation helpers +# --------------------------------------------------------------------------- + +_COMPLETION_RECONCILE_PATH_RE = re.compile( + r"(?P/(?:[^\s`'\"<>|]+/)*[^\s`'\"<>|]+\.(?:md|txt|json|csv|pdf|html|png|jpg|jpeg|gif))", + re.IGNORECASE, +) +_COMPLETION_RECONCILE_EVIDENCE_RE = re.compile( + r"(\bwork\s+complete\b|\bcompleted\s+(?:the\s+)?(?:task|work)\b|" + r"\breport\s+saved\b|\bwrote\s+the\s+required\s+report\b)", + re.IGNORECASE, +) +_COMPLETION_RECONCILE_TAIL_BYTES = 64 * 1024 +_REVIEW_GATE_BLOCK_REASON_RE = re.compile( + r"\b(" + r"review[- ]required|requires review|awaiting review|review gate|" + r"human[- ]decision|human decision|requires human|human approval|manual approval" + r")\b", + re.IGNORECASE, +) + + +def _completion_log_has_evidence(text: str) -> bool: + lower = (text or "").lower() + return "database disk image is malformed" in lower and bool( + _COMPLETION_RECONCILE_EVIDENCE_RE.search(text or "") + ) + + +def _completion_artifact_paths(text: str) -> list[str]: + paths: list[str] = [] + for match in _COMPLETION_RECONCILE_PATH_RE.finditer(text or ""): + candidate = match.group("path").strip().rstrip(".,;:)]}") + if not candidate or candidate.startswith("//"): + continue + if candidate not in paths: + paths.append(candidate) + return paths + + +def _task_has_active_claim(task: Task) -> bool: + """Return True when reconciliation should not touch an active worker. + + Completion reconciliation is for workers that likely finished but could not + persist ``kanban_complete`` during a malformed-DB incident. It must not turn + ordinary live worker logs into completion evidence while a current run is + still claimed. + """ + if task.status != "running": + return False + now = int(time.time()) + if task.claim_expires is not None and int(task.claim_expires) > now: + return True + lock = task.claim_lock or "" + host_prefix = f"{_claimer_id().split(':', 1)[0]}:" + return bool( + lock.startswith(host_prefix) + and task.worker_pid + and _pid_alive(task.worker_pid) + ) + + +def _sqlite_integrity_check(conn: sqlite3.Connection) -> str: + try: + row = conn.execute("PRAGMA integrity_check").fetchone() + return str(row[0] if row else "unknown") + except sqlite3.DatabaseError as exc: + return f"error: {exc}" + + +def _latest_block_reason(conn: sqlite3.Connection, task_id: str) -> str: + """Return the most recent human-facing block reason for a task, if any.""" + row = conn.execute( + """ + SELECT summary + FROM task_runs + WHERE task_id = ? + AND outcome = 'blocked' + AND COALESCE(summary, '') != '' + ORDER BY COALESCE(ended_at, started_at) DESC, id DESC + LIMIT 1 + """, + (task_id,), + ).fetchone() + if row and row["summary"]: + return str(row["summary"]) + + event = conn.execute( + """ + SELECT payload + FROM task_events + WHERE task_id = ? + AND kind = 'blocked' + ORDER BY id DESC + LIMIT 1 + """, + (task_id,), + ).fetchone() + if event and event["payload"]: + try: + payload = json.loads(event["payload"]) + except (TypeError, ValueError): + return "" + if isinstance(payload, dict) and payload.get("reason"): + return str(payload["reason"]) + return "" + + +def _task_is_review_or_human_decision_blocked( + conn: sqlite3.Connection, + task: Task, +) -> tuple[bool, str]: + """Return whether a blocked task is waiting on a review/human gate.""" + if task.status != "blocked": + return False, "" + reason = _latest_block_reason(conn, task.id) + return bool(_REVIEW_GATE_BLOCK_REASON_RE.search(reason)), reason + + +def _review_gate_override_reason(item: dict) -> str: + """Return an explicit manifest override reason, or an empty string.""" + if item.get("override_review_gate") is not True: + return "" + reason = item.get("override_reason", item.get("review_gate_override_reason", "")) + return str(reason or "").strip() + + +def build_completion_reconciliation_manifest( + conn: sqlite3.Connection, + task_ids: Optional[Iterable[str]] = None, +) -> dict: + """Build a non-destructive manifest for workers that finished while + kanban_complete could not persist because the board DB was malformed. + """ + if task_ids is None: + rows = conn.execute( + "SELECT id FROM tasks WHERE status IN ('running', 'ready', 'blocked') " + "ORDER BY created_at DESC" + ).fetchall() + resolved_task_ids = [r["id"] for r in rows] + else: + resolved_task_ids = list(task_ids) + + items: list[dict] = [] + for task_id in resolved_task_ids: + task = get_task(conn, task_id) + if task is None: + items.append({ + "task_id": task_id, + "current_status": None, + "evidence_status": "task_missing", + "evidence_source": None, + "evidence_excerpt": "", + "artifact_paths": [], + "proposed_action": "skip", + "reason": "task not found", + "metadata": {}, + }) + continue + log_path = worker_log_path(task_id) + text = read_worker_log(task_id, tail_bytes=_COMPLETION_RECONCILE_TAIL_BYTES) or "" + has_evidence = _completion_log_has_evidence(text) + artifacts = _completion_artifact_paths(text) + metadata = { + "reconciled": True, + "evidence_source": str(log_path), + "artifact_paths": artifacts, + "artifacts": artifacts, + } + review_gate_blocked, block_reason = _task_is_review_or_human_decision_blocked(conn, task) + if task.status == "done": + proposed = "skip" + evidence_status = "already_done" + reason = "task already terminal (done)" + elif _task_has_active_claim(task): + proposed = "skip" + evidence_status = "active_worker_claim" + reason = "task still has an active worker claim; skipping reconciliation" + elif has_evidence and review_gate_blocked: + proposed = "skip" + evidence_status = "review_gate_blocked" + reason = ( + "task is blocked on review/human decision; refusing completion reconciliation " + f"without explicit override (block reason: {block_reason})" + ) + metadata["blocked_reason"] = block_reason + elif has_evidence: + proposed = "complete" + evidence_status = "malformed_db_completion_log" + reason = "worker log indicates completion but kanban_complete failed on malformed DB" + else: + proposed = "skip" + evidence_status = "no_completion_evidence" + reason = "no malformed-db completion evidence found in worker log" + items.append({ + "task_id": task_id, + "current_status": task.status, + "evidence_status": evidence_status, + "evidence_source": str(log_path), + "evidence_excerpt": text[-2000:], + "artifact_paths": artifacts, + "proposed_action": proposed, + "reason": reason, + "metadata": metadata, + }) + return { + "mode": "dry-run", + "board": get_current_board(), + "db_path": str(kanban_db_path()), + "integrity_check": _sqlite_integrity_check(conn), + "items": items, + } + + +def apply_completion_reconciliation_manifest( + conn: sqlite3.Connection, + manifest: dict, +) -> dict: + """Apply a previously reviewed completion-reconciliation manifest.""" + completed: list[str] = [] + skipped: list[str] = [] + failed: list[dict] = [] + for item in manifest.get("items", []): + task_id = item.get("task_id") + if not task_id or item.get("proposed_action") != "complete": + if task_id: + skipped.append(task_id) + continue + task = get_task(conn, task_id) + if task is None or task.status not in {"running", "ready", "blocked"}: + failed.append({ + "task_id": task_id, + "error": "task was not in a completable status or no longer exists", + }) + continue + if _task_has_active_claim(task): + failed.append({ + "task_id": task_id, + "error": "task still has an active worker claim; refusing reconciliation", + }) + continue + review_gate_blocked, block_reason = _task_is_review_or_human_decision_blocked(conn, task) + review_gate_override = _review_gate_override_reason(item) + if review_gate_blocked and not review_gate_override: + failed.append({ + "task_id": task_id, + "error": "task is blocked on review/human decision; explicit override is required", + }) + continue + current_log_text = read_worker_log( + task_id, tail_bytes=_COMPLETION_RECONCILE_TAIL_BYTES, + ) or "" + if not _completion_log_has_evidence(current_log_text): + failed.append({ + "task_id": task_id, + "error": "current worker log no longer contains malformed-db completion evidence", + }) + continue + manifest_artifacts = [str(p).strip() for p in item.get("artifact_paths", []) if str(p).strip()] + current_artifacts = _completion_artifact_paths(current_log_text) + if manifest_artifacts and manifest_artifacts != current_artifacts: + failed.append({ + "task_id": task_id, + "error": "current worker log artifact paths differ from manifest", + }) + continue + summary = "Reconciled completion from worker log evidence after malformed kanban DB write failure." + metadata = item.get("metadata") or { + "reconciled": True, + "evidence_source": item.get("evidence_source"), + "artifact_paths": current_artifacts, + "artifacts": current_artifacts, + } + if isinstance(metadata, dict): + metadata["artifact_paths"] = current_artifacts + metadata["artifacts"] = current_artifacts + if review_gate_override: + metadata["review_gate_override"] = { + "reason": review_gate_override, + "blocked_reason": block_reason, + } + try: + ok = complete_task(conn, task_id, summary=summary, metadata=metadata) + except Exception as exc: + failed.append({"task_id": task_id, "error": str(exc)}) + continue + if ok: + completed.append(task_id) + else: + failed.append({ + "task_id": task_id, + "error": "task was not in a completable status or no longer exists", + }) + return {"completed": completed, "skipped": skipped, "failed": failed} + + # --------------------------------------------------------------------------- # Assignee enumeration (known profiles + per-profile board stats) # --------------------------------------------------------------------------- diff --git a/tests/hermes_cli/test_kanban_cli.py b/tests/hermes_cli/test_kanban_cli.py index fd9b15725135..7543a107876e 100644 --- a/tests/hermes_cli/test_kanban_cli.py +++ b/tests/hermes_cli/test_kanban_cli.py @@ -166,6 +166,141 @@ def test_run_slash_json_output(kanban_home): assert payload["status"] == "ready" +def _write_malformed_completion_log(task_id: str, artifact: Path) -> None: + artifact.write_text("# done\n", encoding="utf-8") + log_path = kb.worker_log_path(task_id) + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the task work and wrote the required report:\n" + f"{artifact}\n" + "kanban_complete: database disk image is malformed\n", + encoding="utf-8", + ) + + +def _run_kanban_tokens(tokens: list[str]) -> tuple[int, str, str]: + import contextlib + import io + + wrap = argparse.ArgumentParser(prog="/kanban-wrap", add_help=False) + top_sub = wrap.add_subparsers(dest="_top") + parser = kc.build_parser(top_sub) + args = parser.parse_args(tokens) + out = io.StringIO() + err = io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = kc.kanban_command(args) + return rc, out.getvalue().strip(), err.getvalue().strip() + + +def test_run_slash_reconcile_completions_dry_run_writes_aligned_manifest(kanban_home, tmp_path): + with kb.connect() as conn: + tid = kb.create_task(conn, title="reconcile me", assignee="worker") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + artifact = tmp_path / "report.md" + _write_malformed_completion_log(tid, artifact) + manifest_path = tmp_path / "manifest.json" + + out = kc.run_slash(f"reconcile-completions {tid} --json --output {manifest_path}") + + payload = json.loads(out) + assert payload["mode"] == "dry-run" + assert payload["integrity_check"] == "ok" + assert payload["output_path"] == str(manifest_path) + assert manifest_path.exists() + written = json.loads(manifest_path.read_text(encoding="utf-8")) + assert written["output_path"] == str(manifest_path) + assert written["items"] == payload["items"] + item = payload["items"][0] + assert item["task_id"] == tid + assert item["current_status"] == "running" + assert item["evidence_status"] == "malformed_db_completion_log" + assert item["proposed_action"] == "complete" + assert item["artifact_paths"] == [str(artifact)] + + +def test_run_slash_reconcile_completions_apply_requires_manifest_and_makes_backup(kanban_home, tmp_path): + missing_manifest = kc.run_slash("reconcile-completions --apply") + assert "--apply requires --manifest" in missing_manifest + + with kb.connect() as conn: + tid = kb.create_task(conn, title="apply reconcile", assignee="worker") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + _write_malformed_completion_log(tid, tmp_path / "apply-report.md") + manifest_path = tmp_path / "manifest.json" + kc.run_slash(f"reconcile-completions {tid} --json --output {manifest_path}") + apply_output = tmp_path / "apply-result.json" + + out = kc.run_slash( + f"reconcile-completions --apply --manifest {manifest_path} --json --output {apply_output}" + ) + + payload = json.loads(out) + assert payload["mode"] == "apply" + assert payload["manifest_path"] == str(manifest_path) + assert payload["integrity_before"] == "ok" + assert payload["integrity_after"] == "ok" + assert payload["result"]["completed"] == [tid] + assert payload["result"]["failed"] == [] + assert Path(payload["backup_path"]).is_file() + assert Path(payload["output_path"]) == apply_output + written = json.loads(apply_output.read_text(encoding="utf-8")) + assert written["output_path"] == str(apply_output) + assert written["result"]["completed"] == [tid] + with kb.connect() as conn: + task = kb.get_task(conn, tid) + assert task is not None + assert task.status == "done" + + +def test_kanban_command_reconcile_completions_apply_reports_stale_manifest_failure(kanban_home, tmp_path): + with kb.connect() as conn: + tid = kb.create_task(conn, title="stale manifest", assignee="worker") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + _write_malformed_completion_log(tid, tmp_path / "stale-report.md") + manifest_path = tmp_path / "manifest.json" + kc.run_slash(f"reconcile-completions {tid} --json --output {manifest_path}") + with kb.connect() as conn: + assert kb.complete_task(conn, tid, summary="completed before stale apply") + + rc, out, err = _run_kanban_tokens( + ["reconcile-completions", "--apply", "--manifest", str(manifest_path), "--json"] + ) + + assert err == "" + assert rc == 1 + payload = json.loads(out) + assert payload["integrity_after"] == "ok" + assert payload["result"]["completed"] == [] + assert payload["result"]["failed"] == [ + {"task_id": tid, "error": "task was not in a completable status or no longer exists"} + ] + + +def test_kanban_command_reconcile_completions_apply_rejects_manifest_for_other_db(kanban_home, tmp_path): + with kb.connect() as conn: + tid = kb.create_task(conn, title="wrong db", assignee="worker") + kb.claim_task(conn, tid) + _write_malformed_completion_log(tid, tmp_path / "wrong-db-report.md") + manifest_path = tmp_path / "manifest.json" + kc.run_slash(f"reconcile-completions {tid} --json --output {manifest_path}") + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["db_path"] = str(tmp_path / "other-board.db") + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + + rc, out, err = _run_kanban_tokens( + ["reconcile-completions", "--apply", "--manifest", str(manifest_path), "--json"] + ) + + assert rc == 2 + assert out == "" + assert "manifest board/db_path does not match the current board" in err + assert not list(kanban_home.glob("*.backup-*.db")) + + def test_run_slash_dispatch_dry_run_counts(kanban_home): kc.run_slash("create 'a' --assignee alice") kc.run_slash("create 'b' --assignee bob") diff --git a/tests/hermes_cli/test_kanban_core_functionality.py b/tests/hermes_cli/test_kanban_core_functionality.py index a97ddbbe15b5..0f36b08ca126 100644 --- a/tests/hermes_cli/test_kanban_core_functionality.py +++ b/tests/hermes_cli/test_kanban_core_functionality.py @@ -13,6 +13,7 @@ import argparse import json import os +import sqlite3 import subprocess import threading import time @@ -78,6 +79,126 @@ def test_no_idempotency_key_never_collides(kanban_home): conn.close() +# --------------------------------------------------------------------------- +# Dispatcher lease +# --------------------------------------------------------------------------- + +def test_dispatcher_lease_allows_single_owner_until_expiry(kanban_home, monkeypatch): + """Only one gateway dispatcher should own a board at a time. + + Multiple profile gateways can run against the shared Kanban DB. The + dispatcher lease prevents every gateway from spawning the same ready + queue on its own interval; ownership renews for the same process and + fails over after the lease expires. + """ + now = [1_700_000_000] + monkeypatch.setattr(kb.time, "time", lambda: now[0]) + conn_a = kb.connect() + conn_b = kb.connect() + try: + assert kb.acquire_dispatcher_lease( + conn_a, board="default", owner="gateway-a", ttl_seconds=60, + ) + assert kb.owns_dispatcher_lease( + conn_a, board="default", owner="gateway-a", + ) + assert not kb.owns_dispatcher_lease( + conn_b, board="default", owner="gateway-b", + ) + assert not kb.acquire_dispatcher_lease( + conn_b, board="default", owner="gateway-b", ttl_seconds=60, + ) + assert kb.acquire_dispatcher_lease( + conn_a, board="default", owner="gateway-a", ttl_seconds=60, + ) + + now[0] += 61 + assert kb.acquire_dispatcher_lease( + conn_b, board="default", owner="gateway-b", ttl_seconds=60, + ) + assert not kb.release_dispatcher_lease( + conn_a, board="default", owner="gateway-a", + ) + assert kb.release_dispatcher_lease( + conn_b, board="default", owner="gateway-b", + ) + assert kb.acquire_dispatcher_lease( + conn_a, board="default", owner="gateway-a", ttl_seconds=60, + ) + finally: + conn_a.close() + conn_b.close() + + +# --------------------------------------------------------------------------- +# Completion reconciliation manifest +# --------------------------------------------------------------------------- + +def test_completion_reconciliation_manifest_marks_done_task_noop(kanban_home, tmp_path): + """Dry-run reconciliation should not propose mutating already-fixed rows.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="already reconciled", assignee="worker") + kb.claim_task(conn, tid) + assert kb.complete_task(conn, tid, summary="Report saved at /tmp/report.md") + + report = tmp_path / "report.md" + report.write_text("# done\n", encoding="utf-8") + log_path = kb.worker_log_path(tid) + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "work complete\n" + "kanban_complete: database disk image is malformed\n" + f"report saved {report}\n", + encoding="utf-8", + ) + + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + finally: + conn.close() + + item = manifest["items"][0] + assert manifest["integrity_check"] == "ok" + assert manifest["mode"] == "dry-run" + assert item["task_id"] == tid + assert item["current_status"] == "done" + assert item["proposed_action"] == "skip" + assert item["reason"] == "task already terminal (done)" + assert item["artifact_paths"] == [str(report)] + assert "database disk image is malformed" in item["evidence_excerpt"] + + +def test_completion_reconciliation_manifest_proposes_completion_for_active_evidence(kanban_home, tmp_path): + """Dry-run reconciliation names evidence and action for stale active rows.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="stale active", assignee="worker") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + report = tmp_path / "report.md" + report.write_text("# done\n", encoding="utf-8") + log_path = kb.worker_log_path(tid) + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the task work and wrote the required report:\n" + f"{report}\n" + "kanban_complete: database disk image is malformed\n", + encoding="utf-8", + ) + + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + finally: + conn.close() + + item = manifest["items"][0] + assert item["current_status"] == "running" + assert item["proposed_action"] == "complete" + assert item["evidence_source"].endswith(f"{tid}.log") + assert item["evidence_status"] == "malformed_db_completion_log" + assert item["artifact_paths"] == [str(report)] + assert item["metadata"]["evidence_source"].endswith(f"{tid}.log") + + # --------------------------------------------------------------------------- # Spawn-failure circuit breaker # --------------------------------------------------------------------------- @@ -90,20 +211,21 @@ def _bad_spawn(task, ws): conn = kb.connect() try: tid = kb.create_task(conn, title="x", assignee="worker") - assert kb.DEFAULT_FAILURE_LIMIT == 2 - # One default-limit failure → still ready, counter grows. - res1 = kb.dispatch_once(conn, spawn_fn=_bad_spawn) - assert tid not in res1.auto_blocked - task = kb.get_task(conn, tid) - assert task.status == "ready" - assert task.consecutive_failures == 1 + assert kb.DEFAULT_FAILURE_LIMIT == 4 + # Default-limit failures before the threshold → still ready, counter grows. + for expected_failures in range(1, kb.DEFAULT_FAILURE_LIMIT): + res = kb.dispatch_once(conn, spawn_fn=_bad_spawn) + assert tid not in res.auto_blocked + task = kb.get_task(conn, tid) + assert task.status == "ready" + assert task.consecutive_failures == expected_failures - # Second default-limit failure trips the guard. + # Fourth default-limit failure trips the guard. res2 = kb.dispatch_once(conn, spawn_fn=_bad_spawn) assert tid in res2.auto_blocked task = kb.get_task(conn, tid) assert task.status == "blocked" - assert task.consecutive_failures >= 2 + assert task.consecutive_failures >= kb.DEFAULT_FAILURE_LIMIT assert task.last_failure_error and "no PATH" in task.last_failure_error finally: conn.close() @@ -236,7 +358,7 @@ def test_per_task_max_retries_overrides_dispatcher_limit(kanban_home, all_assign def test_per_task_max_retries_allows_more_than_default(kanban_home, all_assignees_spawnable): """A task with ``max_retries=5`` does NOT auto-block at the default - limit of 2 — it must reach the per-task override first.""" + limit of 4 — it must reach the per-task override first.""" conn = kb.connect() try: tid = kb.create_task( @@ -981,7 +1103,7 @@ def _signal_fn(pid, sig): def test_repeated_timeouts_auto_block_at_default_limit(kanban_home): - """Two timed_out outcomes on the same task/profile trip the retry guard.""" + """Default-limit timed_out outcomes on the same task/profile trip the retry guard.""" import hermes_cli.kanban_db as _kb original_alive = _kb._pid_alive _kb._pid_alive = lambda pid: False @@ -1002,7 +1124,7 @@ def _age_active_run(conn, tid): conn, title="long job", assignee="worker", max_runtime_seconds=1, ) - for expected_failures in (1, 2): + for expected_failures in range(1, kb.DEFAULT_FAILURE_LIMIT + 1): kb.claim_task(conn, tid) kb._set_worker_pid(conn, tid, os.getpid()) _age_active_run(conn, tid) @@ -1013,7 +1135,7 @@ def _age_active_run(conn, tid): task = kb.get_task(conn, tid) assert task.status == "blocked" events = kb.list_events(conn, tid) - assert [e.kind for e in events].count("timed_out") == 2 + assert [e.kind for e in events].count("timed_out") == kb.DEFAULT_FAILURE_LIMIT gave_up = [e for e in events if e.kind == "gave_up"] assert gave_up and gave_up[-1].payload["trigger_outcome"] == "timed_out" finally: @@ -3362,8 +3484,64 @@ def test_config_default_dispatch_in_gateway_is_true(): f"{kanban.get('dispatch_in_gateway')!r}" ) interval = kanban.get("dispatch_interval_seconds") - assert isinstance(interval, (int, float)) and interval >= 1, ( - f"dispatch_interval_seconds must be a positive number, got {interval!r}" + assert interval == 120, ( + f"dispatch_interval_seconds should default to safe retry cadence 120, got {interval!r}" + ) + assert kanban.get("max_spawn") == 3, ( + f"max_spawn should default to anti-stampede cap 3, got {kanban.get('max_spawn')!r}" + ) + assert kanban.get("failure_limit") == 4, ( + f"failure_limit should default to 4, got {kanban.get('failure_limit')!r}" + ) + assert kb.DEFAULT_FAILURE_LIMIT == 4 + + +def test_gateway_dispatcher_config_parsers_are_defensive(): + """Malformed retry-policy config should not crash the gateway dispatcher.""" + from gateway.run import ( + _parse_kanban_failure_limit, + _parse_kanban_max_spawn, + _parse_positive_float_or_default, + _parse_positive_int_or_default, + ) + + assert _parse_kanban_max_spawn(None, default=3) is None + assert _parse_kanban_max_spawn("3", default=3) == 3 + assert _parse_kanban_max_spawn("bad", default=3) == 3 + assert _parse_kanban_max_spawn(0, default=3) == 3 + assert _parse_kanban_max_spawn(float("inf"), default=3) == 3 + + assert _parse_kanban_failure_limit("4", default=2) == 4 + assert _parse_kanban_failure_limit("bad", default=4) == 4 + assert _parse_kanban_failure_limit(0, default=4) == 4 + assert _parse_kanban_failure_limit(float("inf"), default=4) == 4 + + assert _parse_positive_float_or_default("120.5", 120.0, label="x") == 120.5 + assert _parse_positive_float_or_default("bad", 120.0, label="x") == 120.0 + assert _parse_positive_float_or_default(0, 120.0, label="x") == 120.0 + assert _parse_positive_float_or_default("nan", 120.0, label="x") == 120.0 + assert _parse_positive_float_or_default("inf", 120.0, label="x") == 120.0 + + assert _parse_positive_int_or_default("240", 240, label="x") == 240 + assert _parse_positive_int_or_default("bad", 240, label="x") == 240 + assert _parse_positive_int_or_default(0, 240, label="x") == 240 + assert _parse_positive_int_or_default(float("inf"), 240, label="x") == 240 + + +def test_gateway_corrupt_board_classifier_catches_truncated_sqlite_files(): + from gateway.run import _is_corrupt_kanban_board_db_error + + assert _is_corrupt_kanban_board_db_error( + sqlite3.DatabaseError("truncated SQLite file for board.db: size_bytes=42") + ) + assert _is_corrupt_kanban_board_db_error( + sqlite3.DatabaseError("file is not a database: invalid SQLite header") + ) + assert _is_corrupt_kanban_board_db_error( + sqlite3.DatabaseError("database disk image is malformed") + ) + assert not _is_corrupt_kanban_board_db_error( + sqlite3.DatabaseError("database is locked") ) @@ -3673,13 +3851,10 @@ async def _sleep(_delay): assert sum("not a valid SQLite database" in msg for msg in messages) == 1 assert not any("tick failed on board" in msg for msg in messages) assert not any(record.exc_info for record in caplog.records) - # First tick connect (dispatch) + two probes per `_has_ready_work` call - # (ready then review, both via _kb.connect). The second dispatch tick - # skips the dispatch connect because the corrupt board fingerprint is - # disabled, but the ready/review probes still each connect. PR f55d94a1e - # added the review-column probe alongside the existing ready-column - # probe, bumping this from 3 → 5. - assert calls["connect"] == 5 + # Auto-decompose performs the first connect/lease check and disables the + # corrupt board fingerprint. Later dispatch ticks skip that board entirely; + # the health probe still attempts one connect per loop and suppresses it. + assert calls["connect"] == 3 # --------------------------------------------------------------------------- @@ -4285,6 +4460,44 @@ def test_detect_crashed_workers_protocol_violation_auto_blocks(kanban_home): conn.close() +def test_protocol_violation_ignores_per_task_max_retries(kanban_home): + """Protocol violations are deterministic loops, not flaky task failures. + + Even if a task requests a larger retry budget, a clean worker exit without + kanban_complete/kanban_block must park the task immediately for review. + """ + import hermes_cli.kanban_db as _kb + + conn = kb.connect() + try: + tid = kb.create_task( + conn, title="quiet-with-budget", assignee="worker", max_retries=5, + ) + host_prefix = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, tid, claimer=f"{host_prefix}:mock") + fake_pid = 999996 + kb._set_worker_pid(conn, tid, fake_pid) + + _kb._record_worker_exit(fake_pid, 0) + original_alive = _kb._pid_alive + _kb._pid_alive = lambda p: False + try: + kb.detect_crashed_workers(conn) + finally: + _kb._pid_alive = original_alive + + task = kb.get_task(conn, tid) + assert task.status == "blocked" + assert task.consecutive_failures == 1 + events = kb.list_events(conn, tid) + gave_up = [e for e in events if e.kind == "gave_up"] + assert gave_up + assert gave_up[-1].payload.get("effective_limit") == 1 + assert gave_up[-1].payload.get("limit_source") == "dispatcher" + finally: + conn.close() + + def test_detect_crashed_workers_nonzero_exit_uses_default_limit(kanban_home): """A worker that exited non-zero (real error / crash) uses the normal counter path — one failure doesn't trip the breaker. @@ -4320,6 +4533,48 @@ def test_detect_crashed_workers_nonzero_exit_uses_default_limit(kanban_home): conn.close() +def test_detect_crashed_workers_pid_not_alive_batch_uses_default_limit(kanban_home): + """A host/gateway incident can make many worker PIDs disappear at once. + + Those batch ``pid not alive`` crashes are systemic infrastructure evidence, + not deterministic task protocol violations. They should consume one normal + retry budget entry and return tasks to ready so the dispatcher can retry at + the configured cadence instead of auto-blocking every affected task at + effective_limit=1. + """ + import hermes_cli.kanban_db as _kb + conn = kb.connect() + try: + tids = [] + host_prefix = _kb._claimer_id().split(":", 1)[0] + for idx in range(3): + tid = kb.create_task(conn, title=f"batch-crash-{idx}", assignee="worker") + kb.claim_task(conn, tid, claimer=f"{host_prefix}:mock-{idx}") + kb._set_worker_pid(conn, tid, 990000 + idx) + tids.append(tid) + + original_alive = _kb._pid_alive + _kb._pid_alive = lambda p: False + try: + crashed = kb.detect_crashed_workers(conn) + finally: + _kb._pid_alive = original_alive + + assert set(crashed) == set(tids) + for tid in tids: + task = kb.get_task(conn, tid) + assert task.status == "ready", ( + f"batch pid-not-alive crash should retry normally, got {task.status} for {tid}" + ) + assert task.consecutive_failures == 1 + events = kb.list_events(conn, tid) + kinds = [e.kind for e in events] + assert "crashed" in kinds + assert "gave_up" not in kinds + finally: + conn.close() + + def test_reclaim_task_clears_failure_counter(kanban_home): """Operator reclaim wipes the counter so the next retry gets a fresh budget.""" diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 24b553e9e655..b04fcbc2a2cc 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -48,6 +48,237 @@ def test_init_creates_expected_tables(kanban_home): assert {"tasks", "task_links", "task_comments", "task_events"} <= names +def test_completion_reconciliation_manifest_finds_malformed_db_completion( + kanban_home, +): + """Dry-run reconciliation should identify completed work not persisted. + + During a transient malformed-kanban.db incident workers could finish work, + print the artifact and completion evidence to their per-task log, and then + fail to persist kanban_complete. The reconciliation dry-run must surface a + non-destructive proposed action rather than silently mutating the board. + """ + with kb.connect() as conn: + tid = kb.create_task(conn, title="write report", assignee="nora") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + log_path = kb.worker_logs_dir() / f"{tid}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the task work and wrote the required report:\n" + "/tmp/reports/network-prereqs.md\n" + "kanban_complete: database disk image is malformed\n" + "So the work is done and the report is saved.\n", + encoding="utf-8", + ) + + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + + assert manifest["integrity_check"] == "ok" + assert len(manifest["items"]) == 1 + item = manifest["items"][0] + assert item["task_id"] == tid + assert item["current_status"] == "running" + assert item["evidence_status"] == "malformed_db_completion_log" + assert item["evidence_source"].endswith(f"{tid}.log") + assert item["artifact_paths"] == ["/tmp/reports/network-prereqs.md"] + assert item["proposed_action"] == "complete" + assert "dry-run" in manifest["mode"] + + +def test_completion_reconciliation_manifest_does_not_count_kanban_complete_alone( + kanban_home, +): + """The failing function name/error line is not enough completion evidence.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="not actually done", assignee="nora") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + log_path = kb.worker_logs_dir() / f"{tid}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "kanban_complete: database disk image is malformed\n" + "This task is incomplete and not completed; it still needs review.\n", + encoding="utf-8", + ) + + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + + item = manifest["items"][0] + assert item["evidence_status"] == "no_completion_evidence" + assert item["proposed_action"] == "skip" + + +def test_completion_reconciliation_skips_active_worker_claims(kanban_home): + """A live/current run log must not be auto-completed by reconciliation.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="currently running", assignee="nora") + kb.claim_task(conn, tid) + log_path = kb.worker_logs_dir() / f"{tid}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the task work and wrote the required report:\n" + "/tmp/reports/current.md\n" + "kanban_complete: database disk image is malformed\n", + encoding="utf-8", + ) + + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + item = manifest["items"][0] + assert item["evidence_status"] == "active_worker_claim" + assert item["proposed_action"] == "skip" + + item["proposed_action"] = "complete" + applied = kb.apply_completion_reconciliation_manifest(conn, manifest) + + assert applied["completed"] == [] + assert applied["failed"] == [ + { + "task_id": tid, + "error": "task still has an active worker claim; refusing reconciliation", + } + ] + assert kb.get_task(conn, tid).status == "running" + + +def test_completion_reconciliation_refuses_review_required_blocked_tasks_by_default( + kanban_home, +): + """Review-gated blocked tasks must not be auto-completed by log reconciliation.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="needs code review", assignee="cody") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + assert kb.block_task(conn, tid, reason="review-required: reviewer must approve the diff") + log_path = kb.worker_logs_dir() / f"{tid}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the task work and wrote the required patch:\n" + "/tmp/reports/review-gated.diff\n" + "kanban_complete: database disk image is malformed\n", + encoding="utf-8", + ) + + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + item = manifest["items"][0] + assert item["current_status"] == "blocked" + assert item["evidence_status"] == "review_gate_blocked" + assert item["proposed_action"] == "skip" + assert "review-required" in item["reason"] + + item["proposed_action"] = "complete" # Simulates a stale/default manifest from before this guard. + applied = kb.apply_completion_reconciliation_manifest(conn, manifest) + + assert applied["completed"] == [] + assert applied["failed"] == [ + { + "task_id": tid, + "error": "task is blocked on review/human decision; explicit override is required", + } + ] + assert kb.get_task(conn, tid).status == "blocked" + + +def test_apply_completion_reconciliation_manifest_completes_only_manifested_tasks( + kanban_home, +): + with kb.connect() as conn: + complete_tid = kb.create_task(conn, title="done elsewhere", assignee="sable") + kb.claim_task(conn, complete_tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (complete_tid,)) + missing_tid = kb.create_task(conn, title="needs review", assignee="quinn") + kb.claim_task(conn, missing_tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (missing_tid,)) + log_path = kb.worker_logs_dir() / f"{complete_tid}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the work for kanban task.\n" + "Report saved here:\n/tmp/reports/source-inventory.md\n" + "Kanban issue encountered: database disk image is malformed\n", + encoding="utf-8", + ) + manifest = kb.build_completion_reconciliation_manifest( + conn, [complete_tid, missing_tid] + ) + + applied = kb.apply_completion_reconciliation_manifest(conn, manifest) + + assert applied["completed"] == [complete_tid] + assert applied["skipped"] == [missing_tid] + assert kb.get_task(conn, complete_tid).status == "done" + assert kb.get_task(conn, missing_tid).status == "running" + run = conn.execute( + "SELECT outcome, summary, metadata FROM task_runs " + "WHERE task_id = ? ORDER BY id DESC LIMIT 1", + (complete_tid,), + ).fetchone() + assert run["outcome"] == "completed" + assert "reconciled" in (run["summary"] or "").lower() + assert "source-inventory.md" in (run["metadata"] or "") + assert '"artifacts"' in (run["metadata"] or "") + + +def test_apply_completion_reconciliation_manifest_revalidates_current_log( + kanban_home, +): + with kb.connect() as conn: + tid = kb.create_task(conn, title="stale evidence", assignee="sable") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + log_path = kb.worker_logs_dir() / f"{tid}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the work.\n" + "Report saved: /tmp/reports/original.md\n" + "database disk image is malformed\n", + encoding="utf-8", + ) + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + log_path.write_text( + "kanban_complete: database disk image is malformed\n" + "This task is incomplete now.\n", + encoding="utf-8", + ) + + applied = kb.apply_completion_reconciliation_manifest(conn, manifest) + + assert applied["completed"] == [] + assert applied["failed"] == [ + { + "task_id": tid, + "error": "current worker log no longer contains malformed-db completion evidence", + } + ] + assert kb.get_task(conn, tid).status == "running" + + +def test_apply_completion_reconciliation_manifest_reports_exception_once(kanban_home, monkeypatch): + with kb.connect() as conn: + tid = kb.create_task(conn, title="raises", assignee="worker") + kb.claim_task(conn, tid) + conn.execute("UPDATE tasks SET claim_expires = 0, worker_pid = NULL WHERE id = ?", (tid,)) + log_path = kb.worker_logs_dir() / f"{tid}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "Completed the work.\n" + "Report saved: /tmp/reports/raises.md\n" + "database disk image is malformed\n", + encoding="utf-8", + ) + manifest = kb.build_completion_reconciliation_manifest(conn, [tid]) + + def _raise(*args, **kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(kb, "complete_task", _raise) + with kb.connect() as conn: + applied = kb.apply_completion_reconciliation_manifest(conn, manifest) + + assert applied["completed"] == [] + assert applied["skipped"] == [] + assert applied["failed"] == [{"task_id": tid, "error": "boom"}] + + def test_connect_rejects_tls_record_in_sqlite_header(tmp_path, monkeypatch): """Kanban should classify TLS-looking page-0 clobbers before WAL setup.""" home = tmp_path / ".hermes" @@ -69,6 +300,35 @@ def test_connect_rejects_tls_record_in_sqlite_header(tmp_path, monkeypatch): assert "53 51 4c 69 74 17 03 03 00 13" in msg +def test_connect_rejects_truncated_sqlite_file_before_wal_setup(tmp_path, monkeypatch): + """Kanban should identify file-size/page-count truncation explicitly.""" + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.delenv("HERMES_KANBAN_DB", raising=False) + monkeypatch.delenv("HERMES_KANBAN_HOME", raising=False) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + + truncated = home / "kanban.db" + with sqlite3.connect(truncated) as conn: + conn.execute("CREATE TABLE t(x TEXT)") + conn.execute("INSERT INTO t VALUES ('ok')") + + data = bytearray(truncated.read_bytes()) + page_size = int.from_bytes(data[16:18], "big") + actual_pages = len(data) // page_size + data[28:32] = (actual_pages + 2).to_bytes(4, "big") + truncated.write_bytes(data) + + with pytest.raises(sqlite3.DatabaseError) as exc_info: + kb.connect(board="default") + + msg = str(exc_info.value) + assert "truncated SQLite file" in msg + assert f"header_pages={actual_pages + 2}" in msg + assert f"actual_pages={actual_pages}" in msg + + def test_connect_migrates_legacy_db_before_optional_column_indexes(tmp_path): """Legacy DBs missing additive indexed columns must migrate cleanly. @@ -502,10 +762,16 @@ def test_stale_claim_reclaim_event_records_diagnostic_payload( assert payload["host_local"] is True -def test_detect_crashed_workers_systemic_failure_fast_block( +def test_detect_crashed_workers_systemic_pid_not_alive_uses_normal_retry( kanban_home, monkeypatch, ): - """When many tasks crash with the same error, trip the breaker faster.""" + """A batch of disappeared PIDs should consume normal retry budget. + + Gateway/host restarts can make many local worker PIDs disappear with the + same ``pid not alive`` fingerprint. That is infrastructure evidence, not a + deterministic task bug, so the dispatcher should return those tasks to + ready instead of fast-blocking the whole batch at effective_limit=1. + """ import hermes_cli.kanban_db as _kb monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) @@ -528,9 +794,13 @@ def test_detect_crashed_workers_systemic_failure_fast_block( for tid in task_ids: task = kb.get_task(conn, tid) - assert task.status == "blocked", ( - f"task {tid} should be blocked (systemic), got {task.status}" + assert task.status == "ready", ( + f"task {tid} should retry normally, got {task.status}" ) + assert task.consecutive_failures == 1 + kinds = [e.kind for e in kb.list_events(conn, tid)] + assert "crashed" in kinds + assert "gave_up" not in kinds def test_detect_crashed_workers_isolated_failure_normal_retry( @@ -1151,6 +1421,36 @@ def fake_spawn(task, workspace): assert kb.get_task(conn, ready_b).status == "ready" +def test_dispatch_max_spawn_and_max_in_progress_fill_one_remaining_slot( + kanban_home, all_assignees_spawnable +): + """Combining live caps should still fill the remaining worker slot.""" + spawns = [] + + def fake_spawn(task, workspace): + spawns.append(task.id) + + with kb.connect() as conn: + running_a = kb.create_task(conn, title="running-a", assignee="alice") + running_b = kb.create_task(conn, title="running-b", assignee="bob") + ready_a = kb.create_task(conn, title="ready-a", assignee="carol") + ready_b = kb.create_task(conn, title="ready-b", assignee="dave") + kb.claim_task(conn, running_a) + kb.claim_task(conn, running_b) + + res = kb.dispatch_once( + conn, + spawn_fn=fake_spawn, + max_spawn=3, + max_in_progress=3, + ) + + assert len(res.spawned) == 1 + assert spawns == [ready_a] + assert kb.get_task(conn, ready_a).status == "running" + assert kb.get_task(conn, ready_b).status == "ready" + + def test_dispatch_reclaims_stale_before_spawning(kanban_home): with kb.connect() as conn: t = kb.create_task(conn, title="x", assignee="alice") @@ -2682,6 +2982,65 @@ def fake_spawn(task, workspace, board=None): assert spawns[0] == t +def test_dispatch_review_max_in_progress_skips_when_at_limit( + kanban_home, all_assignees_spawnable, +): + """max_in_progress applies even when only review tasks are queued.""" + spawns = [] + + def fake_spawn(task, workspace, board=None): + spawns.append(task.id) + return 42 + + with kb.connect() as conn: + running = kb.create_task(conn, title="running", assignee="alice") + review = kb.create_task(conn, title="review", assignee="bob") + kb.claim_task(conn, running) + _set_task_status(conn, review, "review") + + res = kb.dispatch_once( + conn, + spawn_fn=fake_spawn, + max_spawn=3, + max_in_progress=1, + ) + + assert not res.spawned + assert spawns == [] + assert kb.get_task(conn, review).status == "review" + + +def test_dispatch_review_max_in_progress_fills_one_remaining_slot( + kanban_home, all_assignees_spawnable, +): + """Review dispatch fills remaining max_in_progress capacity without over-spawning.""" + spawns = [] + + def fake_spawn(task, workspace, board=None): + spawns.append(task.id) + return 42 + + with kb.connect() as conn: + running = kb.create_task(conn, title="running", assignee="alice") + review_a = kb.create_task(conn, title="review-a", assignee="bob") + review_b = kb.create_task(conn, title="review-b", assignee="carol") + kb.claim_task(conn, running) + _set_task_status(conn, review_a, "review") + _set_task_status(conn, review_b, "review") + + res = kb.dispatch_once( + conn, + spawn_fn=fake_spawn, + max_spawn=3, + max_in_progress=2, + ) + + assert len(res.spawned) == 1 + assert spawns == [review_a] + assert kb.get_task(conn, review_a).status == "running" + assert kb.get_task(conn, review_b).status == "review" + + def test_has_spawnable_review_true(kanban_home): """has_spawnable_review returns True when review tasks exist with real profiles.""" with kb.connect() as conn: From 405ba34c5b117ae5c7d6ab1749d48b1f5818ab05 Mon Sep 17 00:00:00 2001 From: Ryan Didur <224447970+ryandidurlabs@users.noreply.github.com> Date: Mon, 25 May 2026 01:14:48 -0700 Subject: [PATCH 2/2] fix: harden kanban sqlite write failures --- hermes_cli/kanban_db.py | 173 +++++++++++++++++++++++++++-- tests/hermes_cli/test_kanban_db.py | 148 ++++++++++++++++++++++++ 2 files changed, 314 insertions(+), 7 deletions(-) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index a0677d96cebd..b2b1a43415d5 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -71,6 +71,7 @@ from __future__ import annotations import contextlib +import glob import json import os import re @@ -983,14 +984,82 @@ def _looks_like_tls_record_at(data: bytes, offset: int) -> bool: ) +def _dir_has_entries(path: Path) -> bool: + """Return True if ``path`` exists and contains at least one entry.""" + try: + with os.scandir(path) as entries: + return next(entries, None) is not None + except (FileNotFoundError, NotADirectoryError, PermissionError, OSError): + return False + + +def _zero_byte_db_state_markers(path: Path) -> list[str]: + """Return nearby artifacts showing a zero-byte Kanban DB is suspicious. + + A missing or intentionally empty DB is valid for first-run setup, but an + existing zero-byte DB next to board logs/workspaces/metadata/backups is a + likely truncation/recovery artifact. Treat that as corruption rather than + letting ``sqlite3.connect()`` silently initialize an empty board over the + evidence. + """ + markers: list[str] = [] + # Do not treat SQLite sidecars alone as prior-state markers. During a + # concurrent first connect, SQLite can create ``kanban.db-journal`` while + # the main DB is still zero bytes; rejecting that transient breaks fresh + # initialization. Durable Kanban artifacts below are what distinguish a + # lost live board from a first-run DB file. + escaped_name = glob.escape(path.name) + escaped_stem = glob.escape(path.stem) + for pattern in ( + f"{escaped_name}.truncated*", + f"{escaped_name}.replaced-empty*", + f"{escaped_name}.*.bak", + f"{escaped_stem}.backup-*.db", + ): + try: + markers.extend(str(candidate) for candidate in path.parent.glob(pattern)) + except OSError: + pass + + try: + home = kanban_home().resolve(strict=False) + resolved = path.resolve(strict=False) + except OSError: + home = kanban_home() + resolved = path + default_db = (home / "kanban.db").resolve(strict=False) + if resolved == default_db: + default_logs = home / "kanban" / "logs" + default_workspaces = home / "kanban" / "workspaces" + if _dir_has_entries(default_logs): + markers.append(str(default_logs)) + if _dir_has_entries(default_workspaces): + markers.append(str(default_workspaces)) + else: + board_logs = path.parent / "logs" + board_workspaces = path.parent / "workspaces" + # board.json is created before the named board DB is initialized, so + # metadata alone is not durable task-state evidence. Require worker + # logs, workspaces, or backup/recovery artifacts for zero-byte rejection. + if _dir_has_entries(board_logs): + markers.append(str(board_logs)) + if _dir_has_entries(board_workspaces): + markers.append(str(board_workspaces)) + + # Stable order + de-dupe keeps error strings deterministic for tests/logs. + return sorted(dict.fromkeys(markers)) + + def _validate_sqlite_header(path: Path) -> None: """Fail early with an actionable error for non-SQLite Kanban DB files. - ``sqlite3.connect()`` creates missing and zero-byte files, so those are - allowed. Existing non-empty files must have the SQLite header before we - hand them to SQLite/WAL setup. This keeps corrupted page-0 failures from - being collapsed into a generic PRAGMA error and lets the gateway's corrupt - board handling identify the board by fingerprint. + ``sqlite3.connect()`` creates missing and truly new zero-byte files, so + those are allowed. Existing non-empty files must have the SQLite header + before we hand them to SQLite/WAL setup. Existing zero-byte files with + nearby board-state markers are treated as truncation, not as a fresh board. + This keeps corrupted page-0 failures from being collapsed into a generic + PRAGMA error and lets the gateway's corrupt board handling identify the + board by fingerprint. """ try: stat = path.stat() @@ -999,6 +1068,17 @@ def _validate_sqlite_header(path: Path) -> None: except OSError: return if stat.st_size == 0: + markers = _zero_byte_db_state_markers(path) + if markers: + marker_preview = ", ".join(markers[:5]) + if len(markers) > 5: + marker_preview += f", … +{len(markers) - 5} more" + raise sqlite3.DatabaseError( + "truncated SQLite file for " + f"{path}: size_bytes=0 with existing board state markers; " + "refusing empty-board reinitialization; " + f"markers={marker_preview}" + ) return try: with path.open("rb") as handle: @@ -1178,6 +1258,77 @@ def _guard_existing_db_is_healthy(path: Path) -> None: raise KanbanDbCorruptError(resolved, backup, reason) +def _validate_kanban_content_invariants(conn: sqlite3.Connection, path: Path) -> None: + """Reject impossible Kanban recovery shapes before callers use the board. + + A healthy empty board has zero rows in every task-owned table. A recovered + DB with comments/events/runs/links but no tasks means recovery copied child + evidence while losing the parent task table; accepting that as a valid board + silently hides task loss and causes downstream triage to operate on an empty + queue. + """ + table_rows = conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ).fetchall() + tables = {str(row["name"] if isinstance(row, sqlite3.Row) else row[0]) for row in table_rows} + required = {"tasks", "task_comments", "task_events", "task_runs", "task_links"} + if not required <= tables: + return + + counts = { + "tasks": int(conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]), + "task_comments": int(conn.execute("SELECT COUNT(*) FROM task_comments").fetchone()[0]), + "task_events": int(conn.execute("SELECT COUNT(*) FROM task_events").fetchone()[0]), + "task_runs": int(conn.execute("SELECT COUNT(*) FROM task_runs").fetchone()[0]), + "task_links": int(conn.execute("SELECT COUNT(*) FROM task_links").fetchone()[0]), + } + dependent_counts = {k: v for k, v in counts.items() if k != "tasks" and v > 0} + if counts["tasks"] == 0 and dependent_counts: + detail = " ".join(f"{name}={count}" for name, count in counts.items()) + raise sqlite3.DatabaseError( + f"inconsistent Kanban DB for {path}: {detail}; " + "refusing zero-task board with dependent task rows" + ) + + +def _connection_main_db_path(conn: sqlite3.Connection) -> Optional[Path]: + """Return the backing file for a connection's main database, when known.""" + try: + rows = conn.execute("PRAGMA database_list").fetchall() + except (AttributeError, sqlite3.Error, TypeError): + return None + for row in rows: + try: + name = row["name"] if isinstance(row, sqlite3.Row) else row[1] + file_name = row["file"] if isinstance(row, sqlite3.Row) else row[2] + except (IndexError, KeyError, TypeError): + continue + if name == "main" and file_name: + return Path(str(file_name)) + return None + + +def _validate_write_target(conn: sqlite3.Connection) -> None: + """Re-check the on-disk main DB before starting a write transaction.""" + path = _connection_main_db_path(conn) + if path is None: + return + try: + stat = path.stat() + except FileNotFoundError as exc: + raise sqlite3.DatabaseError( + f"missing SQLite file for {path}: refusing write transaction on stale connection" + ) from exc + except OSError: + stat = None + if stat is not None and stat.st_size == 0: + raise sqlite3.DatabaseError( + f"truncated SQLite file for {path}: size_bytes=0; " + "refusing write transaction on stale connection" + ) + _validate_sqlite_header(path) + + def connect( db_path: Optional[Path] = None, *, @@ -1250,6 +1401,7 @@ def connect( conn.executescript(SCHEMA_SQL) _migrate_add_optional_columns(conn) _INITIALIZED_PATHS.add(resolved) + _validate_kanban_content_invariants(conn, path) except Exception: conn.close() raise @@ -1529,13 +1681,20 @@ def write_txn(conn: sqlite3.Connection): Use for any multi-statement write (creating a task + link, claiming a task + recording an event, etc.). A claim CAS inside this context is - atomic -- at most one concurrent writer can succeed. + atomic -- at most one concurrent writer can succeed. If SQLite reports a + disk/VFS failure that already aborted the transaction, rollback may also + fail (``cannot rollback - no transaction is active``); preserve the + original exception so logs keep the actionable root cause. """ + _validate_write_target(conn) conn.execute("BEGIN IMMEDIATE") try: yield conn except Exception: - conn.execute("ROLLBACK") + try: + conn.execute("ROLLBACK") + except sqlite3.Error: + _log.debug("rollback after failed Kanban write transaction failed", exc_info=True) raise else: conn.execute("COMMIT") diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index b04fcbc2a2cc..036040f3b891 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -48,6 +48,154 @@ def test_init_creates_expected_tables(kanban_home): assert {"tasks", "task_links", "task_comments", "task_events"} <= names +def test_write_txn_preserves_original_error_when_rollback_also_fails(): + """A failed rollback must not mask the SQLite write failure being debugged.""" + + class _Rows: + def fetchall(self): + return [] + + class RollbackFailsConnection: + def execute(self, sql, *args): + if sql == "PRAGMA database_list": + return _Rows() + if sql == "ROLLBACK": + raise sqlite3.OperationalError("cannot rollback - no transaction is active") + return None + + with pytest.raises(sqlite3.OperationalError, match="disk I/O error"): + with kb.write_txn(RollbackFailsConnection()): # type: ignore[arg-type] + raise sqlite3.OperationalError("disk I/O error") + + +def test_dispatcher_lease_write_error_preserves_original_error_when_rollback_fails(): + """Lease acquisition must not collapse disk I/O errors into rollback noise.""" + + class _Rows: + def fetchall(self): + return [] + + def fetchone(self): + return None + + class DispatcherLeaseWriteFailsConnection: + def execute(self, sql, *args): + if sql == "PRAGMA database_list": + return _Rows() + if sql.startswith("SELECT owner, expires_at FROM dispatcher_leases"): + return _Rows() + if sql.startswith("INSERT INTO dispatcher_leases"): + raise sqlite3.OperationalError("disk I/O error") + if sql == "ROLLBACK": + raise sqlite3.OperationalError("cannot rollback - no transaction is active") + return _Rows() + + with pytest.raises(sqlite3.OperationalError, match="disk I/O error"): + kb.acquire_dispatcher_lease( + DispatcherLeaseWriteFailsConnection(), # type: ignore[arg-type] + board="default", + owner="test-owner", + ) + + +def test_write_txn_rechecks_database_file_before_starting_write(kanban_home): + """A long-lived connection must not write after the main DB is truncated.""" + db_path = kb.kanban_db_path() + conn = kb.connect() + try: + kb.create_task(conn, title="before-corruption") + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + + data = bytearray(db_path.read_bytes()) + page_size = int.from_bytes(data[16:18], "big") + actual_pages = len(data) // page_size + data[28:32] = (actual_pages + 2).to_bytes(4, "big") + # Mark the header page-count as valid so the preflight catches the + # same durable truncation shape observed in production. + data[92:96] = data[24:28] + db_path.write_bytes(data) + + with pytest.raises(sqlite3.DatabaseError, match="truncated SQLite file"): + with kb.write_txn(conn): + conn.execute("INSERT INTO tasks(id, title, status, created_at) VALUES ('t_after', 'after', 'ready', 1)") + finally: + conn.close() + + +def test_write_txn_rejects_zero_byte_database_file_after_connect(kanban_home): + """Write-path validation is stricter than first-run connect() initialization.""" + db_path = kb.kanban_db_path() + conn = kb.connect() + try: + kb.create_task(conn, title="before-zero-byte-truncation") + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + db_path.write_bytes(b"") + + with pytest.raises(sqlite3.DatabaseError, match="truncated SQLite file"): + with kb.write_txn(conn): + conn.execute("INSERT INTO tasks(id, title, status, created_at) VALUES ('t_after_zero', 'after', 'ready', 1)") + finally: + conn.close() + + +def test_connect_rejects_zero_byte_db_when_board_state_markers_exist(kanban_home): + """A zero-byte DB next to durable board artifacts is truncation, not first-run.""" + db_path = kb.kanban_db_path() + with kb.connect() as conn: + kb.create_task(conn, title="before-zero-byte-marker") + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + logs_dir = kanban_home / "kanban" / "logs" + logs_dir.mkdir(parents=True) + (logs_dir / "t_existing.log").write_text("prior worker log", encoding="utf-8") + db_path.write_bytes(b"") + with pytest.raises(kb.KanbanDbCorruptError, match="existing board state markers"): + kb.connect() + + +def test_zero_byte_backup_marker_globs_escape_database_name(tmp_path): + """Backup marker detection must not expand metacharacters in explicit DB names.""" + db_path = tmp_path / "kanban[abc].db" + db_path.write_bytes(b"") + # This sibling would match the unescaped glob pattern "kanban[abc].db.truncated*" + # even though it is not a marker for ``kanban[abc].db``. + (tmp_path / "kanbana.db.truncated-by-other-board").write_text("not this db", encoding="utf-8") + + with kb.connect(db_path) as conn: + assert conn.execute("PRAGMA quick_check").fetchone()[0] == "ok" + + +def test_write_txn_rejects_missing_database_file_after_connect(kanban_home): + """A stale connection whose main DB disappeared must not accept writes.""" + db_path = kb.kanban_db_path() + conn = kb.connect() + try: + kb.create_task(conn, title="before-missing-db") + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + db_path.unlink() + + with pytest.raises(sqlite3.DatabaseError, match="missing SQLite file"): + with kb.write_txn(conn): + conn.execute("INSERT INTO tasks(id, title, status, created_at) VALUES ('t_after_missing', 'after', 'ready', 1)") + finally: + conn.close() + + +def test_connect_rejects_zero_task_db_with_dependent_task_rows(kanban_home): + """Recovered child rows without parent tasks are not a healthy empty board.""" + db_path = kb.kanban_db_path() + with kb.connect() as conn: + task_id = kb.create_task(conn, title="before-bad-recovery") + conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) + conn.execute( + "INSERT INTO task_events(task_id, kind, payload, created_at) VALUES (?, 'note', NULL, 1)", + (task_id,), + ) + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + + with pytest.raises(sqlite3.DatabaseError, match="zero-task board with dependent task rows"): + kb.connect(db_path) + + def test_completion_reconciliation_manifest_finds_malformed_db_completion( kanban_home, ):