fix(kanban): count board list entries through read-only DB paths

hermes_cli/kanban.py

@@ -19,6 +19,7 @@
 import json
 import os
 import shlex
+import sqlite3
 import sys
 import time
 from pathlib import Path
@@ -1205,18 +1206,59 @@ def _dispatch_boards(args: argparse.Namespace) -> int:
     return 2
 
 
-def _board_task_counts(slug: str) -> dict[str, int]:
-    """Return ``{status: count}`` for a board. Safe to call on an empty DB."""
+def _board_list_db_path(slug: str) -> Path:
+    """Return a board's physical DB path for registry listing.
+
+    Named boards always use their on-disk path under ``boards/<slug>/``.
+    The default board honors a legitimate ``HERMES_KANBAN_DB`` relocation
+    (single-DB deployments / tests), but ignores a worker handoff pin that
+    points at another board's database — otherwise every listed board
+    collapses onto the pinned file.
+    """
+    if slug != kb.DEFAULT_BOARD:
+        return kb.board_dir(slug) / "kanban.db"
+
+    canonical = kb.kanban_home() / "kanban.db"
+    override = os.environ.get("HERMES_KANBAN_DB", "").strip()
+    if not override:
+        return canonical
+
+    override_path = Path(override).expanduser()
+    try:
+        resolved = override_path.resolve()
+    except OSError:
+        return override_path
+
+    # Worker pins inject the active board DB.  If that path is clearly
+    # another board under boards_root, keep default on its canonical file.
     try:
-        path = kb.kanban_db_path(board=slug)
+        rel = resolved.relative_to(kb.boards_root().resolve())
+    except ValueError:
+        # Outside the boards tree — treat as intentional default relocation.
+        return override_path
+    except OSError:
+        return override_path
+
+    if rel.parts and rel.parts[0] not in {"", kb.DEFAULT_BOARD}:
+        return canonical
+    return override_path
+
+
+def _board_task_counts(
+    slug: str, *, db_path: str | Path | None = None
+) -> dict[str, int]:
+    """Return ``{status: count}`` without entering the write/migration path."""
+    try:
+        path = Path(db_path) if db_path is not None else _board_list_db_path(slug)
         if not path.exists():
             return {}
-        with kb.connect_closing(board=slug) as conn:
+        uri = path.resolve().as_uri() + "?mode=ro"
+        with contextlib.closing(sqlite3.connect(uri, uri=True)) as conn:
             rows = conn.execute(
                 "SELECT status, COUNT(*) AS n FROM tasks GROUP BY status"
             ).fetchall()
-        return {r["status"]: int(r["n"]) for r in rows}
-    except Exception:
+        return {str(status): int(count) for status, count in rows}
+    except (OSError, sqlite3.Error):
         return {}
 
 
@@ -1227,7 +1269,12 @@ def _cmd_boards_list(args: argparse.Namespace) -> int:
     current = kb.get_current_board()
     for b in boards:
         b["is_current"] = (b["slug"] == current)
-        b["counts"] = _board_task_counts(b["slug"])
+        path = _board_list_db_path(b["slug"])
+        # list_boards() metadata resolves through kanban_db_path(), whose
+        # worker handoff override intentionally pins one DB.  A registry list
+        # must instead expose and count each board's canonical physical path.
+        b["db_path"] = str(path)
+        b["counts"] = _board_task_counts(b["slug"], db_path=path)
         b["total"] = sum(b["counts"].values())
     if getattr(args, "json", False):
         print(json.dumps(boards, indent=2, ensure_ascii=False))

tests/hermes_cli/test_kanban_boards.py

@@ -15,12 +15,15 @@
 
 from __future__ import annotations
 
+import argparse
 import json
 import os
+import sqlite3
 import subprocess
 import sys
 from pathlib import Path
 
+
 import pytest
 
 # Ensure the worktree (not the stale global clone) is first on sys.path.
@@ -308,6 +311,100 @@ def _cli(args: list[str], env_extra: dict | None = None) -> subprocess.Completed
 
 
 class TestCLI:
