fix: make kanban archive and blocked holds sticky

hermes_cli/kanban_db.py

@@ -267,7 +267,7 @@ def get_current_board() -> str:
     if env:
         try:
             normed = _normalize_board_slug(env)
-            if normed and board_exists(normed):
+            if normed and board_exists(normed) and not read_board_metadata(normed).get("archived"):
                 return normed
         except ValueError:
             pass
@@ -278,7 +278,7 @@ def get_current_board() -> str:
             if val:
                 try:
                     normed = _normalize_board_slug(val)
-                    if normed and board_exists(normed):
+                    if normed and board_exists(normed) and not read_board_metadata(normed).get("archived"):
                         return normed
                 except ValueError:
                     pass
@@ -337,7 +337,15 @@ def board_exists(board: Optional[str] = None) -> bool:
     if slug == DEFAULT_BOARD:
         return True
     d = board_dir(slug)
-    return (d / "board.json").exists() or (d / "kanban.db").exists()
+    meta_path = d / "board.json"
+    if meta_path.exists():
+        try:
+            raw = json.loads(meta_path.read_text(encoding="utf-8"))
+            if isinstance(raw, dict) and raw.get("archived"):
+                return False
+        except (OSError, json.JSONDecodeError):
+            pass
+    return meta_path.exists() or (d / "kanban.db").exists()
 
 
 def kanban_db_path(board: Optional[str] = None) -> Path:
@@ -526,21 +534,26 @@ def create_board(
         description=description,
         icon=icon,
         color=color,
+        archived=False,
         default_workdir=default_workdir,
     )
     # Touch the DB so list_boards() sees it immediately.
     init_db(board=normed)
     return meta
 
 
