"""Tests for the Kanban DB layer (hermes_cli.kanban_db).""" from __future__ import annotations import concurrent.futures import os import sqlite3 import subprocess import sys import time import types import unittest.mock from pathlib import Path import pytest import hermes_state from hermes_cli import kanban_db as kb @pytest.fixture def kanban_home(tmp_path, monkeypatch): """Isolated HERMES_HOME with an empty kanban DB.""" home = tmp_path / ".hermes" home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setattr(Path, "home", lambda: tmp_path) kb.init_db() return home def _init_git_repo(repo: Path) -> None: repo.mkdir(parents=True, exist_ok=True) subprocess.run(["git", "init", "-b", "main", str(repo)], check=True, capture_output=True, text=True) subprocess.run(["git", "-C", str(repo), "config", "user.email", "kanban@example.com"], check=True, capture_output=True, text=True) subprocess.run(["git", "-C", str(repo), "config", "user.name", "Kanban Test"], check=True, capture_output=True, text=True) (repo / "README.md").write_text("hello\n", encoding="utf-8") subprocess.run(["git", "-C", str(repo), "add", "README.md"], check=True, capture_output=True, text=True) subprocess.run(["git", "-C", str(repo), "commit", "-m", "init"], check=True, capture_output=True, text=True) # --------------------------------------------------------------------------- # Schema / init # --------------------------------------------------------------------------- def test_cross_process_init_lock_uses_windows_byte_range_lock(tmp_path, monkeypatch): """Windows must use a real (non-blocking) process lock, not a no-op open. The init lock acquires with LK_NBLCK in a bounded retry loop (#36644) so a wedged holder can never block connect() forever; a clean acquire takes the lock once and releases it once. """ calls: list[tuple[int, int, int]] = [] fake_msvcrt = types.SimpleNamespace( LK_NBLCK=3, LK_UNLCK=2, locking=lambda fd, mode, nbytes: calls.append((fd, mode, nbytes)), ) monkeypatch.setattr(kb, "_IS_WINDOWS", True) monkeypatch.setitem(sys.modules, "msvcrt", fake_msvcrt) db_path = tmp_path / "kanban.db" with kb._cross_process_init_lock(db_path): # Acquired exactly once via the non-blocking byte-range lock. assert [call[1:] for call in calls] == [(fake_msvcrt.LK_NBLCK, 1)] # Released once on exit. assert [call[1:] for call in calls] == [ (fake_msvcrt.LK_NBLCK, 1), (fake_msvcrt.LK_UNLCK, 1), ] def test_connect_migrates_legacy_db_before_optional_column_indexes(tmp_path): """Legacy DBs missing additive indexed columns must migrate cleanly. SCHEMA_SQL runs in ``connect()`` before ``_migrate_add_optional_columns``. Indexes over additive columns therefore must be created after the migration adds those columns, or boards predating the column fail to open before migration can run. Covers all four indexes that sit on additive columns: - ``tasks.session_id`` -> ``idx_tasks_session_id`` (#28447) - ``tasks.tenant`` -> ``idx_tasks_tenant`` (#16081) - ``tasks.idempotency_key`` -> ``idx_tasks_idempotency`` (#17805) - ``task_events.run_id`` -> ``idx_events_run`` (#17805) """ db_path = tmp_path / "legacy-kanban.db" conn = sqlite3.connect(str(db_path)) # Pre-#16081 ``tasks`` shape: missing tenant, idempotency_key, session_id. conn.execute(""" CREATE TABLE tasks ( id TEXT PRIMARY KEY, title TEXT NOT NULL, body TEXT, assignee TEXT, status TEXT NOT NULL, priority INTEGER NOT NULL DEFAULT 0, created_by TEXT, created_at INTEGER NOT NULL, started_at INTEGER, completed_at INTEGER, workspace_kind TEXT NOT NULL DEFAULT 'scratch', workspace_path TEXT, claim_lock TEXT, claim_expires INTEGER ) """) # Pre-#17805 ``task_events`` shape: missing run_id. Required because # ``_migrate_add_optional_columns`` unconditionally runs PRAGMA on # ``task_events`` for run_id back-fill. conn.execute(""" CREATE TABLE task_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, kind TEXT NOT NULL, payload TEXT, created_at INTEGER NOT NULL ) """) conn.execute( "INSERT INTO tasks (id, title, status, created_at) " "VALUES ('legacy', 'old board task', 'ready', 1)" ) conn.commit() conn.close() with kb.connect(db_path) as migrated: task_columns = { row["name"] for row in migrated.execute("PRAGMA table_info(tasks)") } event_columns = { row["name"] for row in migrated.execute("PRAGMA table_info(task_events)") } indexes = { row["name"] for row in migrated.execute( "SELECT name FROM sqlite_master WHERE type = 'index'" ) } # Additive columns added by migration: assert "session_id" in task_columns assert "tenant" in task_columns assert "idempotency_key" in task_columns assert "run_id" in event_columns # And their indexes — the regression scope of this test: assert "idx_tasks_session_id" in indexes assert "idx_tasks_tenant" in indexes assert "idx_tasks_idempotency" in indexes assert "idx_events_run" in indexes # --------------------------------------------------------------------------- # Task creation + status inference # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Links + dependency resolution # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Atomic claim (CAS) # --------------------------------------------------------------------------- def test_schedule_task_parks_time_delay_without_dispatching(kanban_home): with kb.connect() as conn: t = kb.create_task(conn, title="delayed recheck", assignee="ops") assert kb.schedule_task(conn, t, reason="run next week") is True task = kb.get_task(conn, t) assert task.status == "scheduled" assert kb.claim_task(conn, t) is None events = kb.list_events(conn, t) assert any(e.kind == "scheduled" and e.payload == {"reason": "run next week"} for e in events) def test_stale_claim_reclaim_event_records_diagnostic_payload( kanban_home, monkeypatch, ): """``reclaimed`` events should carry claim_expires, last_heartbeat_at, and worker_pid so operators can diagnose why a claim went stale (#23025: previous payload only had ``stale_lock`` which gives no timing context).""" import json import hermes_cli.kanban_db as _kb with kb.connect() as conn: t = kb.create_task(conn, title="x", assignee="a") host = _kb._claimer_id().split(":", 1)[0] kb.claim_task(conn, t, claimer=f"{host}:worker") kb._set_worker_pid(conn, t, 12345) old_expires = int(time.time()) - 3600 hb_at = int(time.time()) - 1800 conn.execute( "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? " "WHERE id = ?", (old_expires, hb_at, t), ) monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) row = conn.execute( "SELECT payload FROM task_events " "WHERE task_id = ? AND kind = 'reclaimed'", (t,), ).fetchone() assert row is not None payload = json.loads(row["payload"]) assert payload["claim_expires"] == old_expires assert payload["last_heartbeat_at"] == hb_at assert payload["worker_pid"] == 12345 assert payload["host_local"] is True # --------------------------------------------------------------------------- # Rate-limit requeue: a worker that bails on a provider quota wall must be # released back to ``ready`` WITHOUT counting a failure, so a long (e.g. # 5-hour) quota window can't trip the circuit breaker and permanently block # the card. The respawn guard then defers it on a cooldown until quota # returns. Regression coverage for the kanban-rate-limit-failure report. # --------------------------------------------------------------------------- def _exited_status(code: int) -> int: """Raw wait-status for a WIFEXITED child with the given exit code.""" return code << 8 def test_rate_limit_exit_requeues_without_counting_failure( kanban_home, monkeypatch, ): """A rate-limit sentinel exit releases the task to ``ready`` and leaves ``consecutive_failures`` untouched — the breaker must never trip on a transient throttle, even across many quota-wall hits.""" import hermes_cli.kanban_db as _kb monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0") with kb.connect() as conn: host = _kb._claimer_id().split(":", 1)[0] tid = kb.create_task(conn, title="rl", assignee="a") # Simulate FAR more quota-wall hits than DEFAULT_FAILURE_LIMIT (2). # If any of these counted as a failure the task would be blocked. for i in range(6): pid = 70000 + i # Claim to open a real run (so detect_crashed_workers can close # it with a rate_limited outcome), then point the claim at this # host + a dead pid so the crash path acts on it. kb.claim_task(conn, tid, claimer=f"{host}:w{i}") conn.execute( "UPDATE tasks SET worker_pid=?, consecutive_failures=? " "WHERE id=?", (pid, 0, tid), ) conn.commit() _kb._record_worker_exit( pid, _exited_status(_kb.KANBAN_RATE_LIMIT_EXIT_CODE) ) crashed = kb.detect_crashed_workers(conn) # Rate-limited requeues are NOT crashes. assert tid not in crashed rl = getattr(_kb.detect_crashed_workers, "_last_rate_limited", []) assert tid in rl task = kb.get_task(conn, tid) assert task.status == "ready", ( f"hit {i}: should requeue ready, got {task.status}" ) assert task.consecutive_failures == 0, ( f"hit {i}: rate-limit must not count a failure, " f"got {task.consecutive_failures}" ) # Last failure error stamped so the respawn guard recognizes the # quota wall. assert task.last_failure_error and "rate-limited" in task.last_failure_error # A ``rate_limited`` run outcome was recorded (not ``crashed``). outcomes = [ r["outcome"] for r in conn.execute( "SELECT outcome FROM task_runs WHERE task_id=?", (tid,), ).fetchall() ] assert "rate_limited" in outcomes assert "crashed" not in outcomes def test_respawn_guard_defers_rate_limited_within_cooldown( kanban_home, monkeypatch, ): """Within the cooldown after a rate-limit requeue, the guard defers the respawn; after the cooldown it allows a probe — and crucially does NOT fall into ``blocker_auth`` (which would defer forever).""" import hermes_cli.kanban_db as _kb monkeypatch.setenv("HERMES_KANBAN_RATE_LIMIT_COOLDOWN_SECONDS", "300") now = 5_000_000 with kb.connect() as conn: tid = kb.create_task(conn, title="rl-guard", assignee="a") # Seed a rate_limited run that just ended + the stamped error. kb.claim_task(conn, tid) run_id = kb.get_task(conn, tid).current_run_id conn.execute( "UPDATE task_runs SET outcome='rate_limited', status='rate_limited', " "ended_at=? WHERE id=?", (now, run_id), ) conn.execute( "UPDATE tasks SET status='ready', current_run_id=NULL, " "claim_lock=NULL, claim_expires=NULL, worker_pid=NULL, " "last_failure_error=? WHERE id=?", ("pid 1 exited rate-limited (quota wall) — requeued", tid), ) conn.commit() # Inside cooldown → defer with the rate-limit-specific reason. monkeypatch.setattr(_kb.time, "time", lambda: now + 100) assert kb.check_respawn_guard(conn, tid) == "rate_limit_cooldown" # Past cooldown → allowed (None), NOT trapped by blocker_auth even # though last_failure_error contains "rate-limited". monkeypatch.setattr(_kb.time, "time", lambda: now + 400) assert kb.check_respawn_guard(conn, tid) is None # --------------------------------------------------------------------------- # Complete / block / unblock / archive / assign # --------------------------------------------------------------------------- def test_recompute_ready_honours_dispatcher_failure_limit(kanban_home): """The guard's effective limit must follow the same resolution order as the circuit breaker (#35072): per-task max_retries → dispatcher failure_limit → DEFAULT_FAILURE_LIMIT. Without threading the dispatcher's ``kanban.failure_limit`` through, the guard falls back to DEFAULT_FAILURE_LIMIT and disagrees with the breaker — sticking a task prematurely (config limit > default) or letting a tripped task escape (config limit < default). """ with kb.connect() as conn: # Config allows MORE retries than the default. A task blocked # with failures below the configured limit must still recover. t = kb.create_task(conn, title="lenient", assignee="a") conn.execute( "UPDATE tasks SET status='blocked', consecutive_failures=? " "WHERE id=?", (kb.DEFAULT_FAILURE_LIMIT, t), ) conn.commit() # Default-limit call would stick it (failures >= default). assert kb.recompute_ready(conn) == 0 assert kb.get_task(conn, t).status == "blocked" # Dispatcher configured a higher limit → recover, preserve counter. promoted = kb.recompute_ready( conn, failure_limit=kb.DEFAULT_FAILURE_LIMIT + 2 ) assert promoted == 1 task = kb.get_task(conn, t) assert task.status == "ready" assert task.consecutive_failures == kb.DEFAULT_FAILURE_LIMIT # Config allows FEWER retries than the default. A task at the # stricter limit must stay blocked even though it's below default. t2 = kb.create_task(conn, title="strict", assignee="a") conn.execute( "UPDATE tasks SET status='blocked', consecutive_failures=1 " "WHERE id=?", (t2,), ) conn.commit() # Default-limit (2) would recover it (1 < 2). # Stricter config limit (1) must keep it blocked (1 >= 1). assert kb.recompute_ready(conn, failure_limit=1) == 0 assert kb.get_task(conn, t2).status == "blocked" # --------------------------------------------------------------------------- # Parent-completion invariant at the claim gate (RCA t_a6acd07d) # --------------------------------------------------------------------------- def test_delete_archived_task_removes_related_rows(kanban_home): with kb.connect() as conn: parent = kb.create_task(conn, title="parent") tid = kb.create_task(conn, title="child", parents=[parent], assignee="worker") kb.add_comment(conn, tid, "user", "cleanup me") kb.claim_task(conn, tid) kb.complete_task(conn, tid, result="done") assert kb.archive_task(conn, tid) conn.execute( "INSERT INTO kanban_notify_subs(task_id, platform, chat_id, thread_id, user_id, created_at, last_event_id) " "VALUES (?, 'telegram', '123', '', 'u', 0, 0)", (tid,), ) conn.commit() assert kb.delete_archived_task(conn, tid) is True assert kb.get_task(conn, tid) is None assert conn.execute("SELECT COUNT(*) FROM task_links WHERE child_id = ? OR parent_id = ?", (tid, tid)).fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM task_comments WHERE task_id = ?", (tid,)).fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM task_events WHERE task_id = ?", (tid,)).fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM task_runs WHERE task_id = ?", (tid,)).fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM kanban_notify_subs WHERE task_id = ?", (tid,)).fetchone()[0] == 0 def test_delete_task_removes_task_and_cascades(kanban_home): with kb.connect() as conn: t = kb.create_task(conn, title="to-delete", assignee="alice") kb.add_comment(conn, t, "user", "comment") kb.add_comment(conn, t, "user", "another") assert kb.delete_task(conn, t) assert kb.get_task(conn, t) is None assert len(kb.list_comments(conn, t)) == 0 assert len(kb.list_events(conn, t)) == 0 assert len(kb.list_runs(conn, t)) == 0 # --------------------------------------------------------------------------- # Comments / events / worker context # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Dispatcher # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Respawn guard (check_respawn_guard + dispatch_once integration) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Workspace resolution # --------------------------------------------------------------------------- def test_worktree_workspace_explicit_target_materializes_linked_worktree(kanban_home, tmp_path): repo = tmp_path / "repo" _init_git_repo(repo) target = repo / ".worktrees" / "custom-task" branch = "wt/custom-task" with kb.connect() as conn: t = kb.create_task( conn, title="ship", workspace_kind="worktree", workspace_path=str(target), branch_name=branch, ) task = kb.get_task(conn, t) assert task is not None ws = kb.resolve_workspace(task) assert ws == target assert ws.exists() repo_common = subprocess.run( ["git", "-C", str(repo), "rev-parse", "--path-format=absolute", "--git-common-dir"], check=True, capture_output=True, text=True, ).stdout.strip() ws_common = subprocess.run( ["git", "-C", str(ws), "rev-parse", "--path-format=absolute", "--git-common-dir"], check=True, capture_output=True, text=True, ).stdout.strip() assert ws_common == repo_common listed = subprocess.run( ["git", "-C", str(repo), "worktree", "list", "--porcelain"], check=True, capture_output=True, text=True, ).stdout assert f"worktree {target}" in listed assert f"branch refs/heads/{branch}" in listed # --------------------------------------------------------------------------- # Scratch cleanup containment (#28818) # --------------------------------------------------------------------------- def test_complete_task_persists_scratch_artifacts_before_cleanup(kanban_home): """Completion artifacts from scratch workspaces survive workspace cleanup.""" with kb.connect() as conn: t = kb.create_task(conn, title="render chart") task = kb.get_task(conn, t) ws = kb.resolve_workspace(task) kb.set_workspace_path(conn, t, ws) artifact = ws / "chart.png" artifact.write_bytes(b"png-bytes") assert kb.complete_task( conn, t, result="ok", metadata={"artifacts": [str(artifact)]}, ) completed = [e for e in kb.list_events(conn, t) if e.kind == "completed"][-1] persisted = Path(completed.payload["artifacts"][0]) run = kb.latest_run(conn, t) assert not ws.exists(), "scratch workspace should still be cleaned up" assert persisted.exists(), "artifact copy should survive scratch cleanup" assert persisted.parent == kb.task_attachments_dir(t) assert persisted.name == "chart.png" assert persisted.read_bytes() == b"png-bytes" assert str(persisted) != str(artifact) assert run is not None assert run.metadata["artifacts"] == [str(persisted)] with kb.connect() as conn: attachments = kb.list_attachments(conn, t) assert [(a.filename, a.stored_path) for a in attachments] == [ ("chart.png", str(persisted.resolve())) ] # --------------------------------------------------------------------------- # Deferred scratch cleanup for parent/child handoff (#33774) # --------------------------------------------------------------------------- def test_dir_child_completion_unblocks_deferred_scratch_parent(kanban_home, tmp_path): """A non-scratch ('dir') child completing must still sweep its scratch parent. Regression for the gap where ``_cleanup_workspace`` returned early for a non-scratch task and never ran the parent sweep — leaking the parent's deferred scratch dir forever. """ child_dir = tmp_path / "persistent-child" child_dir.mkdir() with kb.connect() as conn: parent = kb.create_task(conn, title="scratch parent") child = kb.create_task( conn, title="dir child", workspace_kind="dir", workspace_path=str(child_dir), ) kb.link_tasks(conn, parent, child) p_task = kb.get_task(conn, parent) parent_ws = kb.resolve_workspace(p_task) kb.set_workspace_path(conn, parent, parent_ws) kb.complete_task(conn, parent, result="handoff") assert parent_ws.exists(), "deferred while dir child active" kb.complete_task(conn, child, result="built") assert not parent_ws.exists(), ( "A 'dir' child completing must trigger the parent scratch sweep" ) assert child_dir.exists(), "Non-scratch 'dir' child workspace is never deleted" def test_is_managed_scratch_path_rejects_kanban_metadata_subtrees(kanban_home): """Hermes' own DB/metadata/log subtrees under ``/kanban`` are NOT managed. Regression guard for the Copilot finding on #28819: a scratch task whose ``workspace_path`` was mis-set to the kanban home, the logs dir, or a board's metadata dir (i.e. the board root itself, not its ``workspaces/`` child) must be refused. Without this, the containment check would happily ``shutil.rmtree`` Hermes' DB/metadata/logs on task completion. """ kanban_root = kanban_home / "kanban" kanban_root.mkdir(parents=True, exist_ok=True) assert not kb._is_managed_scratch_path(kanban_root) logs_dir = kanban_root / "logs" logs_dir.mkdir(parents=True, exist_ok=True) assert not kb._is_managed_scratch_path(logs_dir) board_root = kanban_root / "boards" / "my-board" board_root.mkdir(parents=True, exist_ok=True) # The board root itself is NOT a managed scratch dir — only the # ``workspaces/`` child (and its descendants) are. assert not kb._is_managed_scratch_path(board_root) # Sibling subtrees of ``workspaces/`` under a board (e.g. its kanban.db # or board.json living next to ``workspaces/``) are also not managed. board_logs = board_root / "logs" board_logs.mkdir(parents=True, exist_ok=True) assert not kb._is_managed_scratch_path(board_logs) # Now create the board's workspaces dir and a task scratch dir under it — # the latter is the only thing the guard should allow. board_workspaces = board_root / "workspaces" board_workspaces.mkdir(parents=True, exist_ok=True) # The workspaces root itself is also NOT managed — deleting it would # wipe every task's scratch dir at once. assert not kb._is_managed_scratch_path(board_workspaces) task_dir = board_workspaces / "task-42" task_dir.mkdir(parents=True, exist_ok=True) assert kb._is_managed_scratch_path(task_dir) # --------------------------------------------------------------------------- # Tenancy # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Originating session id (ACP propagation) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Shared-board path resolution (issue #19348) # # The kanban board is a cross-profile coordination primitive: a worker # spawned with `hermes -p ` must read/write the same kanban.db # as the dispatcher that claimed the task. These tests exercise the # path-resolution layer directly and would have caught the regression # where `kanban_db_path()` resolved to the active profile's HERMES_HOME. # --------------------------------------------------------------------------- class TestSharedBoardPaths: """`kanban_home`/`kanban_db_path`/`workspaces_root`/`worker_log_path` must anchor at the **shared root**, not the active profile's HERMES_HOME.""" def _set_home(self, monkeypatch, tmp_path, hermes_home): monkeypatch.setattr(Path, "home", lambda: tmp_path) monkeypatch.setenv("HERMES_HOME", str(hermes_home)) monkeypatch.delenv("HERMES_KANBAN_HOME", raising=False) def test_profile_worker_resolves_to_shared_root( self, tmp_path, monkeypatch ): # Reproduces the bug: dispatcher uses ~/.hermes/kanban.db, # worker spawned with -p previously resolved to # ~/.hermes/profiles//kanban.db. After the fix both # converge on ~/.hermes/kanban.db. default_home = tmp_path / ".hermes" default_home.mkdir() profile_home = default_home / "profiles" / "nehemiahkanban" profile_home.mkdir(parents=True) self._set_home(monkeypatch, tmp_path, profile_home) # All four resolvers must anchor at the shared root, not the # profile-local HERMES_HOME. assert kb.kanban_home() == default_home assert kb.kanban_db_path() == default_home / "kanban.db" assert kb.workspaces_root() == default_home / "kanban" / "workspaces" assert ( kb.worker_log_path("t_0d214f19") == default_home / "kanban" / "logs" / "t_0d214f19.log" ) # Sanity: the profile-local path that used to be returned is # explicitly NOT what we resolve to anymore. assert kb.kanban_db_path() != profile_home / "kanban.db" def test_dispatcher_and_worker_share_a_real_database( self, tmp_path, monkeypatch ): # Belt-and-suspenders: round-trip a task across the two # HERMES_HOME perspectives via a real SQLite file. Without the # fix the worker would open a different file and see no rows. default_home = tmp_path / ".hermes" default_home.mkdir() profile_home = default_home / "profiles" / "nehemiahkanban" profile_home.mkdir(parents=True) # Dispatcher creates the board and a task. self._set_home(monkeypatch, tmp_path, default_home) kb.init_db() with kb.connect() as conn: task_id = kb.create_task(conn, title="cross-profile") # Worker switches to the profile HERMES_HOME and reads. monkeypatch.setenv("HERMES_HOME", str(profile_home)) with kb.connect() as conn: task = kb.get_task(conn, task_id) assert task is not None assert task.title == "cross-profile" def test_dispatcher_spawn_injects_kanban_paths_without_stale_session( self, tmp_path, monkeypatch ): # The dispatcher must pin board paths while stripping any unrelated # HERMES_SESSION_* identity inherited from the long-lived gateway. # The one exception is HERMES_SESSION_SOURCE, which the dispatcher # re-sets to its own `kanban` tag AFTER the strip — a value it owns, # never one inherited from whatever the gateway last routed. default_home = tmp_path / ".hermes" default_home.mkdir() self._set_home(monkeypatch, tmp_path, default_home) from gateway import session_context as sc # A dispatcher can launch before the gateway binds its first session. monkeypatch.setattr(sc, "_session_context_engaged", False) sc.reset_session_vars() for key in sc._VAR_MAP: monkeypatch.setenv(key, "stale-routing-value") captured = {} class _FakePopen: def __init__(self, cmd, **kwargs): captured["cmd"] = cmd captured["env"] = kwargs.get("env", {}) self.pid = 4242 monkeypatch.setattr("subprocess.Popen", _FakePopen) task = kb.Task( id="t_dispatch_env", title="x", body=None, assignee="coder", status="ready", priority=0, created_by=None, created_at=0, started_at=None, completed_at=None, workspace_kind="worktree", workspace_path=str(tmp_path / "ws"), claim_lock=None, claim_expires=None, tenant=None, branch_name="wt/t_dispatch_env", ) kb._default_spawn(task, str(tmp_path / "ws")) env = captured["env"] assert env["HERMES_KANBAN_DB"] == str(default_home / "kanban.db") assert env["HERMES_KANBAN_WORKSPACES_ROOT"] == str( default_home / "kanban" / "workspaces" ) assert env["HERMES_KANBAN_TASK"] == "t_dispatch_env" assert env["HERMES_KANBAN_BRANCH"] == "wt/t_dispatch_env" for key in sc._VAR_MAP: if key == "HERMES_SESSION_SOURCE": # Re-set by the dispatcher, so what matters is that it carries # the worker's own tag rather than the inherited routing value. assert env[key] == "kanban" continue assert key not in env # --------------------------------------------------------------------------- # latest_summary / latest_summaries — surface task_runs.summary handoffs # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # NFS / network-filesystem fallback (see hermes_state.apply_wal_with_fallback) # --------------------------------------------------------------------------- def test_connect_falls_back_to_delete_on_locking_protocol(tmp_path, monkeypatch, caplog): """kanban_db.connect() must handle ``locking protocol`` on NFS/SMB. Without this fallback, the gateway's kanban dispatcher crashes every 60s and the kanban migration (``consecutive_failures`` ADD COLUMN) is retried forever — which is what the real-world user report shows (see hermes-agent issue #22032). NOTE: We do NOT use the ``kanban_home`` fixture here because that fixture pre-initializes the DB via ``kb.init_db()`` — putting the file in WAL on disk. The Bug D safety guard now refuses to downgrade to DELETE when the on-disk header is already WAL, so testing the NFS-fallback path requires a truly-fresh DB file (NFS scenario in production: first connection of the first process ever to touch the file, where downgrading is safe because nobody else has WAL state yet). """ import sqlite3 as _sqlite3 from unittest.mock import patch as _patch home = tmp_path / ".hermes" home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setattr(Path, "home", lambda: tmp_path) # These tests exercise the WAL-attempt path; assume a fixed SQLite so the # WAL-reset vulnerability gate doesn't short-circuit before the pragma. import hermes_state as _hermes_state monkeypatch.setattr( _hermes_state, "is_sqlite_wal_reset_vulnerable", lambda version_info=None: False, ) _hermes_state._wal_fallback_warned_paths.clear() # Clear module cache so a fresh connect() is attempted kb._INITIALIZED_PATHS.clear() hermes_state._wal_fallback_warned_paths.clear() real_connect = _sqlite3.connect class _WalBlockingConnection(_sqlite3.Connection): def execute(self, sql, *args, **kwargs): # type: ignore[override] if "journal_mode=wal" in sql.lower().replace(" ", ""): raise _sqlite3.OperationalError("locking protocol") return super().execute(sql, *args, **kwargs) def wal_blocking_connect(*args, **kwargs): # connect_tracked passes a tracking-augmented factory; drop it and # substitute the double, which connect_tracked re-applies to the # returned instance. kwargs.pop("factory", None) return real_connect( *args, factory=_WalBlockingConnection, **kwargs ) with _patch("hermes_cli.kanban_db.sqlite3.connect", side_effect=wal_blocking_connect): with caplog.at_level("ERROR", logger="hermes_state"): conn = kb.connect() # One fallback error, naming kanban.db errors = [ r for r in caplog.records if r.levelname == "ERROR" and "kanban.db" in r.getMessage() ] assert len(errors) >= 1, ( f"Expected a kanban.db ERROR, got: {[r.getMessage() for r in caplog.records]}" ) # DB still usable end-to-end — create + list a task t = kb.create_task(conn, title="post-fallback task") tasks = kb.list_tasks(conn) assert any(row.id == t for row in tasks) conn.close() def test_connect_works_when_wal_is_silently_refused(tmp_path, monkeypatch, caplog): """kanban_db.connect() must stay usable when WAL silently no-ops to DELETE.""" import sqlite3 as _sqlite3 from unittest.mock import patch as _patch home = tmp_path / ".hermes" home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setattr(Path, "home", lambda: tmp_path) kb._INITIALIZED_PATHS.clear() hermes_state._wal_fallback_warned_paths.clear() # Assume a fixed SQLite so the WAL-reset gate doesn't short-circuit. monkeypatch.setattr( hermes_state, "is_sqlite_wal_reset_vulnerable", lambda version_info=None: False, ) real_connect = _sqlite3.connect class _WalSilentNoOpConnection(_sqlite3.Connection): def execute(self, sql, *args, **kwargs): # type: ignore[override] if "journal_mode=wal" in sql.lower().replace(" ", ""): return super().execute("PRAGMA journal_mode=delete", *args, **kwargs) return super().execute(sql, *args, **kwargs) def wal_silent_noop_connect(*args, **kwargs): kwargs.pop("factory", None) return real_connect( *args, factory=_WalSilentNoOpConnection, **kwargs ) with _patch( "hermes_cli.kanban_db.sqlite3.connect", side_effect=wal_silent_noop_connect, ): with caplog.at_level("ERROR", logger="hermes_state"): conn = kb.connect() assert conn.execute("PRAGMA journal_mode").fetchone()[0].lower() == "delete" t = kb.create_task(conn, title="post-silent-fallback task") tasks = kb.list_tasks(conn) assert any(row.id == t for row in tasks) conn.close() errors = [ r for r in caplog.records if r.levelname == "ERROR" and "kanban.db" in r.getMessage() ] assert len(errors) >= 1, ( f"Expected a kanban.db ERROR, got: {[r.getMessage() for r in caplog.records]}" ) def test_unlink_tasks_triggers_recompute_ready(kanban_home): """Regression test for issue #22459. Removing a dependency via unlink_tasks must immediately promote the child to ready when all remaining parents are done — same contract as complete_task and unblock_task. Before the fix, child stayed 'todo' indefinitely after unlink; only the next dispatcher tick or a manual 'hermes kanban recompute' would promote it. """ with kb.connect() as conn: # A is done. a = kb.create_task(conn, title="parent-done") kb.complete_task(conn, a) # C is running (not done) — blocks child B. c = kb.create_task(conn, title="parent-running") kb.claim_task(conn, c, claimer="worker:1") # B depends on both A (done) and C (running) → stays todo. b = kb.create_task(conn, title="child", parents=[a, c]) assert kb.get_task(conn, b).status == "todo" # Remove the blocking dependency C → B. removed = kb.unlink_tasks(conn, c, b) assert removed is True # B's only remaining parent is A (done) → must be ready immediately. assert kb.get_task(conn, b).status == "ready", ( "child should promote to ready immediately after unlink_tasks " "removes its last blocking dependency" ) # --------------------------------------------------------------------------- # _add_column_if_missing / _migrate_add_optional_columns idempotency (#21708) # --------------------------------------------------------------------------- def test_add_column_if_missing_is_idempotent_on_race(kanban_home): """``_add_column_if_missing`` must swallow 'duplicate column name' errors. Regression for #21708: the kanban dispatcher opens the DB twice per tick (once via _tick_once_for_board, once via init_db's discard-and-reconnect path). A second concurrent connection runs _migrate_add_optional_columns before the first one commits, so ALTER TABLE raises OperationalError with 'duplicate column name: consecutive_failures'. Without the idempotency guard that crashes the dispatcher on the first tick after every restart. """ import sqlite3 conn = sqlite3.connect(":memory:") conn.row_factory = sqlite3.Row conn.execute( "CREATE TABLE tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL)" ) # First call adds the column — returns True. added = kb._add_column_if_missing(conn, "tasks", "extra_col", "extra_col TEXT") assert added is True cols = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")} assert "extra_col" in cols # Second call on same connection — column already exists — must return # False without raising, simulating the race the dispatcher hits. added_again = kb._add_column_if_missing( conn, "tasks", "extra_col", "extra_col TEXT" ) assert added_again is False conn.close() def test_migrate_add_optional_columns_tolerates_concurrent_migration(kanban_home): """Full _migrate_add_optional_columns must not raise when columns already exist (issue #21708 race window — two connections migrate concurrently).""" import sqlite3 # Schema already in fully-migrated state (all optional columns present). conn = sqlite3.connect(":memory:") conn.row_factory = sqlite3.Row conn.execute( """ CREATE TABLE tasks ( id INTEGER PRIMARY KEY, title TEXT NOT NULL, tenant TEXT, result TEXT, idempotency_key TEXT, branch_name TEXT, consecutive_failures INTEGER NOT NULL DEFAULT 0, worker_pid INTEGER, last_failure_error TEXT, max_runtime_seconds INTEGER, last_heartbeat_at INTEGER, current_run_id INTEGER, workflow_template_id TEXT, current_step_key TEXT, skills TEXT, max_retries INTEGER, session_id TEXT ) """ ) conn.execute( """ CREATE TABLE task_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL DEFAULT '', run_id INTEGER, kind TEXT NOT NULL DEFAULT '', payload TEXT, created_at INTEGER NOT NULL DEFAULT 0 ) """ ) # Running migration on an already-migrated schema must not raise. kb._migrate_add_optional_columns(conn) conn.close() # --------------------------------------------------------------------------- # Dispatcher spawn invocation — _resolve_hermes_argv() # # Workers spawned by the dispatcher must use a `hermes` invocation that does # not depend on PATH being set up correctly. cron jobs, systemd User= services, # launchd jobs, and other detached processes routinely run with a stripped # $PATH that doesn't include the venv's bin/, so a bare `["hermes", ...]` # spawn fails with FileNotFoundError and the task gets stuck. The resolver # prefers the PATH shim (familiar `ps` output) but falls back to the module # form so the spawn keeps working when PATH is missing the shim. # --------------------------------------------------------------------------- def test_resolve_hermes_argv_falls_back_to_module_form_when_no_path_shim(monkeypatch): """When the shim is not on PATH, fall back to `python -m hermes_cli.main`. Pins the correct module name (NOT `hermes` — there is no top-level `hermes` package). Regression for #23198: the original PR shipped `python -m hermes` which fails with `No module named hermes` on every invocation. """ import shutil import sys import hermes_cli.kanban_db as kb monkeypatch.delenv("HERMES_BIN", raising=False) monkeypatch.setattr(shutil, "which", lambda name: None) argv = kb._resolve_hermes_argv() assert argv == [sys.executable, "-m", "hermes_cli.main"] def test_resolve_hermes_argv_module_actually_runs(): """The fallback module name must be importable + runnable. A unit test that pins the literal string is necessary but not sufficient — if `hermes_cli.main` ever loses `if __name__ == "__main__"` handling or its argparse setup, `python -m hermes_cli.main --version` would fail and so would every dispatcher spawn that hits the fallback. Run it as a real subprocess to catch that regression. """ import subprocess import hermes_cli.kanban_db as kb import shutil import unittest.mock as mock with mock.patch.dict(os.environ, {}, clear=False): os.environ.pop("HERMES_BIN", None) with mock.patch.object(shutil, "which", return_value=None): argv = kb._resolve_hermes_argv() r = subprocess.run(argv + ["--version"], capture_output=True, text=True, timeout=30) assert r.returncode == 0, ( f"`{' '.join(argv)} --version` failed (rc={r.returncode}); " f"stderr={r.stderr[:200]!r}" ) assert "Hermes Agent" in r.stdout, f"unexpected output: {r.stdout[:200]!r}" # --------------------------------------------------------------------------- # task_age — guard against corrupt timestamp values # # The Task dataclass declares ``created_at: int`` but rows come from sqlite # without coercion at the boundary. A row that ever held a non-int (e.g. an # unsubstituted ``'%s'`` from a logged format string, ``None``, an arbitrary # string, or a float-as-string) used to crash ``task_age`` with ``ValueError`` # and turn ``GET /api/plugins/kanban/board`` into a 500 because the dashboard # calls ``task_age`` unguarded for every task in the response. # # After the fix, ``_safe_int`` returns ``None`` on bad input and ``task_age`` # degrades gracefully (per-field ``None`` rather than a hard crash). # --------------------------------------------------------------------------- def _make_task(**overrides) -> "kb.Task": """Minimal Task with all required fields filled in. Override anything.""" defaults = dict( id="t_age", title="x", body=None, assignee=None, status="ready", priority=0, created_by=None, created_at=0, started_at=None, completed_at=None, workspace_kind="scratch", workspace_path=None, claim_lock=None, claim_expires=None, tenant=None, ) defaults.update(overrides) return kb.Task(**defaults) # --------------------------------------------------------------------------- # Board-level default_workdir # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # dispatch_once — max_in_progress # --------------------------------------------------------------------------- # Review column dispatch # --------------------------------------------------------------------------- def _set_task_status(conn: sqlite3.Connection, task_id: str, status: str) -> None: """Test helper: set a task's status directly.""" conn.execute("UPDATE tasks SET status = ? WHERE id = ?", (status, task_id)) # Stale detection — detect_stale_running # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Corruption guard (issue #30687) # --------------------------------------------------------------------------- def _write_corrupt_db(path: Path) -> bytes: """Write a kanban DB with a VALID SQLite header but malformed page content. This is the corruption shape the integrity guard specifically targets (e.g. issue #29507 follow-up reports where the file's first 16 bytes pass the header byte check but ``PRAGMA integrity_check`` then fails because the internal pages are damaged). It's what main's header-only validator was letting through, and what this PR adds the full guard for. """ # 100-byte SQLite header (magic + minimal valid-looking fields) so the # cheap header check passes, then deliberate garbage so sqlite refuses # to read the file past the header. header = b"SQLite format 3\x00" + b"\x10\x00\x02\x02\x00\x40\x20\x20" header += b"\x00\x00\x00\x0c\x00\x00\x23\x46\x00\x00\x00\x00" header = header.ljust(100, b"\x00") payload = b"definitely not a valid sqlite page \x00\x01\x02\x03" * 64 blob = header + payload path.write_bytes(blob) return blob def test_repeated_corrupt_open_reuses_single_backup(tmp_path): """Repeated quarantines of the same corrupt bytes must not amplify disk usage. Regression for the gateway dispatcher's 5-min retry loop on shared kanban DBs across multi-profile fleets: each retry on an unchanged corrupt file used to create a fresh ``.corrupt..bak`` until disk filled. The content-addressed backup name is deterministic in the DB's sha256, so N retries of the same bytes share one backup. """ db_path = tmp_path / "kanban.db" original = _write_corrupt_db(db_path) backups: set[Path] = set() for _ in range(10): kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) with pytest.raises(kb.KanbanDbCorruptError) as excinfo: kb.connect(db_path=db_path) assert excinfo.value.backup_path is not None backups.add(excinfo.value.backup_path) assert len(backups) == 1, f"expected 1 deterministic backup, got {len(backups)}" (backup,) = backups assert backup.exists() assert backup.read_bytes() == original # Mutate the corrupt bytes — fingerprint changes, separate backup preserved. with db_path.open("r+b") as f: f.seek(4096) f.write(b"\xAB" * 64) kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) with pytest.raises(kb.KanbanDbCorruptError) as excinfo2: kb.connect(db_path=db_path) second_backup = excinfo2.value.backup_path assert second_backup is not None assert second_backup != backup assert second_backup.exists() def test_locked_healthy_db_does_not_classify_as_corrupt(tmp_path, monkeypatch): """A transient lock during the probe must not produce a .corrupt backup and must not be reported as :class:`KanbanDbCorruptError`. Raw sqlite ``OperationalError`` (lock/busy) is acceptable and expected.""" db_path = tmp_path / "kanban.db" kb.init_db(db_path=db_path) kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) real_connect = sqlite3.connect def flaky_connect(*args, **kwargs): # First call is the integrity probe — simulate a lock. raise sqlite3.OperationalError("database is locked") monkeypatch.setattr(kb.sqlite3, "connect", flaky_connect) with pytest.raises(sqlite3.OperationalError): kb.connect(db_path=db_path) # No .corrupt backup may be produced for a healthy-but-locked DB. backups = list(tmp_path.glob("*.corrupt.*")) assert backups == [], f"unexpected corrupt backups: {backups}" # And once the lock clears, normal access still works. monkeypatch.setattr(kb.sqlite3, "connect", real_connect) with kb.connect(db_path=db_path) as conn: kb.create_task(conn, title="still here") titles = [t.title for t in kb.list_tasks(conn)] assert "still here" in titles # --------------------------------------------------------------------------- # First-use tip for scratch workspaces # --------------------------------------------------------------------------- def test_maybe_emit_scratch_tip_fires_once_per_install(kanban_home, caplog): """First scratch workspace materialization warns + emits an event. Subsequent scratch workspaces on the SAME install stay silent — the sentinel file under kanban_home() flips after the first emit. """ import logging with kb.connect() as conn: t1 = kb.create_task(conn, title="first scratch") t2 = kb.create_task(conn, title="second scratch") # Sentinel must not exist yet on a fresh install. assert not kb._scratch_tip_shown() with caplog.at_level(logging.WARNING, logger="hermes_cli.kanban_db"): with kb.connect() as conn: kb._maybe_emit_scratch_tip(conn, t1, "scratch") # Sentinel is now set. assert kb._scratch_tip_shown() assert kb._scratch_tip_sentinel_path().exists() # Warning was logged exactly once. tip_records = [ r for r in caplog.records if "scratch workspaces are ephemeral" in r.getMessage() ] assert len(tip_records) == 1, ( f"Expected exactly one tip warning, got {len(tip_records)}: " f"{[r.getMessage() for r in tip_records]!r}" ) # An event row was appended on the first task. with kb.connect() as conn: events = conn.execute( "SELECT kind FROM task_events WHERE task_id = ? ORDER BY id", (t1,), ).fetchall() kinds = [e["kind"] for e in events] assert "tip_scratch_workspace" in kinds, ( f"Expected tip_scratch_workspace event on first scratch task; " f"got {kinds!r}" ) # Second scratch materialization on the same install stays silent. caplog.clear() with caplog.at_level(logging.WARNING, logger="hermes_cli.kanban_db"): with kb.connect() as conn: kb._maybe_emit_scratch_tip(conn, t2, "scratch") tip_records2 = [ r for r in caplog.records if "scratch workspaces are ephemeral" in r.getMessage() ] assert tip_records2 == [], ( f"Tip should not re-fire after sentinel is set; got " f"{[r.getMessage() for r in tip_records2]!r}" ) with kb.connect() as conn: events2 = conn.execute( "SELECT kind FROM task_events WHERE task_id = ? ORDER BY id", (t2,), ).fetchall() assert "tip_scratch_workspace" not in [e["kind"] for e in events2], ( "Tip event should not be appended for subsequent scratch tasks." ) # --------------------------------------------------------------------------- # Connection pragmas (secure_delete, cell_size_check, synchronous=FULL) # --------------------------------------------------------------------------- def test_connect_sets_secure_delete_on(tmp_path): """secure_delete=ON must be active on every new connection.""" db_path = tmp_path / "kanban.db" kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) with kb.connect(db_path=db_path) as conn: row = conn.execute("PRAGMA secure_delete").fetchone() assert row[0] == 1, f"expected secure_delete=1, got {row[0]}" # write_txn — rollback handler must not mask the original exception # --------------------------------------------------------------------------- def test_write_txn_preserves_original_exception_when_rollback_fails(kanban_home): """When a write inside write_txn raises an OperationalError that SQLite has already auto-rolled-back (e.g. ``disk I/O error``, ``database is locked``, ``database disk image is malformed``), the explicit ROLLBACK in ``write_txn.__exit__`` itself raises ``cannot rollback - no transaction is active``. The original cause must NOT be masked by the secondary rollback failure — operators rely on the original cause to diagnose the underlying issue. """ class FailingConnWrapper: """Delegate to a real connection, simulating an EIO during an INSERT that SQLite has already auto-rolled-back.""" def __init__(self, real): self._real = real self._fail_armed = True def execute(self, sql, *args, **kwargs): if ( self._fail_armed and sql.lstrip().upper().startswith("INSERT") and "task_events" in sql.lower() ): self._fail_armed = False # one-shot # Simulate SQLite auto-rolling back the transaction by # issuing a real ROLLBACK now. After this, BEGIN IMMEDIATE # is no longer active and an explicit ROLLBACK would error. try: self._real.execute("ROLLBACK") except sqlite3.OperationalError: pass raise sqlite3.OperationalError("disk I/O error") return self._real.execute(sql, *args, **kwargs) def __getattr__(self, name): return getattr(self._real, name) with kb.connect() as conn: wrapper = FailingConnWrapper(conn) with pytest.raises(sqlite3.OperationalError) as excinfo: with kb.write_txn(wrapper): kb._append_event(wrapper, "t_bogus", "promoted", None) msg = str(excinfo.value) assert "disk I/O error" in msg, ( f"write_txn masked the original exception with rollback failure; " f"got {msg!r} (expected to contain 'disk I/O error')" ) assert "cannot rollback" not in msg, ( f"write_txn surfaced the rollback failure instead of the original " f"OperationalError; got {msg!r}" ) def test_write_txn_check_reads_correct_header_fields(tmp_path): """A genuinely truncated DB is never reported as passing the invariant. The check no longer opens the database file to read header bytes (that open/close would cancel this process's POSIX advisory locks — the corruption route in sqlite.org/howtocorrupt.html §2.2). It asks SQLite for ``page_count`` instead. On a truncated file SQLite refuses that pragma, so the helper reports "not healthy" rather than a page-count mismatch; either way the file must never come back clean. """ import struct from hermes_cli.kanban_db import connect from hermes_cli.sqlite_safe_read import file_length_matches_header db = tmp_path / "synthetic.db" conn = connect(db_path=db) conn.execute("PRAGMA journal_mode=DELETE") page_size = conn.execute("PRAGMA page_size").fetchone()[0] conn.close() with open(db, "rb") as f: data = bytearray(f.read()) real_page_count = struct.unpack(">I", data[28:32])[0] if real_page_count < 2: pytest.skip("DB too small for synthetic truncation test") truncated = bytes(data[: (real_page_count - 1) * page_size]) with open(db, "wb") as f: f.write(truncated) raw_conn = sqlite3.connect(str(db), isolation_level=None) try: assert file_length_matches_header(raw_conn) is not True finally: raw_conn.close() # --------------------------------------------------------------------------- # reap_worker_zombies() tests # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # connect_closing(): context manager that actually closes the FD # Regression coverage for #33159 (kanban.db FD leak — gateway crashes after # ~4 days). sqlite3.Connection's built-in __exit__ commits/rollbacks but # does NOT close, so `with kb.connect() as conn:` leaks the FD in # long-lived processes (gateway run_slash, dashboard decompose handler). # `connect_closing()` is the leak-safe replacement. # --------------------------------------------------------------------------- def test_bare_connect_does_not_close_on_context_exit(tmp_path): """Document the leak that connect_closing exists to prevent. sqlite3.Connection's __exit__ commits/rollbacks but doesn't close. This is the upstream behaviour we cannot change; the regression guard is to make sure connect_closing() does the right thing. """ db_path = tmp_path / "kanban.db" kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) with kb.connect(db_path=db_path) as conn: pass # Still usable after with-block exit (the leak). conn.execute("SELECT 1").fetchone() conn.close() # explicit close to avoid leaking THIS test