+    def test_board_counts_are_read_only_in_delegated_context(self, tmp_path, monkeypatch):
+        """Listing counts must not enter the schema migration/write path."""
+        from hermes_cli import kanban as kanban_cli
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        path = tmp_path / "kanban.db"
+        with sqlite3.connect(path) as conn:
+            conn.execute("CREATE TABLE tasks (status TEXT NOT NULL)")
+            conn.executemany(
+                "INSERT INTO tasks(status) VALUES (?)",
+                [("running",), ("blocked",), ("blocked",)],
+            )
+
+        def reject_migration_path(*args, **kwargs):
+            raise PermissionError("delegated contexts cannot mutate Kanban")
+
+        monkeypatch.setattr(kb, "connect_closing", reject_migration_path)
+
+        assert kanban_cli._board_task_counts("default") == {
+            "blocked": 2,
+            "running": 1,
+        }
+
+    def test_boards_list_counts_each_db_despite_worker_db_pin(
+        self, tmp_path, monkeypatch, capsys
+    ):
+        """A worker's DB pin must not make every listed board show one DB."""
+        from hermes_cli import kanban as kanban_cli
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        paths = []
+        for slug, statuses in (
+            ("default", ["todo"]),
+            ("alpha", ["running"]),
+            ("beta", ["blocked", "blocked"]),
+        ):
+            path = (
+                tmp_path / "kanban.db"
+                if slug == "default"
+                else tmp_path / "kanban" / "boards" / slug / "kanban.db"
+            )
+            path.parent.mkdir(parents=True, exist_ok=True)
+            with sqlite3.connect(path) as conn:
+                conn.execute("CREATE TABLE tasks (status TEXT NOT NULL)")
+                conn.executemany(
+                    "INSERT INTO tasks(status) VALUES (?)",
+                    [(status,) for status in statuses],
+                )
+            paths.append((slug, path))
+
+        monkeypatch.setenv("HERMES_KANBAN_DB", str(paths[0][1]))
+        monkeypatch.setattr(kb, "get_current_board", lambda: "alpha")
+
+        assert kanban_cli._cmd_boards_list(argparse.Namespace(all=False, json=True)) == 0
+        data = json.loads(capsys.readouterr().out)
+        by_slug = {board["slug"]: board for board in data}
+        assert by_slug["default"]["counts"] == {"todo": 1}
+        assert by_slug["alpha"]["counts"] == {"running": 1}
+        assert by_slug["beta"]["counts"] == {"blocked": 2}
+        assert {board["db_path"] for board in data} == {
+            str(path) for _, path in paths
+        }
+
+    def test_boards_list_honors_custom_default_db_override(
+        self, tmp_path, monkeypatch, capsys
+    ):
+        """HERMES_KANBAN_DB may relocate the default board without pinning others."""
+        from hermes_cli import kanban as kanban_cli
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        custom = tmp_path / "custom-default.db"
+        with sqlite3.connect(custom) as conn:
+            conn.execute("CREATE TABLE tasks (status TEXT NOT NULL)")
+            conn.executemany(
+                "INSERT INTO tasks(status) VALUES (?)",
+                [("todo",), ("running",)],
+            )
+        other = tmp_path / "kanban" / "boards" / "alpha" / "kanban.db"
+        other.parent.mkdir(parents=True)
+        with sqlite3.connect(other) as conn:
+            conn.execute("CREATE TABLE tasks (status TEXT NOT NULL)")
+            conn.execute("INSERT INTO tasks(status) VALUES (?)", ("blocked",))
+
+        monkeypatch.setenv("HERMES_KANBAN_DB", str(custom))
+        monkeypatch.setattr(kb, "get_current_board", lambda: "default")
+
+        assert kanban_cli._cmd_boards_list(argparse.Namespace(all=False, json=True)) == 0
+        data = json.loads(capsys.readouterr().out)
+        by_slug = {board["slug"]: board for board in data}
+        assert by_slug["default"]["db_path"] == str(custom)
+        assert by_slug["default"]["counts"] == {"todo": 1, "running": 1}
+        assert by_slug["alpha"]["db_path"] == str(other)
+        assert by_slug["alpha"]["counts"] == {"blocked": 1}
+
     def test_boards_list_default_only(self, tmp_path):
         env = {"HERMES_HOME": str(tmp_path)}
         res = _cli(["boards", "list", "--json"], env_extra=env)