-def list_boards(*, include_archived: bool = True) -> list[dict]:
-    """Enumerate all boards that exist on disk.
+def list_boards(*, include_archived: bool = False) -> list[dict]:
+    """Enumerate boards that exist on disk.
 
     Always includes ``default`` (even when the ``boards/default/``
     metadata dir doesn't exist, because its DB is at the legacy path).
     Other boards are discovered by scanning ``boards/`` for subdirectories
     that either contain a ``kanban.db`` or a ``board.json``.
 
+    Archived tombstones are hidden by default so active-board views stay
+    clean after ``remove_board(archive=True)``. Pass ``include_archived=True``
+    to include those tombstones for recovery/debugging.
+
     Returns a list of metadata dicts, sorted with ``default`` first and
     the rest alphabetically.
     """
@@ -597,6 +610,8 @@ def remove_board(slug: str, *, archive: bool = True) -> dict:
     if not d.exists():
         raise ValueError(f"board {normed!r} does not exist")
 
+    original_meta = read_board_metadata(normed)
+
     # If the user removed the currently-active board, revert to default.
     if get_current_board() == normed:
         clear_current_board()
@@ -616,12 +631,46 @@ def remove_board(slug: str, *, archive: bool = True) -> dict:
         while target.exists():
             target = archive_root / f"{normed}-{ts}-{suffix}"
             suffix += 1
-        d.rename(target)
+        for attempt in range(5):
+            try:
+                d.rename(target)
+                break
+            except PermissionError:
+                if attempt == 4:
+                    raise
+                import gc
+                gc.collect()
+                time.sleep(0.2)
+
+        # Leave an archived tombstone at the original path. Without this,
+        # a gateway dispatcher/notifier tick that enumerated the board just
+        # before the rename can still call connect(board=slug) afterward,
+        # which recreates an empty active board directory/DB. The tombstone
+        # lets that stale connect happen harmlessly while list_boards(False)
+        # keeps the board out of active views on the next tick.
+        tombstone = {
+            "slug": normed,
+            "name": original_meta.get("name") or _default_board_display_name(normed),
+            "description": original_meta.get("description", ""),
+            "icon": original_meta.get("icon", ""),
+            "color": original_meta.get("color", ""),
+            "default_workdir": original_meta.get("default_workdir"),
+            "created_at": original_meta.get("created_at"),
+            "archived": True,
+            "archived_at": ts,
+            "archive_path": str(target),
+        }
+        tombstone_path = board_metadata_path(normed)
+        tombstone_path.parent.mkdir(parents=True, exist_ok=True)
+        tombstone_path.write_text(
+            json.dumps(tombstone, indent=2, ensure_ascii=False) + "\n",
+            encoding="utf-8",
+        )
         return {"slug": normed, "action": "archived", "new_path": str(target)}
-    else:
-        import shutil
-        shutil.rmtree(d)
-        return {"slug": normed, "action": "deleted", "new_path": ""}
+
+    import shutil
+    shutil.rmtree(d)
+    return {"slug": normed, "action": "deleted", "new_path": ""}
 
 
 # ---------------------------------------------------------------------------
@@ -1965,6 +2014,23 @@ def create_task(
                         "skills": list(skills_list) if skills_list else None,
                     },
                 )
+                if task_status == "blocked":
+                    # A task explicitly created as blocked is an operator /
+                    # workflow hold, not a transient circuit-breaker state.
+                    # Emit the same event that block_task() uses so
+                    # recompute_ready() treats it as sticky instead of
+                    # auto-promoting parentless "last task on the board"
+                    # holds back to ready on the next dispatcher tick.
+                    _append_event(
+                        conn,
+                        task_id,
+                        "blocked",
+                        {
+                            "reason": "initial_status=blocked",
+                            "source": "create_task",
+                            "parents": list(parents),
+                        },
+                    )
             return task_id
         except sqlite3.IntegrityError:
             if attempt == 1:
@@ -1974,6 +2040,7 @@ def create_task(
     raise RuntimeError("unreachable")
 
 
+
 def _find_missing_parents(conn: sqlite3.Connection, parents: Iterable[str]) -> list[str]:
     parents = list(parents)
     if not parents:

tests/hermes_cli/test_kanban_blocked_sticky.py

@@ -48,6 +48,23 @@ def kanban_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
     return home
 
 
+def test_task_created_blocked_is_sticky_without_parent_workaround(kanban_home: Path) -> None:
+    """Creating a standalone blocked task is an explicit operator hold.
+
+    This covers the Ask-Hermes/Codex workaround Matthew saw: when the
+    final remaining item on a board was created/parked as blocked, the
+    dispatcher treated it like a parentless dependency block and promoted
+    it back to ready on the next tick.
+    """
+    with kb.connect() as conn:
+        tid = kb.create_task(conn, title="hold for Matthew", initial_status="blocked")
+        assert kb.get_task(conn, tid).status == "blocked"
+
+        for _ in range(3):
+            assert kb.recompute_ready(conn) == 0
+            assert kb.get_task(conn, tid).status == "blocked"
+
+
 # ---------------------------------------------------------------------------
 # Worker-initiated kanban_block must be sticky
 # ---------------------------------------------------------------------------

tests/hermes_cli/test_kanban_boards.py

@@ -249,6 +249,33 @@ def test_remove_archive(self, fresh_home):
         assert Path(res["new_path"]).exists()
         assert "toremove" not in [b["slug"] for b in kb.list_boards()]
 
+    def test_remove_archive_tombstone_prevents_stale_dispatch_recreate(self, fresh_home):
+        """A stale gateway tick may connect to a board after archive.
+
+        The archived tombstone at the original path must keep that harmless
+        connect/init_db from resurrecting an empty active board.
+        """
+        kb.create_board("race-board")
+        res = kb.remove_board("race-board")
+        assert res["action"] == "archived"
+        assert "race-board" not in [b["slug"] for b in kb.list_boards(include_archived=False)]
+
+        with kb.connect(board="race-board"):
+            pass
+
+        assert "race-board" not in [b["slug"] for b in kb.list_boards(include_archived=False)]
+        archived = [b for b in kb.list_boards(include_archived=True) if b["slug"] == "race-board"]
+        assert archived and archived[0]["archived"] is True
+
+        kb.create_board("race-board")
+        assert "race-board" in [b["slug"] for b in kb.list_boards(include_archived=False)]
+
+    def test_archived_current_board_pointer_falls_back_to_default(self, fresh_home):
+        kb.create_board("old-board")
+        kb.set_current_board("old-board")
+        kb.remove_board("old-board")
+        assert kb.get_current_board() == "default"
+
     def test_remove_hard_delete(self, fresh_home):
         kb.create_board("nuke")
         d = kb.board_dir("nuke")
@@ -280,7 +307,7 @@ def test_remove_clears_init_cache_for_recreated_db(self, fresh_home, archive):
         # downstream readers hit `no such table: task_events`.
         kb.create_board("recycle")
         # First connect populates _INITIALIZED_PATHS for this DB.
-        with kb.connect(board="recycle") as conn:
+        with kb.connect_closing(board="recycle") as conn:
             kb.create_task(conn, title="t1", assignee="dev")
         db_path = kb.board_dir("recycle") / "kanban.db"
         assert str(db_path.resolve()) in kb._INITIALIZED_PATHS