feat(kanban): durable multi-profile collaboration board (#17805)

Salvage of PR #16100 onto current main (after emozilla's #17514 fix
that unblocks plugin Pydantic body validation). History preserved on
the standing `feat/kanban-standing` branch; this squashes the 22
iterative commits into one clean landing.

What this lands:
- SQLite kernel (hermes_cli/kanban_db.py) — durable task board with
  tasks, task_links, task_runs, task_comments, task_events,
  kanban_notify_subs tables. WAL mode, atomic claim via CAS,
  tenant-namespaced, skills JSON array per task, max-runtime timeouts,
  worker heartbeats, idempotency keys, circuit breaker on repeated
  spawn failures, crash detection via /proc/<pid>/status, run history
  preserved across attempts.
- Dispatcher — runs inside the gateway by default
  (`kanban.dispatch_in_gateway: true`). Ticks every 60s, reclaims
  stale claims, promotes ready tasks, spawns `hermes -p <assignee>
  chat -q "work kanban task <id>"` with HERMES_KANBAN_TASK +
  HERMES_KANBAN_WORKSPACE env. Auto-loads `--skills kanban-worker`
  plus any per-task skills. Health telemetry warns on stuck ready
  queue.
- Structured tool surface (tools/kanban_tools.py) — 7 tools
  (kanban_show, kanban_complete, kanban_block, kanban_heartbeat,
  kanban_comment, kanban_create, kanban_link). Gated on
  HERMES_KANBAN_TASK via check_fn so zero schema footprint in normal
  sessions.
- System-prompt guidance (agent/prompt_builder.py KANBAN_GUIDANCE)
  injected only when kanban tools are active.
- Dashboard plugin (plugins/kanban/dashboard/) — Linear-style board
  UI: triage/todo/ready/running/blocked/done columns, drag-drop,
  inline create, task drawer with markdown, comments, run history,
  dependency editor, bulk ops, lanes-by-profile grouping, WS-driven
  live refresh. Matches active dashboard theme via CSS variables.
- CLI — `hermes kanban init|create|list|show|assign|link|unlink|
  claim|comment|complete|block|unblock|archive|tail|dispatch|context|
  init|gc|watch|stats|notify|log|heartbeat|runs|assignees` +
  `/kanban` slash in-session.
- Worker + orchestrator skills (skills/devops/kanban-worker +
  kanban-orchestrator) — pattern library for good summary/metadata
  shapes, retry diagnostics, block-reason examples, fan-out patterns.
- Per-task force-loaded skills — `--skill <name>` (repeatable),
  stored as JSON, threaded through to dispatcher argv as one
  `--skills X` pair per skill alongside the built-in kanban-worker.
  Dashboard + CLI + tool parity.
- Deprecation of standalone `hermes kanban daemon` — stub exits 2
  with migration guidance; `--force` escape hatch for headless hosts.
- Docs (website/docs/user-guide/features/kanban.md + kanban-tutorial.md)
  with 11 dashboard screenshots walking through four user stories
  (Solo Dev, Fleet Farming, Role Pipeline, Circuit Breaker).
- Tests (251 passing): kernel schema + migration + CAS atomicity,
  dispatcher logic, circuit breaker, crash detection, max-runtime
  timeouts, claim lifecycle, tenant isolation, idempotency keys, per-
  task skills round-trip + validation + dispatcher argv, tool surface
  (7 tools × round-trip + error paths), dashboard REST (CRUD + bulk
  + links + warnings), gateway-embedded dispatcher (config gate, env
  override, graceful shutdown), CLI deprecation stub, migration from
  legacy schemas.

Gateway integration:
- GatewayRunner._kanban_dispatcher_watcher — new asyncio background
  task, symmetric with _kanban_notifier_watcher. Runs dispatch_once
  via asyncio.to_thread so SQLite WAL never blocks the loop. Sleeps
  in 1s slices for snappy shutdown. Respects HERMES_KANBAN_DISPATCH_IN_GATEWAY=0
  env override for debugging.
- Config: new `kanban` section in DEFAULT_CONFIG with
  `dispatch_in_gateway: true` (default) + `dispatch_interval_seconds: 60`.
  Additive — no \_config_version bump needed.

Forward-compat:
- workflow_template_id / current_step_key columns on tasks (v1 writes
  NULL; v2 will use them for routing).
- task_runs holds claim machinery (claim_lock, claim_expires,
  worker_pid, last_heartbeat_at) so multi-attempt history is first-
  class from day one.

Closes #16102.

Co-authored-by: emozilla <emozilla@nousresearch.com>

tests/hermes_cli/test_kanban_cli.py

@@ -0,0 +1,210 @@
+"""Tests for the kanban CLI surface (hermes_cli.kanban)."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import kanban as kc
+from hermes_cli import kanban_db as kb
+
+
+@pytest.fixture
+def kanban_home(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    kb.init_db()
+    return home
+
+
+# ---------------------------------------------------------------------------
+# Workspace flag parsing
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        ("scratch",              ("scratch", None)),
+        ("worktree",              ("worktree", None)),
+        ("dir:/tmp/work",         ("dir", "/tmp/work")),
+    ],
+)
+def test_parse_workspace_flag_valid(value, expected):
+    assert kc._parse_workspace_flag(value) == expected
+
+
+def test_parse_workspace_flag_expands_user():
+    kind, path = kc._parse_workspace_flag("dir:~/vault")
+    assert kind == "dir"
+    assert path.endswith("/vault")
+    assert not path.startswith("~")
+
+
+@pytest.mark.parametrize("bad", ["cloud", "dir:", "", "worktree:/x"])
+def test_parse_workspace_flag_rejects(bad):
+    if not bad:
+        # Empty -> defaults; not an error.
+        assert kc._parse_workspace_flag(bad) == ("scratch", None)
+        return
+    with pytest.raises(argparse.ArgumentTypeError):
+        kc._parse_workspace_flag(bad)
+
+
+# ---------------------------------------------------------------------------
+# run_slash smoke tests (end-to-end via the same entry both CLI and gateway use)
+# ---------------------------------------------------------------------------
+
+def test_run_slash_no_args_shows_usage(kanban_home):
+    out = kc.run_slash("")
+    assert "kanban" in out.lower()
+    assert "create" in out.lower() or "subcommand" in out.lower() or "action" in out.lower()
+
+
+def test_run_slash_create_and_list(kanban_home):
+    out = kc.run_slash("create 'ship feature' --assignee alice")
+    assert "Created" in out
+    out = kc.run_slash("list")
+    assert "ship feature" in out
+    assert "alice" in out
+
+
+def test_run_slash_create_with_parent_and_cascade(kanban_home):
+    # Parent then child via --parent
+    out1 = kc.run_slash("create 'parent' --assignee alice")
+    # Extract the "t_xxxx" id from "Created t_xxxx (ready, ...)"
+    import re
+    m = re.search(r"(t_[a-f0-9]+)", out1)
+    assert m
+    p = m.group(1)
+    out2 = kc.run_slash(f"create 'child' --assignee bob --parent {p}")
+    assert "todo" in out2  # child starts as todo
+
+    # Complete parent; list should promote child to ready
+    kc.run_slash(f"complete {p}")
+    # Explicit filter: child should now be ready (was todo before complete).
+    ready_list = kc.run_slash("list --status ready")
+    assert "child" in ready_list
+
+
+def test_run_slash_show_includes_comments(kanban_home):
+    out = kc.run_slash("create 'x'")
+    import re
+    tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
+    kc.run_slash(f"comment {tid} 'source is paywalled'")
+    show = kc.run_slash(f"show {tid}")
+    assert "source is paywalled" in show
+
+
+def test_run_slash_block_unblock_cycle(kanban_home):
+    out = kc.run_slash("create 'x' --assignee alice")
+    import re
+    tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
+    # Claim first so block() finds it running
+    kc.run_slash(f"claim {tid}")
+    assert "Blocked" in kc.run_slash(f"block {tid} 'need decision'")
+    assert "Unblocked" in kc.run_slash(f"unblock {tid}")
+
+
+def test_run_slash_json_output(kanban_home):
+    out = kc.run_slash("create 'jsontask' --assignee alice --json")
+    payload = json.loads(out)
+    assert payload["title"] == "jsontask"
+    assert payload["assignee"] == "alice"
+    assert payload["status"] == "ready"
+
+
+def test_run_slash_dispatch_dry_run_counts(kanban_home):
+    kc.run_slash("create 'a' --assignee alice")
+    kc.run_slash("create 'b' --assignee bob")
+    out = kc.run_slash("dispatch --dry-run")
+    assert "Spawned:" in out
+
+
+def test_run_slash_context_output_format(kanban_home):
+    out = kc.run_slash("create 'tech spec' --assignee alice --body 'write an RFC'")
+    import re
+    tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
+    kc.run_slash(f"comment {tid} 'remember to include performance section'")
+    ctx = kc.run_slash(f"context {tid}")
+    assert "tech spec" in ctx
+    assert "write an RFC" in ctx
+    assert "performance section" in ctx
+
+
+def test_run_slash_tenant_filter(kanban_home):
+    kc.run_slash("create 'biz-a task' --tenant biz-a --assignee alice")
+    kc.run_slash("create 'biz-b task' --tenant biz-b --assignee alice")
+    a = kc.run_slash("list --tenant biz-a")
+    b = kc.run_slash("list --tenant biz-b")
+    assert "biz-a task" in a and "biz-b task" not in a
+    assert "biz-b task" in b and "biz-a task" not in b
+
+
+def test_run_slash_usage_error_returns_message(kanban_home):
+    # Missing required argument for create
+    out = kc.run_slash("create")
+    assert "usage" in out.lower() or "error" in out.lower()
+
+
+def test_run_slash_assign_reassigns(kanban_home):
+    out = kc.run_slash("create 'x' --assignee alice")
+    import re
+    tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
+    assert "Assigned" in kc.run_slash(f"assign {tid} bob")
+    show = kc.run_slash(f"show {tid}")
+    assert "bob" in show
+
+
+def test_run_slash_link_unlink(kanban_home):
+    a = kc.run_slash("create 'a'")
+    b = kc.run_slash("create 'b'")
+    import re
+    ta = re.search(r"(t_[a-f0-9]+)", a).group(1)
+    tb = re.search(r"(t_[a-f0-9]+)", b).group(1)
+    assert "Linked" in kc.run_slash(f"link {ta} {tb}")
+    # After link, b is todo
+    show = kc.run_slash(f"show {tb}")
+    assert "todo" in show
+    assert "Unlinked" in kc.run_slash(f"unlink {ta} {tb}")
+
+
+# ---------------------------------------------------------------------------
+# Integration with the COMMAND_REGISTRY
+# ---------------------------------------------------------------------------
+
+def test_kanban_is_resolvable():
+    from hermes_cli.commands import resolve_command
+
+    cmd = resolve_command("kanban")
+    assert cmd is not None
+    assert cmd.name == "kanban"
+
+
+def test_kanban_bypasses_active_session_guard():
+    from hermes_cli.commands import should_bypass_active_session
+
+    assert should_bypass_active_session("kanban")
+
+
+def test_kanban_in_autocomplete_table():
+    from hermes_cli.commands import COMMANDS, SUBCOMMANDS
+
+    assert "/kanban" in COMMANDS
+    subs = SUBCOMMANDS.get("/kanban") or []
+    assert "create" in subs
+    assert "dispatch" in subs
+
+
+def test_kanban_not_gateway_only():
+    # kanban is available in BOTH CLI and gateway surfaces.
+    from hermes_cli.commands import COMMAND_REGISTRY
+
+    cmd = next(c for c in COMMAND_REGISTRY if c.name == "kanban")
+    assert not cmd.cli_only
+    assert not cmd.gateway_only

tests/hermes_cli/test_kanban_db.py

@@ -0,0 +1,438 @@
+"""Tests for the Kanban DB layer (hermes_cli.kanban_db)."""
+
+from __future__ import annotations
+
+import concurrent.futures
+import os
+import time
+from pathlib import Path
+
+import pytest
+
+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
+
+
+# ---------------------------------------------------------------------------
+# Schema / init
+# ---------------------------------------------------------------------------
+
+def test_init_db_is_idempotent(kanban_home):
+    # Second call should not error or drop data.
+    with kb.connect() as conn:
+        kb.create_task(conn, title="persisted")
+    kb.init_db()
+    with kb.connect() as conn:
+        tasks = kb.list_tasks(conn)
+    assert len(tasks) == 1
+    assert tasks[0].title == "persisted"
+
+
+def test_init_creates_expected_tables(kanban_home):
+    with kb.connect() as conn:
+        rows = conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
+        ).fetchall()
+    names = {r["name"] for r in rows}
+    assert {"tasks", "task_links", "task_comments", "task_events"} <= names
+
+
+# ---------------------------------------------------------------------------
+# Task creation + status inference
+# ---------------------------------------------------------------------------
+
+def test_create_task_no_parents_is_ready(kanban_home):
+    with kb.connect() as conn:
+        tid = kb.create_task(conn, title="ship it", assignee="alice")
+        t = kb.get_task(conn, tid)
+    assert t is not None
+    assert t.status == "ready"
+    assert t.assignee == "alice"
+    assert t.workspace_kind == "scratch"
+
+
+def test_create_task_with_parent_is_todo_until_parent_done(kanban_home):
+    with kb.connect() as conn:
+        p = kb.create_task(conn, title="parent")
+        c = kb.create_task(conn, title="child", parents=[p])
+        assert kb.get_task(conn, c).status == "todo"
+        kb.complete_task(conn, p, result="ok")
+        assert kb.get_task(conn, c).status == "ready"
+
+
+def test_create_task_unknown_parent_errors(kanban_home):
+    with kb.connect() as conn, pytest.raises(ValueError, match="unknown parent"):
+        kb.create_task(conn, title="orphan", parents=["t_ghost"])
+
+
+def test_workspace_kind_validation(kanban_home):
+    with kb.connect() as conn, pytest.raises(ValueError, match="workspace_kind"):
+        kb.create_task(conn, title="bad ws", workspace_kind="cloud")
+
+
+# ---------------------------------------------------------------------------
+# Links + dependency resolution
+# ---------------------------------------------------------------------------
+
+def test_link_demotes_ready_child_to_todo_when_parent_not_done(kanban_home):
+    with kb.connect() as conn:
+        a = kb.create_task(conn, title="a")
+        b = kb.create_task(conn, title="b")
+        assert kb.get_task(conn, b).status == "ready"
+        kb.link_tasks(conn, a, b)
+        assert kb.get_task(conn, b).status == "todo"
+
+
+def test_link_keeps_ready_child_when_parent_already_done(kanban_home):
+    with kb.connect() as conn:
+        a = kb.create_task(conn, title="a")
+        kb.complete_task(conn, a)
+        b = kb.create_task(conn, title="b")
+        assert kb.get_task(conn, b).status == "ready"
+        kb.link_tasks(conn, a, b)
+        assert kb.get_task(conn, b).status == "ready"
+
+
+def test_link_rejects_self_loop(kanban_home):
+    with kb.connect() as conn:
+        a = kb.create_task(conn, title="a")
+        with pytest.raises(ValueError, match="itself"):
+            kb.link_tasks(conn, a, a)
+
+
+def test_link_detects_cycle(kanban_home):
+    with kb.connect() as conn:
+        a = kb.create_task(conn, title="a")
+        b = kb.create_task(conn, title="b", parents=[a])
+        c = kb.create_task(conn, title="c", parents=[b])
+        with pytest.raises(ValueError, match="cycle"):
+            kb.link_tasks(conn, c, a)
+        with pytest.raises(ValueError, match="cycle"):
+            kb.link_tasks(conn, b, a)
+
+
+def test_recompute_ready_cascades_through_chain(kanban_home):
+    with kb.connect() as conn:
+        a = kb.create_task(conn, title="a")
+        b = kb.create_task(conn, title="b", parents=[a])
+        c = kb.create_task(conn, title="c", parents=[b])
+        assert [kb.get_task(conn, x).status for x in (a, b, c)] == \
+               ["ready", "todo", "todo"]
+        kb.complete_task(conn, a)
+        assert kb.get_task(conn, b).status == "ready"
+        kb.complete_task(conn, b)
+        assert kb.get_task(conn, c).status == "ready"
+
+
+def test_recompute_ready_fan_in_waits_for_all_parents(kanban_home):
+    with kb.connect() as conn:
+        a = kb.create_task(conn, title="a")
+        b = kb.create_task(conn, title="b")
+        c = kb.create_task(conn, title="c", parents=[a, b])
+        kb.complete_task(conn, a)
+        assert kb.get_task(conn, c).status == "todo"
+        kb.complete_task(conn, b)
+        assert kb.get_task(conn, c).status == "ready"
+
+
+# ---------------------------------------------------------------------------
+# Atomic claim (CAS)
+# ---------------------------------------------------------------------------
+
+def test_claim_once_wins_second_loses(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="a")
+        first = kb.claim_task(conn, t, claimer="host:1")
+        assert first is not None and first.status == "running"
+        second = kb.claim_task(conn, t, claimer="host:2")
+        assert second is None
+
+
+def test_claim_fails_on_non_ready(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x")
+        # Move to todo by introducing an unsatisfied parent.
+        p = kb.create_task(conn, title="p")
+        kb.link_tasks(conn, p, t)
+        assert kb.get_task(conn, t).status == "todo"
+        assert kb.claim_task(conn, t) is None
+
+
+def test_stale_claim_reclaimed(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="a")
+        kb.claim_task(conn, t)
+        # Rewind claim_expires so it looks stale.
+        conn.execute(
+            "UPDATE tasks SET claim_expires = ? WHERE id = ?",
+            (int(time.time()) - 3600, t),
+        )
+        reclaimed = kb.release_stale_claims(conn)
+        assert reclaimed == 1
+        assert kb.get_task(conn, t).status == "ready"
+
+
+def test_heartbeat_extends_claim(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="a")
+        claimer = "host:hb"
+        kb.claim_task(conn, t, claimer=claimer, ttl_seconds=60)
+        original = kb.get_task(conn, t).claim_expires
+        # Rewind then heartbeat.
+        conn.execute("UPDATE tasks SET claim_expires = ? WHERE id = ?", (0, t))
+        ok = kb.heartbeat_claim(conn, t, claimer=claimer, ttl_seconds=3600)
+        assert ok
+        new = kb.get_task(conn, t).claim_expires
+        assert new > int(time.time()) + 3000
+
+
+def test_concurrent_claims_only_one_wins(kanban_home):
+    """Fire N threads claiming the same task; exactly one must win."""
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="race", assignee="a")
+
+    def attempt(i):
+        with kb.connect() as c:
+            return kb.claim_task(c, t, claimer=f"host:{i}")
+
+    n_workers = 8
+    with concurrent.futures.ThreadPoolExecutor(max_workers=n_workers) as ex:
+        results = list(ex.map(attempt, range(n_workers)))
+    winners = [r for r in results if r is not None]
+    assert len(winners) == 1
+    assert winners[0].status == "running"
+
+
+# ---------------------------------------------------------------------------
+# Complete / block / unblock / archive / assign
+# ---------------------------------------------------------------------------
+
+def test_complete_records_result(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x")
+        assert kb.complete_task(conn, t, result="done and dusted")
+        task = kb.get_task(conn, t)
+    assert task.status == "done"
+    assert task.result == "done and dusted"
+    assert task.completed_at is not None
+
+
+def test_block_then_unblock(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="a")
+        kb.claim_task(conn, t)
+        assert kb.block_task(conn, t, reason="need input")
+        assert kb.get_task(conn, t).status == "blocked"
+        assert kb.unblock_task(conn, t)
+        assert kb.get_task(conn, t).status == "ready"
+
+
+def test_assign_refuses_while_running(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="a")
+        kb.claim_task(conn, t)
+        with pytest.raises(RuntimeError, match="currently running"):
+            kb.assign_task(conn, t, "b")
+
+
+def test_assign_reassigns_when_not_running(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="a")
+        assert kb.assign_task(conn, t, "b")
+        assert kb.get_task(conn, t).assignee == "b"
+
+
+def test_archive_hides_from_default_list(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x")
+        kb.complete_task(conn, t)
+        assert kb.archive_task(conn, t)
+        assert len(kb.list_tasks(conn)) == 0
+        assert len(kb.list_tasks(conn, include_archived=True)) == 1
+
+
+# ---------------------------------------------------------------------------
+# Comments / events / worker context
+# ---------------------------------------------------------------------------
+
+def test_comments_recorded_in_order(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x")
+        kb.add_comment(conn, t, "user", "first")
+        kb.add_comment(conn, t, "researcher", "second")
+        comments = kb.list_comments(conn, t)
+    assert [c.body for c in comments] == ["first", "second"]
+    assert [c.author for c in comments] == ["user", "researcher"]
+
+
+def test_empty_comment_rejected(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x")
+        with pytest.raises(ValueError, match="body is required"):
+            kb.add_comment(conn, t, "user", "")
+
+
+def test_events_capture_lifecycle(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="a")
+        kb.claim_task(conn, t)
+        kb.complete_task(conn, t, result="ok")
+        events = kb.list_events(conn, t)
+    kinds = [e.kind for e in events]
+    assert "created" in kinds
+    assert "claimed" in kinds
+    assert "completed" in kinds
+
+
+def test_worker_context_includes_parent_results_and_comments(kanban_home):
+    with kb.connect() as conn:
+        p = kb.create_task(conn, title="p")
+        kb.complete_task(conn, p, result="PARENT_RESULT_MARKER")
+        c = kb.create_task(conn, title="child", parents=[p])
+        kb.add_comment(conn, c, "user", "CLARIFICATION_MARKER")
+        ctx = kb.build_worker_context(conn, c)
+    assert "PARENT_RESULT_MARKER" in ctx
+    assert "CLARIFICATION_MARKER" in ctx
+    assert c in ctx
+    assert "child" in ctx
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher
+# ---------------------------------------------------------------------------
+
+def test_dispatch_dry_run_does_not_claim(kanban_home):
+    with kb.connect() as conn:
+        t1 = kb.create_task(conn, title="a", assignee="alice")
+        t2 = kb.create_task(conn, title="b", assignee="bob")
+        res = kb.dispatch_once(conn, dry_run=True)
+    assert {s[0] for s in res.spawned} == {t1, t2}
+    with kb.connect() as conn:
+        # Dry run must NOT mutate status.
+        assert kb.get_task(conn, t1).status == "ready"
+        assert kb.get_task(conn, t2).status == "ready"
+
+
+def test_dispatch_skips_unassigned(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="floater")
+        res = kb.dispatch_once(conn, dry_run=True)
+    assert t in res.skipped_unassigned
+    assert not res.spawned
+
+
+def test_dispatch_promotes_ready_and_spawns(kanban_home):
+    spawns = []
+
+    def fake_spawn(task, workspace):
+        spawns.append((task.id, task.assignee, workspace))
+
+    with kb.connect() as conn:
+        p = kb.create_task(conn, title="p", assignee="alice")
+        c = kb.create_task(conn, title="c", assignee="bob", parents=[p])
+        # Finish parent outside dispatch; promotion happens inside.
+        kb.complete_task(conn, p)
+        res = kb.dispatch_once(conn, spawn_fn=fake_spawn)
+    # Spawned c (a was already done when dispatch was called).
+    assert len(spawns) == 1
+    assert spawns[0][0] == c
+    assert spawns[0][1] == "bob"
+    # c is now running
+    with kb.connect() as conn:
+        assert kb.get_task(conn, c).status == "running"
+
+
+def test_dispatch_spawn_failure_releases_claim(kanban_home):
+    def boom(task, workspace):
+        raise RuntimeError("spawn failed")
+
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="boom", assignee="alice")
+        kb.dispatch_once(conn, spawn_fn=boom)
+        # Must return to ready so the next tick can retry.
+        assert kb.get_task(conn, t).status == "ready"
+        assert kb.get_task(conn, t).claim_lock is None
+
+
+def test_dispatch_reclaims_stale_before_spawning(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x", assignee="alice")
+        kb.claim_task(conn, t)
+        conn.execute(
+            "UPDATE tasks SET claim_expires = ? WHERE id = ?",
+            (int(time.time()) - 1, t),
+        )
+        res = kb.dispatch_once(conn, dry_run=True)
+    assert res.reclaimed == 1
+
+
+# ---------------------------------------------------------------------------
+# Workspace resolution
+# ---------------------------------------------------------------------------
+
+def test_scratch_workspace_created_under_hermes_home(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="x")
+        task = kb.get_task(conn, t)
+        ws = kb.resolve_workspace(task)
+    assert ws.exists()
+    assert ws.is_dir()
+    assert "kanban" in str(ws)
+
+
+def test_dir_workspace_honors_given_path(kanban_home, tmp_path):
+    target = tmp_path / "my-vault"
+    with kb.connect() as conn:
+        t = kb.create_task(
+            conn, title="biz", workspace_kind="dir", workspace_path=str(target)
+        )
+        task = kb.get_task(conn, t)
+        ws = kb.resolve_workspace(task)
+    assert ws == target
+    assert ws.exists()
+
+
+def test_worktree_workspace_returns_intended_path(kanban_home, tmp_path):
+    target = str(tmp_path / ".worktrees" / "my-task")
+    with kb.connect() as conn:
+        t = kb.create_task(
+            conn, title="ship", workspace_kind="worktree", workspace_path=target
+        )
+        task = kb.get_task(conn, t)
+        ws = kb.resolve_workspace(task)
+    # We do NOT auto-create worktrees; the worker's skill handles that.
+    assert str(ws) == target
+
+
+# ---------------------------------------------------------------------------
+# Tenancy
+# ---------------------------------------------------------------------------
+
+def test_tenant_column_filters_listings(kanban_home):
+    with kb.connect() as conn:
+        kb.create_task(conn, title="a1", tenant="biz-a")
+        kb.create_task(conn, title="b1", tenant="biz-b")
+        kb.create_task(conn, title="shared")  # no tenant
+        biz_a = kb.list_tasks(conn, tenant="biz-a")
+        biz_b = kb.list_tasks(conn, tenant="biz-b")
+    assert [t.title for t in biz_a] == ["a1"]
+    assert [t.title for t in biz_b] == ["b1"]
+
+
+def test_tenant_propagates_to_events(kanban_home):
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="tenant-task", tenant="biz-a")
+        events = kb.list_events(conn, t)
+    # The "created" event should have tenant in its payload.
+    created = [e for e in events if e.kind == "created"]
+    assert created and created[0].payload.get("tenant") == "biz-a"

tests/plugins/test_kanban_dashboard_plugin.py

@@ -0,0 +1,889 @@
+"""Tests for the Kanban dashboard plugin backend (plugins/kanban/dashboard/plugin_api.py).
+
+The plugin mounts as /api/plugins/kanban/ inside the dashboard's FastAPI app,
+but here we attach its router to a bare FastAPI instance so we can test the
+REST surface without spinning up the whole dashboard.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+import time
+from pathlib import Path
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from hermes_cli import kanban_db as kb
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+def _load_plugin_router():
+    """Dynamically load plugins/kanban/dashboard/plugin_api.py and return its router."""
+    repo_root = Path(__file__).resolve().parents[2]
+    plugin_file = repo_root / "plugins" / "kanban" / "dashboard" / "plugin_api.py"
+    assert plugin_file.exists(), f"plugin file missing: {plugin_file}"
+
+    spec = importlib.util.spec_from_file_location(
+        "hermes_dashboard_plugin_kanban_test", plugin_file,
+    )
+    assert spec is not None and spec.loader is not None
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = mod
+    spec.loader.exec_module(mod)
+    return mod.router
+
+
+@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
+
+
+@pytest.fixture
+def client(kanban_home):
+    app = FastAPI()
+    app.include_router(_load_plugin_router(), prefix="/api/plugins/kanban")
+    return TestClient(app)
+
+
+# ---------------------------------------------------------------------------
+# GET /board on an empty DB
+# ---------------------------------------------------------------------------
+
+
+def test_board_empty(client):
+    r = client.get("/api/plugins/kanban/board")
+    assert r.status_code == 200
+    data = r.json()
+    # All canonical columns present (triage + the rest), each empty.
+    names = [c["name"] for c in data["columns"]]
+    for expected in ("triage", "todo", "ready", "running", "blocked", "done"):
+        assert expected in names, f"missing column {expected}: {names}"
+    assert all(len(c["tasks"]) == 0 for c in data["columns"])
+    assert data["tenants"] == []
+    assert data["assignees"] == []
+    assert data["latest_event_id"] == 0
+
+
+# ---------------------------------------------------------------------------
+# POST /tasks then GET /board sees it
+# ---------------------------------------------------------------------------
+
+
+def test_create_task_appears_on_board(client):
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={
+            "title": "Research LLM caching",
+            "assignee": "researcher",
+            "priority": 3,
+            "tenant": "acme",
+        },
+    )
+    assert r.status_code == 200, r.text
+    task = r.json()["task"]
+    assert task["title"] == "Research LLM caching"
+    assert task["assignee"] == "researcher"
+    assert task["status"] == "ready"  # no parents -> immediately ready
+    assert task["priority"] == 3
+    assert task["tenant"] == "acme"
+    task_id = task["id"]
+
+    # Board now lists it under 'ready'.
+    r = client.get("/api/plugins/kanban/board")
+    assert r.status_code == 200
+    data = r.json()
+    ready = next(c for c in data["columns"] if c["name"] == "ready")
+    assert len(ready["tasks"]) == 1
+    assert ready["tasks"][0]["id"] == task_id
+    assert "acme" in data["tenants"]
+    assert "researcher" in data["assignees"]
+
+
+def test_tenant_filter(client):
+    client.post("/api/plugins/kanban/tasks", json={"title": "A", "tenant": "t1"})
+    client.post("/api/plugins/kanban/tasks", json={"title": "B", "tenant": "t2"})
+
+    r = client.get("/api/plugins/kanban/board?tenant=t1")
+    counts = {c["name"]: len(c["tasks"]) for c in r.json()["columns"]}
+    total = sum(counts.values())
+    assert total == 1
+
+    r = client.get("/api/plugins/kanban/board?tenant=t2")
+    total = sum(len(c["tasks"]) for c in r.json()["columns"])
+    assert total == 1
+
+
+# ---------------------------------------------------------------------------
+# GET /tasks/:id returns body + comments + events + links
+# ---------------------------------------------------------------------------
+
+
+def test_task_detail_includes_links_and_events(client):
+    parent = client.post(
+        "/api/plugins/kanban/tasks", json={"title": "parent"},
+    ).json()["task"]
+    child = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "child", "parents": [parent["id"]]},
+    ).json()["task"]
+    assert child["status"] == "todo"  # parent not done yet
+
+    # Detail for the child shows the parent link.
+    r = client.get(f"/api/plugins/kanban/tasks/{child['id']}")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["task"]["id"] == child["id"]
+    assert parent["id"] in data["links"]["parents"]
+
+    # Detail for the parent shows the child.
+    r = client.get(f"/api/plugins/kanban/tasks/{parent['id']}")
+    assert child["id"] in r.json()["links"]["children"]
+
+    # Events exist from creation.
+    assert len(data["events"]) >= 1
+
+
+def test_task_detail_404_on_unknown(client):
+    r = client.get("/api/plugins/kanban/tasks/does-not-exist")
+    assert r.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# PATCH /tasks/:id — status transitions
+# ---------------------------------------------------------------------------
+
+
+def test_patch_status_complete(client):
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}",
+        json={"status": "done", "result": "shipped"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["status"] == "done"
+
+    # Board reflects the move.
+    done = next(
+        c for c in client.get("/api/plugins/kanban/board").json()["columns"]
+        if c["name"] == "done"
+    )
+    assert any(x["id"] == t["id"] for x in done["tasks"])
+
+
+def test_patch_block_then_unblock(client):
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}",
+        json={"status": "blocked", "block_reason": "need input"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["status"] == "blocked"
+
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}",
+        json={"status": "ready"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["status"] == "ready"
+
+
+def test_patch_drag_drop_move_todo_to_ready(client):
+    """Direct status write: the drag-drop path for statuses without a
+    dedicated verb (e.g. manually promoting todo -> ready)."""
+    parent = client.post("/api/plugins/kanban/tasks", json={"title": "p"}).json()["task"]
+    child = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "c", "parents": [parent["id"]]},
+    ).json()["task"]
+    assert child["status"] == "todo"
+
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{child['id']}",
+        json={"status": "ready"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["status"] == "ready"
+
+
+def test_patch_reassign(client):
+    t = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "x", "assignee": "a"},
+    ).json()["task"]
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}",
+        json={"assignee": "b"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["assignee"] == "b"
+
+
+def test_patch_priority_and_edit(client):
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}",
+        json={"priority": 5, "title": "renamed"},
+    )
+    assert r.status_code == 200
+    data = r.json()["task"]
+    assert data["priority"] == 5
+    assert data["title"] == "renamed"
+
+
+def test_patch_invalid_status(client):
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}",
+        json={"status": "banana"},
+    )
+    assert r.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Comments + Links
+# ---------------------------------------------------------------------------
+
+
+def test_add_comment(client):
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    r = client.post(
+        f"/api/plugins/kanban/tasks/{t['id']}/comments",
+        json={"body": "how's progress?", "author": "teknium"},
+    )
+    assert r.status_code == 200
+
+    r = client.get(f"/api/plugins/kanban/tasks/{t['id']}")
+    comments = r.json()["comments"]
+    assert len(comments) == 1
+    assert comments[0]["body"] == "how's progress?"
+    assert comments[0]["author"] == "teknium"
+
+
+def test_add_comment_empty_rejected(client):
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    r = client.post(
+        f"/api/plugins/kanban/tasks/{t['id']}/comments",
+        json={"body": "   "},
+    )
+    assert r.status_code == 400
+
+
+def test_add_link_and_delete_link(client):
+    a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
+    b = client.post("/api/plugins/kanban/tasks", json={"title": "b"}).json()["task"]
+
+    r = client.post(
+        "/api/plugins/kanban/links",
+        json={"parent_id": a["id"], "child_id": b["id"]},
+    )
+    assert r.status_code == 200
+
+    r = client.get(f"/api/plugins/kanban/tasks/{b['id']}")
+    assert a["id"] in r.json()["links"]["parents"]
+
+    r = client.delete(
+        "/api/plugins/kanban/links",
+        params={"parent_id": a["id"], "child_id": b["id"]},
+    )
+    assert r.status_code == 200
+    assert r.json()["ok"] is True
+
+
+def test_add_link_cycle_rejected(client):
+    a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
+    b = client.post("/api/plugins/kanban/tasks", json={"title": "b"}).json()["task"]
+    client.post(
+        "/api/plugins/kanban/links",
+        json={"parent_id": a["id"], "child_id": b["id"]},
+    )
+    r = client.post(
+        "/api/plugins/kanban/links",
+        json={"parent_id": b["id"], "child_id": a["id"]},
+    )
+    assert r.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Dispatch nudge
+# ---------------------------------------------------------------------------
+
+
+def test_dispatch_dry_run(client):
+    client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "work", "assignee": "researcher"},
+    )
+    r = client.post("/api/plugins/kanban/dispatch?dry_run=true&max=4")
+    assert r.status_code == 200
+    body = r.json()
+    # DispatchResult is serialized as a dataclass dict.
+    assert isinstance(body, dict)
+
+
+# ---------------------------------------------------------------------------
+# Triage column (new v1 status)
+# ---------------------------------------------------------------------------
+
+
+def test_create_triage_lands_in_triage_column(client):
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "rough idea, spec me", "triage": True},
+    )
+    assert r.status_code == 200
+    task = r.json()["task"]
+    assert task["status"] == "triage"
+
+    r = client.get("/api/plugins/kanban/board")
+    triage = next(c for c in r.json()["columns"] if c["name"] == "triage")
+    assert len(triage["tasks"]) == 1
+    assert triage["tasks"][0]["title"] == "rough idea, spec me"
+
+
+def test_triage_task_not_promoted_to_ready(client):
+    """Triage tasks must stay in triage even when they have no parents."""
+    client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "must stay put", "triage": True},
+    )
+    # Run the dispatcher — it should NOT promote the triage task.
+    client.post("/api/plugins/kanban/dispatch?dry_run=false&max=4")
+    r = client.get("/api/plugins/kanban/board")
+    triage = next(c for c in r.json()["columns"] if c["name"] == "triage")
+    ready = next(c for c in r.json()["columns"] if c["name"] == "ready")
+    assert len(triage["tasks"]) == 1
+    assert len(ready["tasks"]) == 0
+
+
+def test_patch_status_triage_works(client):
+    """A user (or specifier) can push a task back into triage, and out of it."""
+    t = client.post(
+        "/api/plugins/kanban/tasks", json={"title": "x"},
+    ).json()["task"]
+    # Normal creation is 'ready'; push to triage.
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}", json={"status": "triage"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["status"] == "triage"
+
+    # Now promote to todo.
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{t['id']}", json={"status": "todo"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["status"] == "todo"
+
+
+# ---------------------------------------------------------------------------
+# Progress rollup (done children / total children)
+# ---------------------------------------------------------------------------
+
+
+def test_board_progress_rollup(client):
+    parent = client.post(
+        "/api/plugins/kanban/tasks", json={"title": "parent"},
+    ).json()["task"]
+    child_a = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "a", "parents": [parent["id"]]},
+    ).json()["task"]
+    child_b = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "b", "parents": [parent["id"]]},
+    ).json()["task"]
+    # Children start as "todo" because the parent isn't done yet; promote
+    # them to "ready" so complete_task will accept the transition.
+    for cid in (child_a["id"], child_b["id"]):
+        r = client.patch(
+            f"/api/plugins/kanban/tasks/{cid}", json={"status": "ready"},
+        )
+        assert r.status_code == 200
+
+    # 0/2 done.
+    r = client.get("/api/plugins/kanban/board")
+    parent_row = next(
+        t for col in r.json()["columns"] for t in col["tasks"]
+        if t["id"] == parent["id"]
+    )
+    assert parent_row["progress"] == {"done": 0, "total": 2}
+
+    # Complete one child. 1/2.
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{child_a['id']}",
+        json={"status": "done"},
+    )
+    assert r.status_code == 200
+    r = client.get("/api/plugins/kanban/board")
+    parent_row = next(
+        t for col in r.json()["columns"] for t in col["tasks"]
+        if t["id"] == parent["id"]
+    )
+    assert parent_row["progress"] == {"done": 1, "total": 2}
+
+    # Childless tasks report progress=None, not {0/0}.
+    assert next(
+        t for col in r.json()["columns"] for t in col["tasks"]
+        if t["id"] == child_b["id"]
+    )["progress"] is None
+
+
+# ---------------------------------------------------------------------------
+# Auto-init on first board read
+# ---------------------------------------------------------------------------
+
+
+def test_board_auto_initializes_missing_db(tmp_path, monkeypatch):
+    """If kanban.db doesn't exist yet, GET /board must create it, not 500."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    # Deliberately DO NOT call kb.init_db().
+
+    app = FastAPI()
+    app.include_router(_load_plugin_router(), prefix="/api/plugins/kanban")
+    c = TestClient(app)
+    r = c.get("/api/plugins/kanban/board")
+    assert r.status_code == 200
+    assert (home / "kanban.db").exists(), "init_db wasn't invoked by /board"
+
+
+# ---------------------------------------------------------------------------
+# WebSocket auth (query-param token)
+# ---------------------------------------------------------------------------
+
+
+def test_ws_events_rejects_when_token_required(tmp_path, monkeypatch):
+    """When _SESSION_TOKEN is set (normal dashboard context), a missing or
+    wrong ?token= query param must be rejected with policy-violation."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    kb.init_db()
+
+    # Stub web_server so _check_ws_token has a token to compare against.
+    import types
+    stub = types.SimpleNamespace(_SESSION_TOKEN="secret-xyz")
+    monkeypatch.setitem(sys.modules, "hermes_cli.web_server", stub)
+
+    app = FastAPI()
+    app.include_router(_load_plugin_router(), prefix="/api/plugins/kanban")
+    c = TestClient(app)
+
+    # No token → policy violation close.
+    from starlette.websockets import WebSocketDisconnect
+    with pytest.raises(WebSocketDisconnect) as exc:
+        with c.websocket_connect("/api/plugins/kanban/events"):
+            pass
+    assert exc.value.code == 1008
+
+    # Wrong token → policy violation close.
+    with pytest.raises(WebSocketDisconnect) as exc:
+        with c.websocket_connect("/api/plugins/kanban/events?token=nope"):
+            pass
+    assert exc.value.code == 1008
+
+    # Correct token → accepted (connect then close cleanly from our side).
+    with c.websocket_connect(
+        "/api/plugins/kanban/events?token=secret-xyz"
+    ) as ws:
+        assert ws is not None  # handshake succeeded
+
+
+# ---------------------------------------------------------------------------
+# Bulk actions
+# ---------------------------------------------------------------------------
+
+
+def test_bulk_status_ready(client):
+    a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
+    b = client.post("/api/plugins/kanban/tasks", json={"title": "b"}).json()["task"]
+    c2 = client.post("/api/plugins/kanban/tasks", json={"title": "c"}).json()["task"]
+    # Parent-less tasks land in "ready" already; push them to blocked first.
+    for tid in (a["id"], b["id"], c2["id"]):
+        client.patch(f"/api/plugins/kanban/tasks/{tid}",
+                     json={"status": "blocked", "block_reason": "wait"})
+
+    r = client.post("/api/plugins/kanban/tasks/bulk",
+                    json={"ids": [a["id"], b["id"], c2["id"]], "status": "ready"})
+    assert r.status_code == 200
+    results = r.json()["results"]
+    assert all(r["ok"] for r in results)
+    # All three are now ready.
+    board = client.get("/api/plugins/kanban/board").json()
+    ready = next(col for col in board["columns"] if col["name"] == "ready")
+    ids = {t["id"] for t in ready["tasks"]}
+    assert {a["id"], b["id"], c2["id"]}.issubset(ids)
+
+
+def test_bulk_archive(client):
+    a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
+    b = client.post("/api/plugins/kanban/tasks", json={"title": "b"}).json()["task"]
+    r = client.post("/api/plugins/kanban/tasks/bulk",
+                    json={"ids": [a["id"], b["id"]], "archive": True})
+    assert r.status_code == 200
+    assert all(r["ok"] for r in r.json()["results"])
+    # Default board (archived hidden) — both gone.
+    board = client.get("/api/plugins/kanban/board").json()
+    ids = {t["id"] for col in board["columns"] for t in col["tasks"]}
+    assert a["id"] not in ids
+    assert b["id"] not in ids
+
+
+def test_bulk_reassign(client):
+    a = client.post("/api/plugins/kanban/tasks",
+                    json={"title": "a", "assignee": "old"}).json()["task"]
+    b = client.post("/api/plugins/kanban/tasks",
+                    json={"title": "b", "assignee": "old"}).json()["task"]
+    r = client.post("/api/plugins/kanban/tasks/bulk",
+                    json={"ids": [a["id"], b["id"]], "assignee": "new"})
+    assert r.status_code == 200
+    for tid in (a["id"], b["id"]):
+        t = client.get(f"/api/plugins/kanban/tasks/{tid}").json()["task"]
+        assert t["assignee"] == "new"
+
+
+def test_bulk_unassign_via_empty_string(client):
+    a = client.post("/api/plugins/kanban/tasks",
+                    json={"title": "a", "assignee": "x"}).json()["task"]
+    r = client.post("/api/plugins/kanban/tasks/bulk",
+                    json={"ids": [a["id"]], "assignee": ""})
+    assert r.status_code == 200
+    t = client.get(f"/api/plugins/kanban/tasks/{a['id']}").json()["task"]
+    assert t["assignee"] is None
+
+
+def test_bulk_partial_failure_doesnt_abort_siblings(client):
+    """One bad id in the middle of a batch must not prevent others from
+    applying."""
+    a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
+    c2 = client.post("/api/plugins/kanban/tasks", json={"title": "c"}).json()["task"]
+    r = client.post("/api/plugins/kanban/tasks/bulk",
+                    json={"ids": [a["id"], "bogus-id", c2["id"]], "priority": 7})
+    assert r.status_code == 200
+    results = r.json()["results"]
+    assert len(results) == 3
+    ok_ids = {r["id"] for r in results if r["ok"]}
+    assert a["id"] in ok_ids
+    assert c2["id"] in ok_ids
+    assert any(not r["ok"] and r["id"] == "bogus-id" for r in results)
+    # Good siblings actually got the priority bump.
+    for tid in (a["id"], c2["id"]):
+        t = client.get(f"/api/plugins/kanban/tasks/{tid}").json()["task"]
+        assert t["priority"] == 7
+
+
+def test_bulk_empty_ids_400(client):
+    r = client.post("/api/plugins/kanban/tasks/bulk", json={"ids": []})
+    assert r.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# /config endpoint
+# ---------------------------------------------------------------------------
+
+
+def test_config_returns_defaults_when_section_missing(client):
+    r = client.get("/api/plugins/kanban/config")
+    assert r.status_code == 200
+    data = r.json()
+    # Defaults when dashboard.kanban is missing.
+    assert data["default_tenant"] == ""
+    assert data["lane_by_profile"] is True
+    assert data["include_archived_by_default"] is False
+    assert data["render_markdown"] is True
+
+
+def test_config_reads_dashboard_kanban_section(tmp_path, monkeypatch, client):
+    home = Path(os.environ["HERMES_HOME"])
+    (home / "config.yaml").write_text(
+        "dashboard:\n"
+        "  kanban:\n"
+        "    default_tenant: acme\n"
+        "    lane_by_profile: false\n"
+        "    include_archived_by_default: true\n"
+        "    render_markdown: false\n"
+    )
+    r = client.get("/api/plugins/kanban/config")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["default_tenant"] == "acme"
+    assert data["lane_by_profile"] is False
+    assert data["include_archived_by_default"] is True
+    assert data["render_markdown"] is False
+
+
+# ---------------------------------------------------------------------------
+# Runs surfacing (vulcan-artivus RFC feedback)
+# ---------------------------------------------------------------------------
+
+def test_task_detail_includes_runs(client):
+    """GET /tasks/:id carries a runs[] array with the attempt history."""
+    r = client.post("/api/plugins/kanban/tasks",
+                    json={"title": "port x", "assignee": "worker"}).json()
+    tid = r["task"]["id"]
+
+    # Drive status running to force a run creation: PATCH to running
+    # doesn't call claim_task (the PATCH path uses _set_status_direct),
+    # so use the bulk/claim indirection via the kernel.
+    import hermes_cli.kanban_db as _kb
+    conn = _kb.connect()
+    try:
+        _kb.claim_task(conn, tid)
+        _kb.complete_task(
+            conn, tid,
+            result="done",
+            summary="tested on rate limiter",
+            metadata={"changed_files": ["limiter.py"]},
+        )
+    finally:
+        conn.close()
+
+    d = client.get(f"/api/plugins/kanban/tasks/{tid}").json()
+    assert "runs" in d
+    assert len(d["runs"]) == 1
+    run = d["runs"][0]
+    assert run["outcome"] == "completed"
+    assert run["profile"] == "worker"
+    assert run["summary"] == "tested on rate limiter"
+    assert run["metadata"] == {"changed_files": ["limiter.py"]}
+    assert run["ended_at"] is not None
+
+
+def test_task_detail_runs_empty_before_claim(client):
+    """A task that's never been claimed has an empty runs[] list, not
+    a missing key."""
+    r = client.post("/api/plugins/kanban/tasks", json={"title": "fresh"}).json()
+    d = client.get(f"/api/plugins/kanban/tasks/{r['task']['id']}").json()
+    assert d["runs"] == []
+
+
+def test_patch_status_done_with_summary_and_metadata(client):
+    """PATCH /tasks/:id with status=done + summary + metadata must
+    reach complete_task, so the dashboard has CLI parity."""
+    # Create + claim.
+    r = client.post("/api/plugins/kanban/tasks", json={"title": "x", "assignee": "worker"})
+    tid = r.json()["task"]["id"]
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        kb.claim_task(conn, tid)
+    finally:
+        conn.close()
+
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{tid}",
+        json={
+            "status": "done",
+            "summary": "shipped the thing",
+            "metadata": {"changed_files": ["a.py", "b.py"], "tests_run": 7},
+        },
+    )
+    assert r.status_code == 200, r.text
+
+    # The run must have the summary + metadata attached.
+    conn = kb.connect()
+    try:
+        run = kb.latest_run(conn, tid)
+        assert run.outcome == "completed"
+        assert run.summary == "shipped the thing"
+        assert run.metadata == {"changed_files": ["a.py", "b.py"], "tests_run": 7}
+    finally:
+        conn.close()
+
+
+def test_patch_status_done_without_summary_still_works(client):
+    """Back-compat: PATCH without the new fields still completes."""
+    r = client.post("/api/plugins/kanban/tasks", json={"title": "y", "assignee": "worker"})
+    tid = r.json()["task"]["id"]
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        kb.claim_task(conn, tid)
+    finally:
+        conn.close()
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{tid}",
+        json={"status": "done", "result": "legacy shape"},
+    )
+    assert r.status_code == 200, r.text
+    conn = kb.connect()
+    try:
+        run = kb.latest_run(conn, tid)
+        assert run.outcome == "completed"
+        assert run.summary == "legacy shape"  # falls back to result
+    finally:
+        conn.close()
+
+
+def test_patch_status_archive_closes_running_run(client):
+    """PATCH to archived while running must close the in-flight run."""
+    r = client.post("/api/plugins/kanban/tasks", json={"title": "z", "assignee": "worker"})
+    tid = r.json()["task"]["id"]
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        kb.claim_task(conn, tid)
+        open_run = kb.latest_run(conn, tid)
+        assert open_run.ended_at is None
+    finally:
+        conn.close()
+    r = client.patch(
+        f"/api/plugins/kanban/tasks/{tid}",
+        json={"status": "archived"},
+    )
+    assert r.status_code == 200, r.text
+    conn = kb.connect()
+    try:
+        task = kb.get_task(conn, tid)
+        assert task.status == "archived"
+        assert task.current_run_id is None
+        assert kb.latest_run(conn, tid).outcome == "reclaimed"
+    finally:
+        conn.close()
+
+
+def test_event_dict_includes_run_id(client):
+    """GET /tasks/:id returns events with run_id populated."""
+    r = client.post("/api/plugins/kanban/tasks", json={"title": "e", "assignee": "worker"})
+    tid = r.json()["task"]["id"]
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        kb.claim_task(conn, tid)
+        run_id = kb.latest_run(conn, tid).id
+        kb.complete_task(conn, tid, summary="wss")
+    finally:
+        conn.close()
+
+    r = client.get(f"/api/plugins/kanban/tasks/{tid}")
+    assert r.status_code == 200
+    events = r.json()["events"]
+    # Every event in the response must have a run_id key (None or int).
+    for e in events:
+        assert "run_id" in e, f"missing run_id in event: {e}"
+    # completed event must have the actual run_id.
+    comp = [e for e in events if e["kind"] == "completed"]
+    assert comp[0]["run_id"] == run_id
+
+
+
+# ---------------------------------------------------------------------------
+# Per-task force-loaded skills via REST
+# ---------------------------------------------------------------------------
+
+def test_create_task_with_skills_roundtrips(client):
+    """POST /tasks accepts `skills: [...]`, GET /tasks/:id returns it."""
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={
+            "title": "translate docs",
+            "assignee": "linguist",
+            "skills": ["translation", "github-code-review"],
+        },
+    )
+    assert r.status_code == 200, r.text
+    task = r.json()["task"]
+    assert task["skills"] == ["translation", "github-code-review"]
+
+    # Fetch via GET /tasks/:id as the drawer does.
+    got = client.get(f"/api/plugins/kanban/tasks/{task['id']}").json()
+    assert got["task"]["skills"] == ["translation", "github-code-review"]
+
+
+def test_create_task_without_skills_defaults_to_empty_list(client):
+    """_task_dict serializes Task.skills=None as [] so the drawer can
+    always .length check without guarding against null."""
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "no skills", "assignee": "x"},
+    )
+    assert r.status_code == 200, r.text
+    task = r.json()["task"]
+    # Task.skills is None in-memory; _task_dict serializes via
+    # dataclasses.asdict which keeps it None. The drawer's
+    # `t.skills && t.skills.length > 0` guard handles both null and [].
+    assert task.get("skills") in (None, [])
+
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher-presence warning in POST /tasks response
+# ---------------------------------------------------------------------------
+
+def test_create_task_includes_warning_when_no_dispatcher(client, monkeypatch):
+    """ready+assigned task + no gateway -> response has `warning` field
+    so the dashboard UI can surface a banner."""
+    # Force the dispatcher probe to report "not running".
+    monkeypatch.setattr(
+        "hermes_cli.kanban._check_dispatcher_presence",
+        lambda: (False, "No gateway is running — start `hermes gateway start`."),
+    )
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "warn-me", "assignee": "worker"},
+    )
+    assert r.status_code == 200
+    data = r.json()
+    assert data.get("warning")
+    assert "gateway" in data["warning"].lower()
+
+
+def test_create_task_no_warning_when_dispatcher_up(client, monkeypatch):
+    """Dispatcher running -> no `warning` field in the response."""
+    monkeypatch.setattr(
+        "hermes_cli.kanban._check_dispatcher_presence",
+        lambda: (True, ""),
+    )
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "silent", "assignee": "worker"},
+    )
+    assert r.status_code == 200
+    assert "warning" not in r.json() or not r.json()["warning"]
+
+
+def test_create_task_no_warning_on_triage(client, monkeypatch):
+    """Triage tasks never get the warning (they can't be dispatched
+    anyway until promoted)."""
+    monkeypatch.setattr(
+        "hermes_cli.kanban._check_dispatcher_presence",
+        lambda: (False, "oh no"),
+    )
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "triage-task", "assignee": "worker", "triage": True},
+    )
+    assert r.status_code == 200
+    assert "warning" not in r.json() or not r.json()["warning"]
+
+
+def test_create_task_probe_error_does_not_break_create(client, monkeypatch):
+    """Probe failure must never break task creation."""
+    def _raise():
+        raise RuntimeError("probe crashed")
+    monkeypatch.setattr(
+        "hermes_cli.kanban._check_dispatcher_presence", _raise,
+    )
+    r = client.post(
+        "/api/plugins/kanban/tasks",
+        json={"title": "resilient", "assignee": "worker"},
+    )
+    assert r.status_code == 200
+    assert r.json()["task"]["title"] == "resilient"

tests/stress/README.md

@@ -0,0 +1,41 @@
+# Stress / battle-test suite
+
+Long-running tests that exercise the Kanban kernel under adversarial
+conditions. **Not run by `scripts/run_tests.sh`** because they can
+take 30+ seconds each and spawn real subprocesses.
+
+Run manually:
+
+```bash
+./venv/bin/python -m pytest tests/stress/ -v -s
+# or individual files:
+./venv/bin/python tests/stress/test_concurrency.py
+./venv/bin/python tests/stress/test_subprocess_e2e.py
+./venv/bin/python tests/stress/test_property_fuzzing.py
+./venv/bin/python tests/stress/test_benchmarks.py
+```
+
+## What's covered
+
+- **test_concurrency.py** — 5 workers, 100 tasks, race-for-claim. Asserts
+  no double-claims, no orphan runs, no SQLite errors escape retry.
+- **test_concurrency_mixed.py** — 10 workers + 1 reclaimer, 500 tasks,
+  random ops (claim/complete/block/unblock/archive). Same invariants
+  under adversarial scheduling.
+- **test_concurrency_reclaim_race.py** — TTL < work duration so the
+  reclaimer intentionally yanks tasks mid-work; verifies the worker's
+  late-complete is refused cleanly (CAS guard works).
+- **test_subprocess_e2e.py** — dispatcher spawns real Python subprocess
+  workers that heartbeat + complete via the CLI; crash detection
+  against a real dead PID.
+- **test_property_fuzzing.py** — 500 random operation sequences,
+  ~40k operations total, 9 invariant checks after each step.
+- **test_atypical_scenarios.py** — 28 scenarios covering atypical
+  user inputs: unicode/emoji/RTL, 1 MB strings, SQL injection
+  attempts, cycles, self-parents, wide fan-in/out, clock skew,
+  HERMES_HOME with spaces/unicode/symlinks, 1000 runs on one
+  task, idempotency-key race across processes, terminal-state
+  resurrection attempts, dashboard REST with weird JSON.
+- **test_benchmarks.py** — latency at 100/1k/10k tasks for dispatch,
+  recompute_ready, list_tasks, build_worker_context, etc. Results saved
+  to JSON for regression diffing.

tests/stress/_fake_worker.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+"""Fake worker process that exercises the real subprocess contract.
+
+Reads HERMES_KANBAN_TASK from env, heartbeats periodically, does short
+work, completes via the CLI. Designed to be spawned by the dispatcher
+exactly the way `hermes chat -q` would be, minus the LLM cost.
+"""
+
+import json
+import os
+import subprocess
+import sys
+import time
+
+
+def main():
+    tid = os.environ["HERMES_KANBAN_TASK"]
+    workspace = os.environ.get("HERMES_KANBAN_WORKSPACE", "")
+
+    # Announce via CLI (goes through real argparse + init_db + etc)
+    subprocess.run(
+        ["hermes", "kanban", "heartbeat", tid, "--note", "started"],
+        check=True, capture_output=True,
+    )
+
+    # Simulate work with periodic heartbeats
+    for i in range(3):
+        time.sleep(0.3)
+        subprocess.run(
+            ["hermes", "kanban", "heartbeat", tid, "--note", f"progress {i+1}/3"],
+            check=True, capture_output=True,
+        )
+
+    # Complete with structured handoff
+    subprocess.run(
+        [
+            "hermes", "kanban", "complete", tid,
+            "--summary", f"real-subprocess worker finished {tid}",
+            "--metadata", json.dumps({
+                "workspace": workspace,
+                "worker_pid": os.getpid(),
+                "iterations": 3,
+            }),
+        ],
+        check=True, capture_output=True,
+    )
+
+
+if __name__ == "__main__":
+    main()

tests/stress/conftest.py

@@ -0,0 +1,37 @@
+"""pytest config for the stress/ subdirectory.
+
+These tests are slow (30s+), spawn subprocesses, and are not run by
+default. Enable via `pytest --run-stress` or by running the scripts
+directly.
+
+The scripts are primarily __main__-executable entry points; pytest
+isn't expected to collect individual test functions from them.
+"""
+import pytest
+
+
+def pytest_collection_modifyitems(config, items):
+    if config.getoption("--run-stress", default=False):
+        return
+    skip_stress = pytest.mark.skip(
+        reason="stress test (opt-in via --run-stress or run script directly)"
+    )
+    for item in items:
+        if "tests/stress" in str(item.fspath):
+            item.add_marker(skip_stress)
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--run-stress",
+        action="store_true",
+        default=False,
+        help="Run the stress/battle-test suite (slow, spawns subprocesses).",
+    )
+
+
+collect_ignore_glob = [
+    # The stress scripts have top-level code and hard-coded paths; they're
+    # meant to run as `python tests/stress/<name>.py`, not as pytest modules.
+    "*.py",
+]

tests/stress/test_benchmarks.py

@@ -0,0 +1,221 @@
+"""Scale benchmarks for the Kanban kernel.
+
+Measures:
+  - dispatch_once latency at 100, 1000, 10000 tasks
+  - recompute_ready latency at 100, 1000, 10000 todo tasks with wide parent graphs
+  - build_worker_context latency with 1, 10, 50 parent dependencies
+  - board list/stats query latency
+  - task_runs query latency at scale
+
+Results printed as a table. Saved to JSON for regression-diffing in CI
+or future reviews. Not a pass/fail test — records numbers so we know
+when a change regresses latency by 10x and can decide whether to care.
+"""
+
+import json
+import os
+import random
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+WT = str(Path(__file__).resolve().parents[2])
+
+
+def bench(label, fn, iterations=5):
+    """Time fn over `iterations` runs, return (min, median, max) in ms."""
+    times = []
+    for _ in range(iterations):
+        t0 = time.perf_counter()
+        fn()
+        times.append((time.perf_counter() - t0) * 1000)
+    times.sort()
+    mn = times[0]
+    md = times[len(times) // 2]
+    mx = times[-1]
+    return {"label": label, "iter": iterations, "min_ms": mn, "median_ms": md, "max_ms": mx}
+
+
+def seed_tasks(conn, kb, n, assignee="bench-worker", with_parents=False):
+    """Seed n tasks. Optionally give each task 5 parents."""
+    ids = []
+    for i in range(n):
+        if with_parents and i >= 5:
+            parents = random.sample(ids[:i], 5)
+        else:
+            parents = ()
+        tid = kb.create_task(
+            conn, title=f"bench {i}", assignee=assignee,
+            tenant="bench", parents=parents,
+        )
+        ids.append(tid)
+    return ids
+
+
+def main():
+    home = tempfile.mkdtemp(prefix="hermes_bench_")
+    os.environ["HERMES_HOME"] = home
+    os.environ["HOME"] = home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    kb.init_db()
+
+    results = []
+
+    # ============ dispatch_once latency ============
+    for n in [100, 1000, 10000]:
+        print(f"\n== dispatch_once @ {n} tasks ==")
+        # Fresh DB each time so we're not measuring cumulative effects
+        import shutil
+        shutil.rmtree(home, ignore_errors=True)
+        os.makedirs(home)
+        kb._INITIALIZED_PATHS.clear()
+        kb.init_db()
+        conn = kb.connect()
+        seed_tasks(conn, kb, n, assignee=None)  # no assignee → won't spawn
+        r = bench(
+            f"dispatch_once (n={n}, no spawn)",
+            lambda: kb.dispatch_once(conn, spawn_fn=lambda *_: None),
+            iterations=5,
+        )
+        print(f"  min={r['min_ms']:.1f} median={r['median_ms']:.1f} max={r['max_ms']:.1f} ms")
+        r["n"] = n
+        results.append(r)
+        conn.close()
+
+    # ============ recompute_ready at scale with parent graphs ============
+    for n in [100, 1000, 10000]:
+        print(f"\n== recompute_ready @ {n} tasks (5 parents each) ==")
+        shutil.rmtree(home, ignore_errors=True)
+        os.makedirs(home)
+        kb._INITIALIZED_PATHS.clear()
+        kb.init_db()
+        conn = kb.connect()
+        ids = seed_tasks(conn, kb, n, assignee=None, with_parents=True)
+        # Complete the first 100 so some todo tasks might get promoted
+        for tid in ids[:min(100, n // 10)]:
+            kb.complete_task(conn, tid, result="bench")
+        r = bench(
+            f"recompute_ready (n={n}, with parents)",
+            lambda: kb.recompute_ready(conn),
+            iterations=5,
+        )
+        print(f"  min={r['min_ms']:.1f} median={r['median_ms']:.1f} max={r['max_ms']:.1f} ms")
+        r["n"] = n
+        results.append(r)
+        conn.close()
+
+    # ============ build_worker_context with N parents ============
+    for parent_count in [1, 10, 50]:
+        print(f"\n== build_worker_context with {parent_count} parents ==")
+        shutil.rmtree(home, ignore_errors=True)
+        os.makedirs(home)
+        kb._INITIALIZED_PATHS.clear()
+        kb.init_db()
+        conn = kb.connect()
+        # Create parents, complete them with summaries+metadata
+        parent_ids = []
+        for i in range(parent_count):
+            pid = kb.create_task(conn, title=f"parent {i}", assignee="p")
+            kb.claim_task(conn, pid)
+            kb.complete_task(
+                conn, pid,
+                summary=f"parent {i} result that is longer than a single token "
+                        f"so we actually measure the IO",
+                metadata={"files": [f"file_{j}.py" for j in range(5)], "i": i},
+            )
+            parent_ids.append(pid)
+        child_id = kb.create_task(
+            conn, title="child", assignee="c", parents=parent_ids,
+        )
+        r = bench(
+            f"build_worker_context (parents={parent_count})",
+            lambda: kb.build_worker_context(conn, child_id),
+            iterations=10,
+        )
+        print(f"  min={r['min_ms']:.1f} median={r['median_ms']:.1f} max={r['max_ms']:.1f} ms")
+        r["parent_count"] = parent_count
+        results.append(r)
+        conn.close()
+
+    # ============ list_tasks at scale ============
+    for n in [100, 1000, 10000]:
+        print(f"\n== list_tasks @ {n} ==")
+        shutil.rmtree(home, ignore_errors=True)
+        os.makedirs(home)
+        kb._INITIALIZED_PATHS.clear()
+        kb.init_db()
+        conn = kb.connect()
+        seed_tasks(conn, kb, n)
+        r = bench(
+            f"list_tasks (n={n})",
+            lambda: kb.list_tasks(conn),
+            iterations=5,
+        )
+        print(f"  min={r['min_ms']:.1f} median={r['median_ms']:.1f} max={r['max_ms']:.1f} ms")
+        r["n"] = n
+        results.append(r)
+        conn.close()
+
+    # ============ board_stats at scale ============
+    for n in [100, 1000, 10000]:
+        print(f"\n== board_stats @ {n} ==")
+        shutil.rmtree(home, ignore_errors=True)
+        os.makedirs(home)
+        kb._INITIALIZED_PATHS.clear()
+        kb.init_db()
+        conn = kb.connect()
+        seed_tasks(conn, kb, n)
+        r = bench(
+            f"board_stats (n={n})",
+            lambda: kb.board_stats(conn),
+            iterations=5,
+        )
+        print(f"  min={r['min_ms']:.1f} median={r['median_ms']:.1f} max={r['max_ms']:.1f} ms")
+        r["n"] = n
+        results.append(r)
+        conn.close()
+
+    # ============ list_runs at scale ============
+    for n in [100, 1000]:
+        print(f"\n== list_runs for task with {n} attempts ==")
+        shutil.rmtree(home, ignore_errors=True)
+        os.makedirs(home)
+        kb._INITIALIZED_PATHS.clear()
+        kb.init_db()
+        conn = kb.connect()
+        tid = kb.create_task(conn, title="x", assignee="w")
+        # Create N attempts via claim/release
+        for i in range(n):
+            kb.claim_task(conn, tid, ttl_seconds=0)
+            kb.release_stale_claims(conn)
+        r = bench(
+            f"list_runs (runs={n})",
+            lambda: kb.list_runs(conn, tid),
+            iterations=10,
+        )
+        print(f"  min={r['min_ms']:.1f} median={r['median_ms']:.1f} max={r['max_ms']:.1f} ms")
+        r["run_count"] = n
+        results.append(r)
+        conn.close()
+
+    # ============ SUMMARY TABLE ============
+    print()
+    print("=" * 60)
+    print("SUMMARY")
+    print("=" * 60)
+    print(f"{'Benchmark':<50} {'min':>8} {'median':>8} {'max':>8}")
+    for r in results:
+        print(f"{r['label']:<50} {r['min_ms']:>7.1f}ms {r['median_ms']:>7.1f}ms {r['max_ms']:>7.1f}ms")
+
+    # Save for future diffing.
+    out_path = "/tmp/kanban_bench_results.json"
+    with open(out_path, "w") as f:
+        json.dump(results, f, indent=2)
+    print(f"\nResults saved to {out_path}")
+
+
+if __name__ == "__main__":
+    main()

tests/stress/test_concurrency.py

@@ -0,0 +1,302 @@
+"""Multi-process concurrency stress test for the Kanban kernel.
+
+5 worker processes race for claims on a shared DB with 100 tasks. Each
+worker loops: claim -> simulate work -> complete. Asserts the invariants
+that make the system worth building:
+
+  - No task claimed by two workers simultaneously
+  - No task completed twice
+  - Every claim produces exactly one run row
+  - Every completion closes exactly one run row
+  - Zero SQLite locking errors that escape the retry layer
+  - Total run count == total claim events == total completed events
+
+This test is the primary justification for WAL + CAS-based claim. If it
+passes, the architecture holds. If it fails, we have a real bug to fix
+before anyone runs this in anger.
+"""
+
+import json
+import multiprocessing as mp
+import os
+import random
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+
+NUM_WORKERS = 5
+NUM_TASKS = 100
+WORKER_TIMEOUT_S = 60
+WT = str(Path(__file__).resolve().parents[2])
+
+
+def worker_loop(worker_id: int, hermes_home: str, result_file: str) -> None:
+    """One worker's inner loop. Runs in a fresh Python process.
+
+    Tries to claim a ready task, marks it done with a per-worker summary,
+    repeats until the ready pool is empty. Records every claim + complete
+    into its own JSON result file for later aggregation.
+    """
+    os.environ["HERMES_HOME"] = hermes_home
+    os.environ["HOME"] = hermes_home
+    sys.path.insert(0, WT)
+
+    from hermes_cli import kanban_db as kb
+
+    events = []
+    empty_polls = 0
+    start = time.monotonic()
+
+    while time.monotonic() - start < WORKER_TIMEOUT_S:
+        conn = kb.connect()
+        try:
+            # Find any ready task (non-deterministic order intentional — we
+            # want workers to race on popular assignees).
+            row = conn.execute(
+                "SELECT id FROM tasks WHERE status = 'ready' "
+                "AND claim_lock IS NULL LIMIT 1"
+            ).fetchone()
+            if row is None:
+                empty_polls += 1
+                if empty_polls > 20:
+                    break  # queue empty long enough, stop
+                time.sleep(0.01)
+                continue
+            empty_polls = 0
+
+            tid = row["id"]
+            try:
+                claimed = kb.claim_task(
+                    conn, tid, claimer=f"worker-{worker_id}",
+                )
+            except sqlite3.OperationalError as e:
+                events.append({"kind": "sqlite_err_on_claim", "task": tid, "err": str(e)})
+                continue
+            if claimed is None:
+                # Someone else beat us — expected contention, not an error.
+                events.append({"kind": "lost_claim_race", "task": tid})
+                continue
+
+            run = kb.latest_run(conn, tid)
+            events.append({
+                "kind": "claimed",
+                "task": tid,
+                "worker": worker_id,
+                "run_id": run.id,
+                "t": time.monotonic() - start,
+            })
+
+            # Simulate short, variable work
+            time.sleep(random.uniform(0.001, 0.05))
+
+            try:
+                kb.complete_task(
+                    conn, tid,
+                    result=f"done by worker-{worker_id}",
+                    summary=f"worker-{worker_id} finished task {tid}",
+                    metadata={"worker_id": worker_id, "run_id": run.id},
+                )
+            except sqlite3.OperationalError as e:
+                events.append({"kind": "sqlite_err_on_complete", "task": tid, "err": str(e)})
+                continue
+            events.append({
+                "kind": "completed",
+                "task": tid,
+                "worker": worker_id,
+                "run_id": run.id,
+                "t": time.monotonic() - start,
+            })
+        finally:
+            conn.close()
+
+    with open(result_file, "w") as f:
+        json.dump(events, f)
+
+
+def main():
+    home = tempfile.mkdtemp(prefix="hermes_concurrency_")
+    print(f"HERMES_HOME = {home}")
+
+    # Seed.
+    os.environ["HERMES_HOME"] = home
+    os.environ["HOME"] = home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    kb.init_db()
+    conn = kb.connect()
+    tids = []
+    for i in range(NUM_TASKS):
+        tid = kb.create_task(
+            conn, title=f"task #{i}", assignee="shared",
+            tenant="concurrency-test",
+        )
+        tids.append(tid)
+    conn.close()
+    print(f"Seeded {NUM_TASKS} tasks.")
+
+    # Spawn workers.
+    ctx = mp.get_context("spawn")
+    result_files = [f"/tmp/concurrency_worker_{i}.json" for i in range(NUM_WORKERS)]
+    procs = []
+    start = time.monotonic()
+    for i in range(NUM_WORKERS):
+        p = ctx.Process(target=worker_loop, args=(i, home, result_files[i]))
+        p.start()
+        procs.append(p)
+
+    for p in procs:
+        p.join(timeout=WORKER_TIMEOUT_S + 30)
+        if p.is_alive():
+            p.terminate()
+            p.join()
+
+    elapsed = time.monotonic() - start
+    print(f"All workers done in {elapsed:.1f}s")
+
+    # Aggregate worker events.
+    all_events = []
+    for i, f in enumerate(result_files):
+        if not os.path.isfile(f):
+            print(f"  WORKER {i} produced no result file — died?")
+            continue
+        with open(f) as fh:
+            events = json.load(fh)
+        all_events.extend(events)
+
+    # ============ INVARIANT CHECKS ============
+    print()
+    print("=" * 60)
+    print("INVARIANT CHECKS")
+    print("=" * 60)
+
+    failures = []
+
+    # Check 1: no task claimed by two different workers
+    claims_by_task = {}
+    for e in all_events:
+        if e["kind"] == "claimed":
+            if e["task"] in claims_by_task:
+                prev = claims_by_task[e["task"]]
+                if prev["worker"] != e["worker"]:
+                    failures.append(
+                        f"DOUBLE CLAIM: task {e['task']} claimed by "
+                        f"worker {prev['worker']} AND worker {e['worker']}"
+                    )
+            claims_by_task[e["task"]] = e
+
+    # Check 2: every completion has a matching claim from the same worker
+    for e in all_events:
+        if e["kind"] == "completed":
+            prev_claim = claims_by_task.get(e["task"])
+            if prev_claim is None:
+                failures.append(f"COMPLETION WITHOUT CLAIM: task {e['task']}")
+            elif prev_claim["worker"] != e["worker"]:
+                failures.append(
+                    f"WORKER MISMATCH: task {e['task']} claimed by "
+                    f"{prev_claim['worker']} but completed by {e['worker']}"
+                )
+
+    # Check 3: DB state — every task should be in 'done', no dangling claims
+    conn = kb.connect()
+    try:
+        bad_status = conn.execute(
+            "SELECT id, status, claim_lock, current_run_id FROM tasks "
+            "WHERE status != 'done' OR claim_lock IS NOT NULL "
+            "OR current_run_id IS NOT NULL"
+        ).fetchall()
+        if bad_status:
+            for row in bad_status:
+                failures.append(
+                    f"BAD FINAL STATE: task {row['id']} status={row['status']} "
+                    f"claim_lock={row['claim_lock']} current_run_id={row['current_run_id']}"
+                )
+
+        # Check 4: exactly one run per task, all closed as completed
+        bad_runs = conn.execute(
+            "SELECT task_id, COUNT(*) as n FROM task_runs "
+            "GROUP BY task_id HAVING n != 1"
+        ).fetchall()
+        if bad_runs:
+            for row in bad_runs:
+                failures.append(
+                    f"WRONG RUN COUNT: task {row['task_id']} has {row['n']} runs (expected 1)"
+                )
+
+        open_runs = conn.execute(
+            "SELECT id, task_id FROM task_runs WHERE ended_at IS NULL"
+        ).fetchall()
+        for row in open_runs:
+            failures.append(f"OPEN RUN: run {row['id']} on task {row['task_id']}")
+
+        wrong_outcomes = conn.execute(
+            "SELECT task_id, outcome FROM task_runs "
+            "WHERE outcome IS NULL OR outcome != 'completed'"
+        ).fetchall()
+        for row in wrong_outcomes:
+            failures.append(
+                f"WRONG OUTCOME: task {row['task_id']} run outcome={row['outcome']}"
+            )
+
+        # Check 5: event counts — exactly NUM_TASKS completed events
+        completed_events = conn.execute(
+            "SELECT COUNT(*) as n FROM task_events WHERE kind='completed'"
+        ).fetchone()["n"]
+        if completed_events != NUM_TASKS:
+            failures.append(
+                f"EVENT COUNT MISMATCH: {completed_events} completed events "
+                f"expected {NUM_TASKS}"
+            )
+
+        # Check 6: count SQLite errors that escaped retry
+        sqlite_errs = sum(
+            1 for e in all_events if e["kind"].startswith("sqlite_err")
+        )
+        if sqlite_errs > 0:
+            failures.append(f"UNRETRIED SQLITE ERRORS: {sqlite_errs}")
+
+    finally:
+        conn.close()
+
+    # ============ STATS ============
+    print()
+    total_claims = sum(1 for e in all_events if e["kind"] == "claimed")
+    total_completes = sum(1 for e in all_events if e["kind"] == "completed")
+    total_lost_races = sum(1 for e in all_events if e["kind"] == "lost_claim_race")
+
+    per_worker = {}
+    for e in all_events:
+        if e["kind"] == "completed":
+            per_worker.setdefault(e["worker"], 0)
+            per_worker[e["worker"]] += 1
+
+    print(f"Total claims:      {total_claims}")
+    print(f"Total completes:   {total_completes}")
+    print(f"Lost claim races:  {total_lost_races}  (expected contention; not a bug)")
+    print(f"Elapsed:           {elapsed:.2f}s")
+    print(f"Throughput:        {NUM_TASKS/elapsed:.1f} tasks/sec")
+    print(f"Per-worker completions:")
+    for w in sorted(per_worker.keys()):
+        print(f"  worker-{w}: {per_worker[w]}")
+
+    if failures:
+        print()
+        print("=" * 60)
+        print(f"FAILURES ({len(failures)}):")
+        print("=" * 60)
+        for f in failures[:20]:
+            print(f"  {f}")
+        if len(failures) > 20:
+            print(f"  ... and {len(failures) - 20} more")
+        sys.exit(1)
+    else:
+        print()
+        print("✔ ALL INVARIANTS HELD")
+
+
+if __name__ == "__main__":
+    main()

tests/stress/test_concurrency_mixed.py

@@ -0,0 +1,350 @@
+"""Harder concurrency stress: mixed operations + larger scale.
+
+Scales to 500 tasks, 10 workers, 60s runtime. Each worker randomly:
+  - claims + completes (70%)
+  - claims + blocks with a reason (15%)
+  - unblocks a random blocked task (10%)
+  - archives a random done task (5%)
+
+Adds a background "dispatcher" process that calls release_stale_claims
+and detect_crashed_workers every 200ms, racing against the workers to
+surface TTL + crash detection races.
+
+Pass criteria: runs invariant holds, no double-completions, no orphan
+runs, no SQLite errors escape the retry layer.
+"""
+
+import json
+import multiprocessing as mp
+import os
+import random
+import sqlite3
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+NUM_WORKERS = 10
+NUM_TASKS = 500
+RUN_DURATION_S = 30
+WT = str(Path(__file__).resolve().parents[2])
+
+
+def worker_loop(worker_id: int, hermes_home: str, result_file: str) -> None:
+    os.environ["HERMES_HOME"] = hermes_home
+    os.environ["HOME"] = hermes_home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    events = []
+    start = time.monotonic()
+    idle_rounds = 0
+
+    while time.monotonic() - start < RUN_DURATION_S:
+        conn = kb.connect()
+        try:
+            op = random.random()
+
+            if op < 0.10:
+                # Try to unblock a blocked task.
+                row = conn.execute(
+                    "SELECT id FROM tasks WHERE status='blocked' "
+                    "ORDER BY RANDOM() LIMIT 1"
+                ).fetchone()
+                if row:
+                    try:
+                        ok = kb.unblock_task(conn, row["id"])
+                        events.append({"kind": "unblocked" if ok else "unblock_noop",
+                                       "task": row["id"], "worker": worker_id})
+                    except sqlite3.OperationalError as e:
+                        events.append({"kind": "sqlite_err", "op": "unblock",
+                                       "task": row["id"], "err": str(e)[:100]})
+                continue
+
+            if op < 0.15:
+                # Try to archive a done task.
+                row = conn.execute(
+                    "SELECT id FROM tasks WHERE status='done' "
+                    "ORDER BY RANDOM() LIMIT 1"
+                ).fetchone()
+                if row:
+                    try:
+                        kb.archive_task(conn, row["id"])
+                        events.append({"kind": "archived", "task": row["id"],
+                                       "worker": worker_id})
+                    except sqlite3.OperationalError as e:
+                        events.append({"kind": "sqlite_err", "op": "archive",
+                                       "task": row["id"], "err": str(e)[:100]})
+                continue
+
+            # Default: claim + complete-or-block.
+            row = conn.execute(
+                "SELECT id FROM tasks WHERE status='ready' "
+                "AND claim_lock IS NULL LIMIT 1"
+            ).fetchone()
+            if row is None:
+                idle_rounds += 1
+                if idle_rounds > 50:
+                    break
+                time.sleep(0.02)
+                continue
+            idle_rounds = 0
+
+            tid = row["id"]
+            try:
+                claimed = kb.claim_task(
+                    conn, tid, claimer=f"worker-{worker_id}",
+                    ttl_seconds=5,  # short TTL so reclaim races in
+                )
+            except sqlite3.OperationalError as e:
+                events.append({"kind": "sqlite_err", "op": "claim",
+                               "task": tid, "err": str(e)[:100]})
+                continue
+            if claimed is None:
+                events.append({"kind": "lost_claim_race", "task": tid})
+                continue
+
+            run = kb.latest_run(conn, tid)
+            events.append({"kind": "claimed", "task": tid, "worker": worker_id,
+                           "run_id": run.id, "t": time.monotonic() - start})
+
+            time.sleep(random.uniform(0.005, 0.05))
+
+            # 20% of the time, block instead of complete
+            if random.random() < 0.20:
+                try:
+                    kb.block_task(conn, tid,
+                                  reason=f"blocked by worker-{worker_id}")
+                    events.append({"kind": "blocked", "task": tid,
+                                   "worker": worker_id, "run_id": run.id})
+                except sqlite3.OperationalError as e:
+                    events.append({"kind": "sqlite_err", "op": "block",
+                                   "task": tid, "err": str(e)[:100]})
+            else:
+                try:
+                    kb.complete_task(
+                        conn, tid,
+                        result=f"done by worker-{worker_id}",
+                        summary=f"worker-{worker_id} ok",
+                        metadata={"worker_id": worker_id},
+                    )
+                    events.append({"kind": "completed", "task": tid,
+                                   "worker": worker_id, "run_id": run.id,
+                                   "t": time.monotonic() - start})
+                except sqlite3.OperationalError as e:
+                    events.append({"kind": "sqlite_err", "op": "complete",
+                                   "task": tid, "err": str(e)[:100]})
+        finally:
+            conn.close()
+
+    with open(result_file, "w") as f:
+        json.dump(events, f)
+
+
+def reclaimer_loop(hermes_home: str, result_file: str) -> None:
+    """Background dispatcher-like loop that reclaims stale tasks."""
+    os.environ["HERMES_HOME"] = hermes_home
+    os.environ["HOME"] = hermes_home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    events = []
+    start = time.monotonic()
+    while time.monotonic() - start < RUN_DURATION_S + 2:
+        conn = kb.connect()
+        try:
+            try:
+                reclaimed = kb.release_stale_claims(conn)
+                if reclaimed:
+                    events.append({"kind": "reclaimed", "count": reclaimed,
+                                   "t": time.monotonic() - start})
+            except sqlite3.OperationalError as e:
+                events.append({"kind": "sqlite_err", "op": "reclaim",
+                               "err": str(e)[:100]})
+        finally:
+            conn.close()
+        time.sleep(0.2)
+
+    with open(result_file, "w") as f:
+        json.dump(events, f)
+
+
+def main():
+    home = tempfile.mkdtemp(prefix="hermes_mixed_stress_")
+    print(f"HERMES_HOME = {home}")
+
+    os.environ["HERMES_HOME"] = home
+    os.environ["HOME"] = home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    kb.init_db()
+    conn = kb.connect()
+    for i in range(NUM_TASKS):
+        kb.create_task(
+            conn, title=f"t#{i}", assignee="shared", tenant="mixed-stress",
+        )
+    conn.close()
+    print(f"Seeded {NUM_TASKS} tasks, launching {NUM_WORKERS} workers + 1 reclaimer")
+
+    ctx = mp.get_context("spawn")
+    worker_results = [f"/tmp/mixed_worker_{i}.json" for i in range(NUM_WORKERS)]
+    reclaim_result = "/tmp/mixed_reclaim.json"
+
+    procs = []
+    start = time.monotonic()
+    for i in range(NUM_WORKERS):
+        p = ctx.Process(target=worker_loop, args=(i, home, worker_results[i]))
+        p.start()
+        procs.append(p)
+    r = ctx.Process(target=reclaimer_loop, args=(home, reclaim_result))
+    r.start()
+    procs.append(r)
+
+    for p in procs:
+        p.join(timeout=RUN_DURATION_S + 30)
+        if p.is_alive():
+            p.terminate()
+            p.join()
+
+    elapsed = time.monotonic() - start
+    print(f"Done in {elapsed:.1f}s")
+
+    # Aggregate.
+    all_events = []
+    for i, f in enumerate(worker_results):
+        if os.path.isfile(f):
+            with open(f) as fh:
+                all_events.extend(json.load(fh))
+        else:
+            print(f"  WORKER {i} died with no result file!")
+    reclaim_events = []
+    if os.path.isfile(reclaim_result):
+        with open(reclaim_result) as fh:
+            reclaim_events = json.load(fh)
+
+    # ============ INVARIANT CHECKS ============
+    print()
+    print("=" * 60)
+    print("INVARIANT CHECKS")
+    print("=" * 60)
+
+    failures = []
+
+    # Per-run attribution tracking
+    claims = [e for e in all_events if e["kind"] == "claimed"]
+    completions = [e for e in all_events if e["kind"] == "completed"]
+    blocks = [e for e in all_events if e["kind"] == "blocked"]
+
+    # Every completion must have a matching claim on the same run_id AND
+    # the same worker (workers don't steal each other's runs).
+    claims_by_run = {c["run_id"]: c for c in claims}
+    for comp in completions:
+        claim = claims_by_run.get(comp["run_id"])
+        if claim is None:
+            # It's possible this worker saw a reclaimed run from another worker
+            # — that's still a bug: the worker shouldn't be able to complete
+            # a run it didn't claim. But let me check if reclaim happened first.
+            failures.append(
+                f"COMPLETION WITHOUT CLAIM: task {comp['task']} run {comp['run_id']} "
+                f"by worker {comp['worker']}"
+            )
+        elif claim["worker"] != comp["worker"]:
+            failures.append(
+                f"CROSS-WORKER COMPLETION: run {comp['run_id']} claimed by "
+                f"worker {claim['worker']} but completed by worker {comp['worker']}"
+            )
+
+    # SQLite errors that escaped the retry layer
+    sqlite_errs = [e for e in all_events if e["kind"] == "sqlite_err"]
+    if sqlite_errs:
+        for e in sqlite_errs[:5]:
+            failures.append(f"SQLITE ERROR: op={e.get('op')} err={e.get('err')}")
+        if len(sqlite_errs) > 5:
+            failures.append(f"  ... and {len(sqlite_errs) - 5} more sqlite errs")
+
+    # DB final state — every task should be in a clean terminal state.
+    conn = kb.connect()
+    try:
+        # Invariant: current_run_id NULL iff latest run is terminal
+        inconsistent = conn.execute("""
+            SELECT t.id, t.status, t.current_run_id
+            FROM tasks t
+            WHERE t.current_run_id IS NOT NULL
+              AND EXISTS (SELECT 1 FROM task_runs r
+                          WHERE r.id = t.current_run_id AND r.ended_at IS NOT NULL)
+        """).fetchall()
+        for row in inconsistent:
+            failures.append(
+                f"INVARIANT VIOLATION: task {row['id']} status={row['status']} "
+                f"has current_run_id={row['current_run_id']} but run is ended"
+            )
+
+        # Invariant: no orphan open runs
+        orphans = conn.execute("""
+            SELECT r.id, r.task_id, r.status
+            FROM task_runs r
+            LEFT JOIN tasks t ON t.current_run_id = r.id
+            WHERE r.ended_at IS NULL AND t.id IS NULL
+        """).fetchall()
+        for row in orphans:
+            failures.append(
+                f"ORPHAN OPEN RUN: run {row['id']} on task {row['task_id']}"
+            )
+
+        # Counts — should roughly balance.
+        status_counts = dict(
+            conn.execute("SELECT status, COUNT(*) FROM tasks GROUP BY status").fetchall()
+        )
+        run_outcome_counts = dict(
+            conn.execute(
+                "SELECT outcome, COUNT(*) FROM task_runs "
+                "WHERE ended_at IS NOT NULL GROUP BY outcome"
+            ).fetchall()
+        )
+        active_runs = conn.execute(
+            "SELECT COUNT(*) FROM task_runs WHERE ended_at IS NULL"
+        ).fetchone()[0]
+
+    finally:
+        conn.close()
+
+    # ============ STATS ============
+    print()
+    print(f"Workers: {NUM_WORKERS}, Tasks: {NUM_TASKS}")
+    print(f"Elapsed: {elapsed:.1f}s")
+    print(f"Events collected: {len(all_events)} (+{len(reclaim_events)} reclaim)")
+    print()
+    print("Operations:")
+    op_counts = {}
+    for e in all_events:
+        op_counts[e["kind"]] = op_counts.get(e["kind"], 0) + 1
+    for k in sorted(op_counts.keys()):
+        print(f"  {k:<25} {op_counts[k]}")
+
+    print()
+    print("Final task status:")
+    for s, n in sorted(status_counts.items()):
+        print(f"  {s:<10} {n}")
+    print("Final run outcomes:")
+    for o, n in sorted(run_outcome_counts.items(), key=lambda x: (x[0] or '',)):
+        print(f"  {o:<12} {n}")
+    print(f"  active       {active_runs}")
+
+    if failures:
+        print()
+        print("=" * 60)
+        print(f"FAILURES ({len(failures)}):")
+        print("=" * 60)
+        for f in failures[:30]:
+            print(f"  {f}")
+        if len(failures) > 30:
+            print(f"  ... and {len(failures) - 30} more")
+        sys.exit(1)
+    else:
+        print()
+        print("✔ ALL INVARIANTS HELD UNDER MIXED STRESS")
+
+
+if __name__ == "__main__":
+    main()

tests/stress/test_concurrency_reclaim_race.py

@@ -0,0 +1,241 @@
+"""Target the reclaim race specifically.
+
+Workers claim tasks with a 1s TTL but sleep 2s before completing. The
+reclaimer runs every 200ms. Scenario: worker claims, reclaimer expires
+the claim mid-work, worker tries to complete AFTER its run has been
+reclaimed.
+
+Expected behavior (per design): the worker's complete_task should
+either succeed on the reclaimed-and-re-claimed-by-another-worker case
+(no, it should refuse — the claim was invalidated), OR succeed by
+grace (we "forgive" a late complete from the original worker if no
+one else picked it up).
+
+Actually looking at complete_task: it doesn't check claim_lock. It just
+transitions from 'running' -> 'done'. So if the reclaimer moved it back
+to 'ready', the late worker's complete_task will fail (CAS on
+status='running' fails). This is the CORRECT behavior.
+
+Invariant being tested: race between worker.complete and
+dispatcher.reclaim must not produce a double-run-close or other
+inconsistency.
+"""
+
+import json
+import multiprocessing as mp
+import os
+import random
+import sqlite3
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+NUM_WORKERS = 5
+NUM_TASKS = 50
+TTL = 1
+WORK_DURATION_S = 2.0  # longer than TTL => reclaimer wins
+WT = str(Path(__file__).resolve().parents[2])
+
+
+def worker_loop(worker_id: int, hermes_home: str, result_file: str) -> None:
+    os.environ["HERMES_HOME"] = hermes_home
+    os.environ["HOME"] = hermes_home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    events = []
+    start = time.monotonic()
+    idle = 0
+
+    while time.monotonic() - start < 40:
+        conn = kb.connect()
+        try:
+            row = conn.execute(
+                "SELECT id FROM tasks WHERE status='ready' AND claim_lock IS NULL LIMIT 1"
+            ).fetchone()
+            if row is None:
+                idle += 1
+                if idle > 30:
+                    break
+                time.sleep(0.05)
+                continue
+            idle = 0
+            tid = row["id"]
+            try:
+                claimed = kb.claim_task(conn, tid, claimer=f"worker-{worker_id}",
+                                        ttl_seconds=TTL)
+            except sqlite3.OperationalError as e:
+                events.append({"kind": "sqlite_err", "op": "claim", "err": str(e)[:100]})
+                continue
+            if claimed is None:
+                events.append({"kind": "lost_claim", "task": tid})
+                continue
+            run = kb.latest_run(conn, tid)
+            events.append({"kind": "claimed", "task": tid, "worker": worker_id,
+                           "run_id": run.id})
+
+            # Sleep longer than TTL so reclaimer has a chance to intervene
+            time.sleep(WORK_DURATION_S + random.uniform(-0.3, 0.3))
+
+            try:
+                ok = kb.complete_task(
+                    conn, tid,
+                    result=f"by worker-{worker_id}",
+                    summary=f"worker-{worker_id} finished",
+                )
+                events.append({"kind": "complete_ok" if ok else "complete_refused",
+                               "task": tid, "worker": worker_id, "run_id": run.id})
+            except sqlite3.OperationalError as e:
+                events.append({"kind": "sqlite_err", "op": "complete", "err": str(e)[:100]})
+        finally:
+            conn.close()
+
+    with open(result_file, "w") as f:
+        json.dump(events, f)
+
+
+def reclaimer_loop(hermes_home: str, result_file: str) -> None:
+    os.environ["HERMES_HOME"] = hermes_home
+    os.environ["HOME"] = hermes_home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    events = []
+    start = time.monotonic()
+    while time.monotonic() - start < 42:
+        conn = kb.connect()
+        try:
+            try:
+                n = kb.release_stale_claims(conn)
+                if n:
+                    events.append({"kind": "reclaimed", "count": n,
+                                   "t": time.monotonic() - start})
+            except sqlite3.OperationalError as e:
+                events.append({"kind": "sqlite_err", "err": str(e)[:100]})
+        finally:
+            conn.close()
+        time.sleep(0.2)
+    with open(result_file, "w") as f:
+        json.dump(events, f)
+
+
+def main():
+    home = tempfile.mkdtemp(prefix="hermes_reclaim_race_")
+    os.environ["HERMES_HOME"] = home
+    os.environ["HOME"] = home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    kb.init_db()
+    conn = kb.connect()
+    for i in range(NUM_TASKS):
+        kb.create_task(conn, title=f"t{i}", assignee="shared",
+                       tenant="reclaim-race")
+    conn.close()
+    print(f"Seeded {NUM_TASKS} tasks. TTL={TTL}s, work_duration={WORK_DURATION_S}s")
+    print(f"(worker work > TTL guarantees reclaims)")
+
+    ctx = mp.get_context("spawn")
+    worker_results = [f"/tmp/rc_worker_{i}.json" for i in range(NUM_WORKERS)]
+    reclaim_result = "/tmp/rc_reclaim.json"
+    procs = []
+    for i in range(NUM_WORKERS):
+        p = ctx.Process(target=worker_loop, args=(i, home, worker_results[i]))
+        p.start()
+        procs.append(p)
+    r = ctx.Process(target=reclaimer_loop, args=(home, reclaim_result))
+    r.start()
+    procs.append(r)
+
+    for p in procs:
+        p.join(timeout=60)
+        if p.is_alive():
+            p.terminate()
+            p.join()
+
+    # Aggregate.
+    all_events = []
+    for f in worker_results:
+        if os.path.isfile(f):
+            with open(f) as fh:
+                all_events.extend(json.load(fh))
+    reclaim_events = []
+    if os.path.isfile(reclaim_result):
+        with open(reclaim_result) as fh:
+            reclaim_events = json.load(fh)
+
+    op_counts = {}
+    for e in all_events:
+        op_counts[e["kind"]] = op_counts.get(e["kind"], 0) + 1
+    total_reclaims = sum(e.get("count", 0) for e in reclaim_events)
+    print(f"\nReclaimer fired {len(reclaim_events)} times, total tasks reclaimed: {total_reclaims}")
+    print("Worker events:")
+    for k in sorted(op_counts):
+        print(f"  {k:<25} {op_counts[k]}")
+
+    # Invariant checks
+    failures = []
+    conn = kb.connect()
+    try:
+        # Any task stuck with current_run_id pointing at a closed run?
+        bad = conn.execute("""
+            SELECT t.id, t.status, t.current_run_id, r.ended_at, r.outcome
+            FROM tasks t
+            JOIN task_runs r ON r.id = t.current_run_id
+            WHERE r.ended_at IS NOT NULL
+        """).fetchall()
+        for row in bad:
+            failures.append(
+                f"INVARIANT VIOLATION: task {row['id']} status={row['status']} "
+                f"current_run_id={row['current_run_id']} but run ended "
+                f"outcome={row['outcome']}"
+            )
+        # Every run with NULL ended_at should still have the task pointing at it
+        orphans = conn.execute("""
+            SELECT r.id, r.task_id
+            FROM task_runs r
+            LEFT JOIN tasks t ON t.current_run_id = r.id
+            WHERE r.ended_at IS NULL AND t.id IS NULL
+        """).fetchall()
+        for row in orphans:
+            failures.append(f"ORPHAN OPEN RUN: run {row['id']} on task {row['task_id']}")
+        # Event counts
+        claim_evts = conn.execute(
+            "SELECT COUNT(*) FROM task_events WHERE kind='claimed'").fetchone()[0]
+        reclaim_evts = conn.execute(
+            "SELECT COUNT(*) FROM task_events WHERE kind='reclaimed'").fetchone()[0]
+        comp_evts = conn.execute(
+            "SELECT COUNT(*) FROM task_events WHERE kind='completed'").fetchone()[0]
+        print(f"\nDB event counts: claimed={claim_evts} reclaimed={reclaim_evts} completed={comp_evts}")
+        # Every reclaimed run must have ended_at set
+        unended_reclaims = conn.execute(
+            "SELECT COUNT(*) FROM task_runs WHERE outcome='reclaimed' AND ended_at IS NULL"
+        ).fetchone()[0]
+        if unended_reclaims:
+            failures.append(f"UNENDED RECLAIMED RUNS: {unended_reclaims}")
+        # Count of completed runs
+        comp_runs = conn.execute(
+            "SELECT COUNT(*) FROM task_runs WHERE outcome='completed'"
+        ).fetchone()[0]
+        reclaim_runs = conn.execute(
+            "SELECT COUNT(*) FROM task_runs WHERE outcome='reclaimed'"
+        ).fetchone()[0]
+        print(f"DB run outcomes: completed={comp_runs} reclaimed={reclaim_runs}")
+    finally:
+        conn.close()
+
+    if reclaim_runs == 0:
+        failures.append("NO RECLAIMS HAPPENED — test didn't stress what it was supposed to")
+
+    if failures:
+        print(f"\nFAILURES ({len(failures)}):")
+        for f in failures[:20]:
+            print(f"  {f}")
+        sys.exit(1)
+    else:
+        print("\n✔ RECLAIM RACE INVARIANTS HELD")
+
+
+if __name__ == "__main__":
+    main()

tests/stress/test_property_fuzzing.py

@@ -0,0 +1,283 @@
+"""Randomized property testing for the Kanban kernel.
+
+Generates 1000 random operation sequences, each 20-50 ops, on small
+task graphs. After each step, checks the full invariant set:
+
+  I1. If tasks.current_run_id IS NOT NULL, the run MUST exist AND
+      ended_at MUST be NULL (we never point at a closed run).
+  I2. If a run has ended_at NULL, SOME task MUST have current_run_id
+      pointing at it (no orphan open runs).
+  I3. task.status in the valid set {triage, todo, ready, running,
+      blocked, done, archived}.
+  I4. task.claim_lock NULL iff status not in (running,).
+  I5. Every run has started_at <= ended_at (or ended_at is NULL).
+  I6. If outcome is set, ended_at must also be set.
+  I7. Events are strictly monotonic in (created_at, id).
+  I8. task_events.run_id references a task_runs.id that exists
+      (or is NULL).
+  I9. Parent completion invariant: if all parents are 'done', the
+      child cannot be in 'todo' status (recompute_ready should have
+      promoted it). This is called out in the comment on
+      recompute_ready; verify it holds after every random seq.
+
+Not using hypothesis the lib; just Python random for simplicity.
+"""
+
+import os
+import random
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+WT = str(Path(__file__).resolve().parents[2])
+NUM_SEQUENCES = 500
+OPS_PER_SEQUENCE = 100
+TASK_POOL = 10
+
+OPS = [
+    "create", "create_child", "claim", "complete", "block", "unblock",
+    "archive", "heartbeat", "release_stale", "detect_crashed",
+    "recompute_ready", "reassign",
+]
+
+
+def assert_invariants(conn, kb, ops_log):
+    """Run all invariant checks; raise AssertionError with context on any."""
+    failures = []
+
+    # I1: current_run_id → run exists and not ended
+    bad_ptr = conn.execute("""
+        SELECT t.id, t.current_run_id, r.ended_at, r.outcome
+        FROM tasks t
+        LEFT JOIN task_runs r ON r.id = t.current_run_id
+        WHERE t.current_run_id IS NOT NULL
+          AND (r.id IS NULL OR r.ended_at IS NOT NULL)
+    """).fetchall()
+    for row in bad_ptr:
+        if row["ended_at"] is None and row["outcome"] is None:
+            detail = "missing"
+        else:
+            detail = f"closed ({row['outcome']})"
+        failures.append(
+            f"I1: task {row['id']} points at run {row['current_run_id']} "
+            f"which is {detail}"
+        )
+
+    # I2: open run → some task points at it
+    orphans = conn.execute("""
+        SELECT r.id, r.task_id
+        FROM task_runs r
+        WHERE r.ended_at IS NULL
+          AND NOT EXISTS (SELECT 1 FROM tasks t WHERE t.current_run_id = r.id)
+    """).fetchall()
+    for row in orphans:
+        failures.append(f"I2: open run {row['id']} on task {row['task_id']} has no pointer")
+
+    # I3: valid statuses
+    valid = {"triage", "todo", "ready", "running", "blocked", "done", "archived"}
+    bad_status = conn.execute("SELECT id, status FROM tasks").fetchall()
+    for row in bad_status:
+        if row["status"] not in valid:
+            failures.append(f"I3: task {row['id']} has invalid status {row['status']!r}")
+
+    # I4: claim_lock set only when running
+    bad_lock = conn.execute("""
+        SELECT id, status, claim_lock FROM tasks
+        WHERE (status != 'running' AND claim_lock IS NOT NULL)
+    """).fetchall()
+    for row in bad_lock:
+        failures.append(
+            f"I4: task {row['id']} status={row['status']} but claim_lock={row['claim_lock']!r}"
+        )
+
+    # I5: run started_at <= ended_at
+    bad_times = conn.execute("""
+        SELECT id, started_at, ended_at FROM task_runs
+        WHERE ended_at IS NOT NULL AND started_at > ended_at
+    """).fetchall()
+    for row in bad_times:
+        failures.append(
+            f"I5: run {row['id']} started_at={row['started_at']} > ended_at={row['ended_at']}"
+        )
+
+    # I6: outcome set → ended_at set
+    bad_outcome = conn.execute("""
+        SELECT id, outcome, ended_at FROM task_runs
+        WHERE outcome IS NOT NULL AND ended_at IS NULL
+    """).fetchall()
+    for row in bad_outcome:
+        failures.append(f"I6: run {row['id']} outcome={row['outcome']} but ended_at NULL")
+
+    # I7: events monotonic in id (always true for autoincrement)
+    # Skip — autoincrement guarantees it.
+
+    # I8: event.run_id references existing run
+    bad_ev_fk = conn.execute("""
+        SELECT e.id, e.run_id FROM task_events e
+        LEFT JOIN task_runs r ON r.id = e.run_id
+        WHERE e.run_id IS NOT NULL AND r.id IS NULL
+    """).fetchall()
+    for row in bad_ev_fk:
+        failures.append(f"I8: event {row['id']} references missing run {row['run_id']}")
+
+    # I9: if all parents done → child not in todo
+    # (Only applies to children with at least one parent)
+    orphaned_todo = conn.execute("""
+        SELECT c.id AS child_id,
+               COUNT(*) AS n_parents,
+               SUM(CASE WHEN p.status = 'done' THEN 1 ELSE 0 END) AS done_parents
+        FROM tasks c
+        JOIN task_links l ON l.child_id = c.id
+        JOIN tasks p ON p.id = l.parent_id
+        WHERE c.status = 'todo'
+        GROUP BY c.id
+        HAVING n_parents > 0 AND n_parents = done_parents
+    """).fetchall()
+    for row in orphaned_todo:
+        failures.append(
+            f"I9: task {row['child_id']} is todo but all {row['n_parents']} parents are done"
+        )
+
+    if failures:
+        print(f"\n!!! INVARIANT VIOLATION after {len(ops_log)} ops:")
+        for f in failures[:10]:
+            print(f"  {f}")
+        if len(failures) > 10:
+            print(f"  ... and {len(failures) - 10} more")
+        print("\nLast 10 ops:")
+        for op in ops_log[-10:]:
+            print(f"  {op}")
+        return False
+    return True
+
+
+def random_op(rng, conn, kb, task_pool):
+    op = rng.choice(OPS)
+
+    if op == "create":
+        tid = kb.create_task(
+            conn,
+            title=f"rand {rng.randint(0, 1000)}",
+            assignee=rng.choice(["w1", "w2", "w3", None]),
+        )
+        task_pool.append(tid)
+        return {"op": "create", "tid": tid}
+
+    if op == "create_child" and task_pool:
+        parent = rng.choice(task_pool)
+        tid = kb.create_task(
+            conn, title=f"child of {parent}",
+            assignee=rng.choice(["w1", "w2", "w3", None]),
+            parents=[parent],
+        )
+        task_pool.append(tid)
+        return {"op": "create_child", "tid": tid, "parent": parent}
+
+    if not task_pool:
+        return None
+
+    tid = rng.choice(task_pool)
+    task = kb.get_task(conn, tid)
+    if task is None:
+        task_pool.remove(tid)
+        return None
+
+    if op == "claim":
+        claimed = kb.claim_task(conn, tid, ttl_seconds=rng.choice([1, 3, 10]))
+        return {"op": "claim", "tid": tid, "ok": claimed is not None}
+    if op == "complete":
+        summary = rng.choice([None, f"done via op {rng.randint(0, 1000)}"])
+        ok = kb.complete_task(conn, tid, summary=summary)
+        return {"op": "complete", "tid": tid, "ok": ok}
+    if op == "block":
+        reason = rng.choice([None, "rand block"])
+        ok = kb.block_task(conn, tid, reason=reason)
+        return {"op": "block", "tid": tid, "ok": ok}
+    if op == "unblock":
+        ok = kb.unblock_task(conn, tid)
+        return {"op": "unblock", "tid": tid, "ok": ok}
+    if op == "archive":
+        ok = kb.archive_task(conn, tid)
+        if ok:
+            task_pool.remove(tid)
+        return {"op": "archive", "tid": tid, "ok": ok}
+    if op == "heartbeat":
+        ok = kb.heartbeat_worker(conn, tid)
+        return {"op": "heartbeat", "tid": tid, "ok": ok}
+    if op == "release_stale":
+        n = kb.release_stale_claims(conn)
+        return {"op": "release_stale", "n": n}
+    if op == "detect_crashed":
+        # Force-kill a fake PID first so there's something to detect
+        crashed = kb.detect_crashed_workers(conn)
+        return {"op": "detect_crashed", "n": len(crashed)}
+    if op == "recompute_ready":
+        n = kb.recompute_ready(conn)
+        return {"op": "recompute_ready", "promoted": n}
+    if op == "reassign":
+        # Reassignment isn't a direct API; simulate via assign_task
+        new_a = rng.choice(["w1", "w2", "w3", None])
+        try:
+            kb.assign_task(conn, tid, new_a)
+            return {"op": "reassign", "tid": tid, "to": new_a}
+        except Exception as e:
+            return {"op": "reassign", "tid": tid, "err": str(e)[:50]}
+
+    return None
+
+
+def main():
+    total_ops = 0
+    total_violations = 0
+
+    for seq_idx in range(NUM_SEQUENCES):
+        seed = random.randint(0, 10**9)
+        rng = random.Random(seed)
+        home = tempfile.mkdtemp(prefix=f"hermes_fuzz_{seq_idx}_")
+        os.environ["HERMES_HOME"] = home
+        os.environ["HOME"] = home
+        sys.path.insert(0, WT)
+
+        # Fresh module state per sequence to avoid cached init paths.
+        for m in list(sys.modules.keys()):
+            if m.startswith("hermes_cli"):
+                del sys.modules[m]
+        from hermes_cli import kanban_db as kb
+
+        kb.init_db()
+        conn = kb.connect()
+        task_pool = []
+        ops_log = []
+
+        try:
+            for i in range(OPS_PER_SEQUENCE):
+                result = random_op(rng, conn, kb, task_pool)
+                if result is None:
+                    continue
+                ops_log.append(result)
+                total_ops += 1
+                if not assert_invariants(conn, kb, ops_log):
+                    total_violations += 1
+                    print(f"  sequence {seq_idx} (seed={seed}) failed at op {i}")
+                    break
+        finally:
+            conn.close()
+
+        if seq_idx % 10 == 0:
+            print(f"  seq {seq_idx:3d}: {total_ops} ops so far, {total_violations} violations")
+
+    print()
+    print("=" * 60)
+    print(f"Total sequences: {NUM_SEQUENCES}")
+    print(f"Total operations: {total_ops}")
+    print(f"Invariant violations: {total_violations}")
+    if total_violations == 0:
+        print("\n✔ ALL INVARIANTS HELD ACROSS RANDOMIZED SEQUENCES")
+    else:
+        print("\n✗ INVARIANT VIOLATIONS FOUND")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

tests/stress/test_subprocess_e2e.py

@@ -0,0 +1,228 @@
+"""E2E: dispatcher spawns real Python subprocess workers.
+
+This validates the IPC + lifecycle story that mocks can't:
+  - spawn_fn returns a real PID
+  - the child process resolves hermes_cli.kanban_db on its own
+  - the child writes heartbeats via the CLI (real argparse, real init_db)
+  - the child completes via the CLI with --summary + --metadata
+  - the dispatcher observes all of this through the DB only
+  - worker logs are captured to HERMES_HOME/kanban/logs/<task>.log
+  - crash detection works against a real dead PID
+"""
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+
+WT = str(Path(__file__).resolve().parents[2])
+FAKE_WORKER = str(Path(__file__).parent / "_fake_worker.py")
+PY = sys.executable
+
+
+def make_spawn_fn(home: str):
+    """Return a spawn_fn the dispatcher can call. Launches the fake
+    worker as a detached subprocess."""
+
+    def _spawn(task, workspace):
+        log_path = os.path.join(home, f"worker_{task.id}.log")
+        env = {
+            **os.environ,
+            "HERMES_HOME": home,
+            "HOME": home,
+            "PYTHONPATH": WT,
+            "HERMES_KANBAN_TASK": task.id,
+            "HERMES_KANBAN_WORKSPACE": workspace,
+            "PATH": f"{os.path.dirname(PY)}:{os.environ.get('PATH','')}",
+        }
+        log_f = open(log_path, "ab")
+        proc = subprocess.Popen(
+            [PY, FAKE_WORKER],
+            stdin=subprocess.DEVNULL,
+            stdout=log_f,
+            stderr=subprocess.STDOUT,
+            env=env,
+            start_new_session=True,
+        )
+        return proc.pid
+
+    return _spawn
+
+
+def main():
+    home = tempfile.mkdtemp(prefix="hermes_e2e_")
+    os.environ["HERMES_HOME"] = home
+    os.environ["HOME"] = home
+    sys.path.insert(0, WT)
+    from hermes_cli import kanban_db as kb
+
+    # Point the `hermes` CLI child processes will run at the worktree
+    # hermes_cli.main. We do this by putting a shim on PATH.
+    shim_dir = os.path.join(home, "bin")
+    os.makedirs(shim_dir, exist_ok=True)
+    shim_path = os.path.join(shim_dir, "hermes")
+    with open(shim_path, "w") as f:
+        f.write(f"""#!/bin/sh
+exec {PY} -m hermes_cli.main "$@"
+""")
+    os.chmod(shim_path, 0o755)
+    os.environ["PATH"] = f"{shim_dir}:{os.environ.get('PATH','')}"
+
+    kb.init_db()
+    conn = kb.connect()
+
+    # ============ SCENARIO A: happy path, 3 tasks ============
+    print("=" * 60)
+    print("A. Real-subprocess happy path (3 tasks)")
+    print("=" * 60)
+
+    tids = []
+    for i in range(3):
+        tid = kb.create_task(
+            conn, title=f"real-e2e-{i}", assignee="worker",
+        )
+        tids.append(tid)
+
+    spawn_fn = make_spawn_fn(home)
+    result = kb.dispatch_once(conn, spawn_fn=spawn_fn)
+    print(f"  dispatched: {len(result.spawned)} spawned")
+    spawned_pids = []
+    # The dispatcher sets worker_pid on each claimed task via _set_worker_pid.
+    for tid in tids:
+        task = kb.get_task(conn, tid)
+        spawned_pids.append(task.worker_pid)
+        print(f"  task {tid}: pid={task.worker_pid} status={task.status}")
+
+    # Wait for all workers to complete (up to 10s).
+    deadline = time.monotonic() + 10
+    while time.monotonic() < deadline:
+        statuses = [kb.get_task(conn, tid).status for tid in tids]
+        if all(s == "done" for s in statuses):
+            break
+        time.sleep(0.2)
+
+    print()
+    failures = []
+    for tid in tids:
+        task = kb.get_task(conn, tid)
+        runs = kb.list_runs(conn, tid)
+        print(f"  task {tid}: status={task.status}, current_run_id={task.current_run_id}, "
+              f"runs={[(r.id, r.outcome) for r in runs]}")
+        if task.status != "done":
+            failures.append(f"task {tid} not done: status={task.status}")
+        if task.current_run_id is not None:
+            failures.append(f"task {tid} has dangling current_run_id={task.current_run_id}")
+        if len(runs) != 1:
+            failures.append(f"task {tid} has {len(runs)} runs, expected 1")
+        else:
+            r = runs[0]
+            if r.outcome != "completed":
+                failures.append(f"task {tid} run outcome={r.outcome}, expected completed")
+            if not r.summary or "real-subprocess worker finished" not in r.summary:
+                failures.append(f"task {tid} summary missing: {r.summary!r}")
+            if not r.metadata or r.metadata.get("iterations") != 3:
+                failures.append(f"task {tid} metadata missing iterations: {r.metadata}")
+            # Heartbeat events should be present
+            events = kb.list_events(conn, tid)
+            heartbeats = [e for e in events if e.kind == "heartbeat"]
+            if len(heartbeats) < 3:  # start + 3 progress
+                failures.append(f"task {tid} heartbeats={len(heartbeats)} expected >=3")
+
+    if failures:
+        print("\nFAILURES:")
+        for f in failures:
+            print(f"  {f}")
+        sys.exit(1)
+
+    print("\n  ✔ Scenario A: all 3 real-subprocess workers completed cleanly")
+
+    # ============ SCENARIO B: crashed worker ============
+    print()
+    print("=" * 60)
+    print("B. Crashed worker (kill -9 mid-heartbeat)")
+    print("=" * 60)
+
+    crash_tid = kb.create_task(
+        conn, title="crash-e2e", assignee="worker",
+    )
+
+    # Spawn a worker that sleeps long enough for us to kill it.
+    # CRITICAL: spawn through a double-fork so when we kill the child it
+    # doesn't zombify under our pid (which would fool kill -0 liveness
+    # checks into thinking it's still alive). In production the
+    # dispatcher daemon is long-lived but its workers are reaped by init
+    # after exit; the test needs to match that orphaning behavior.
+    def spawn_sleeper(task, workspace):
+        r, w = os.pipe()
+        middleman = subprocess.Popen(
+            [
+                PY, "-c",
+                "import os,sys,subprocess;"
+                "p=subprocess.Popen(['sleep','30'],"
+                "stdin=subprocess.DEVNULL,"
+                "stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,"
+                "start_new_session=True);"
+                "os.write(int(sys.argv[1]), str(p.pid).encode());"
+                "sys.exit(0)",
+                str(w),
+            ],
+            pass_fds=(w,),
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        os.close(w)
+        middleman.wait()  # middleman exits immediately, orphaning the sleep
+        grandchild_pid = int(os.read(r, 16))
+        os.close(r)
+        return grandchild_pid
+
+    result = kb.dispatch_once(conn, spawn_fn=spawn_sleeper)
+    task = kb.get_task(conn, crash_tid)
+    print(f"  spawned sleeper pid={task.worker_pid} for {crash_tid}")
+    # Kill the sleeper forcibly
+    os.kill(task.worker_pid, 9)
+    # Give the OS a moment to reap
+    time.sleep(0.5)
+
+    # Simulate next dispatcher tick — should detect the crashed PID
+    crashed = kb.detect_crashed_workers(conn)
+    print(f"  detect_crashed_workers returned {len(crashed)} crashed (expected 1)")
+
+    task = kb.get_task(conn, crash_tid)
+    runs = kb.list_runs(conn, crash_tid)
+    print(f"  task status={task.status}, runs={[(r.id, r.outcome) for r in runs]}")
+
+    if len(crashed) < 1:
+        print("  ✗ crash NOT detected")
+        sys.exit(1)
+    if task.status != "ready":
+        print(f"  ✗ task should be back to ready, got {task.status}")
+        sys.exit(1)
+    if runs[0].outcome != "crashed":
+        print(f"  ✗ run outcome should be 'crashed', got {runs[0].outcome!r}")
+        sys.exit(1)
+    print("\n  ✔ Scenario B: crash detected, task re-queued, run outcome=crashed")
+
+    # ============ SCENARIO C: worker log was captured ============
+    print()
+    print("=" * 60)
+    print("C. Worker log captured to disk")
+    print("=" * 60)
+    # Scenario A workers wrote to /tmp/hermes_e2e_*/worker_*.log
+    import glob
+    logs = glob.glob(os.path.join(home, "worker_*.log"))
+    print(f"  {len(logs)} worker log files")
+    for lp in logs[:3]:
+        size = os.path.getsize(lp)
+        print(f"    {os.path.basename(lp)}: {size} bytes")
+        # Our fake worker is quiet (no prints); size=0 is fine
+
+    conn.close()
+    print("\n✔ ALL E2E SCENARIOS PASS")
+
+
+if __name__ == "__main__":
+    main()

tests/tools/test_kanban_tools.py

@@ -0,0 +1,494 @@
+"""Tests for the Kanban tool surface (tools/kanban_tools.py).
+
+Verifies:
+  - Tools are gated on HERMES_KANBAN_TASK: a normal chat session sees
+    zero kanban tools in its schema; a worker session sees all seven.
+  - Each handler's happy path.
+  - Error paths (missing required args, bad metadata type, etc).
+"""
+from __future__ import annotations
+
+import json
+import os
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Gating
+# ---------------------------------------------------------------------------
+
+def test_kanban_tools_hidden_without_env_var(monkeypatch, tmp_path):
+    """Normal `hermes chat` sessions (no HERMES_KANBAN_TASK) must have
+    zero kanban_* tools in their schema."""
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    import tools.kanban_tools  # ensure registered
+    from tools.registry import registry
+    from toolsets import resolve_toolset
+
+    schema = registry.get_definitions(set(resolve_toolset("hermes-cli")), quiet=True)
+    names = {s["function"].get("name") for s in schema if "function" in s}
+    kanban = {n for n in names if n and n.startswith("kanban_")}
+    assert kanban == set(), (
+        f"kanban tools leaked into normal chat schema: {kanban}"
+    )
+
+
+def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path):
+    """Worker sessions (HERMES_KANBAN_TASK set) must have all 7 tools."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    import tools.kanban_tools  # ensure registered
+    from tools.registry import registry
+    from toolsets import resolve_toolset
+
+    schema = registry.get_definitions(set(resolve_toolset("hermes-cli")), quiet=True)
+    names = {s["function"].get("name") for s in schema if "function" in s}
+    kanban = {n for n in names if n and n.startswith("kanban_")}
+    expected = {
+        "kanban_show", "kanban_complete", "kanban_block", "kanban_heartbeat",
+        "kanban_comment", "kanban_create", "kanban_link",
+    }
+    assert kanban == expected, f"expected {expected}, got {kanban}"
+
+
+# ---------------------------------------------------------------------------
+# Handler happy paths
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def worker_env(monkeypatch, tmp_path):
+    """Simulate being a worker: HERMES_HOME isolated, HERMES_KANBAN_TASK set
+    after we've created the task."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setenv("HERMES_PROFILE", "test-worker")
+    from pathlib import Path as _Path
+    monkeypatch.setattr(_Path, "home", lambda: tmp_path)
+
+    from hermes_cli import kanban_db as kb
+    kb._INITIALIZED_PATHS.clear()
+    kb.init_db()
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="worker-test", assignee="test-worker")
+        kb.claim_task(conn, tid)
+    finally:
+        conn.close()
+    monkeypatch.setenv("HERMES_KANBAN_TASK", tid)
+    return tid
+
+
+def test_show_defaults_to_env_task_id(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_show({})
+    d = json.loads(out)
+    assert "task" in d
+    assert d["task"]["id"] == worker_env
+    assert d["task"]["status"] == "running"
+    assert "worker_context" in d
+    assert "runs" in d
+
+
+def test_show_explicit_task_id(worker_env):
+    """Peek at a different task than the one in env."""
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        other = kb.create_task(conn, title="other task", assignee="peer")
+    finally:
+        conn.close()
+    from tools import kanban_tools as kt
+    out = kt._handle_show({"task_id": other})
+    d = json.loads(out)
+    assert d["task"]["id"] == other
+
+
+def test_complete_happy_path(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_complete({
+        "summary": "got the thing done",
+        "metadata": {"files": 2},
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    assert d["task_id"] == worker_env
+    # Verify via kernel
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        run = kb.latest_run(conn, worker_env)
+        assert run.outcome == "completed"
+        assert run.summary == "got the thing done"
+        assert run.metadata == {"files": 2}
+    finally:
+        conn.close()
+
+
+def test_complete_with_result_only(worker_env):
+    """`result` alone (without summary) is accepted for legacy compat."""
+    from tools import kanban_tools as kt
+    out = kt._handle_complete({"result": "legacy result"})
+    d = json.loads(out)
+    assert d["ok"] is True
+
+
+def test_complete_rejects_no_handoff(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_complete({})
+    assert json.loads(out).get("error"), "should have errored"
+
+
+def test_complete_rejects_non_dict_metadata(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_complete({"summary": "x", "metadata": [1, 2, 3]})
+    assert json.loads(out).get("error")
+
+
+def test_block_happy_path(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_block({"reason": "need clarification"})
+    d = json.loads(out)
+    assert d["ok"] is True
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        assert kb.get_task(conn, worker_env).status == "blocked"
+    finally:
+        conn.close()
+
+
+def test_block_rejects_empty_reason(worker_env):
+    from tools import kanban_tools as kt
+    for bad in ["", "   ", None]:
+        out = kt._handle_block({"reason": bad})
+        assert json.loads(out).get("error")
+
+
+def test_heartbeat_happy_path(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_heartbeat({"note": "progress"})
+    d = json.loads(out)
+    assert d["ok"] is True
+
+
+def test_heartbeat_without_note(worker_env):
+    """note is optional."""
+    from tools import kanban_tools as kt
+    out = kt._handle_heartbeat({})
+    d = json.loads(out)
+    assert d["ok"] is True
+
+
+def test_comment_happy_path(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_comment({
+        "task_id": worker_env,
+        "body": "hello thread",
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    assert d["comment_id"]
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        comments = kb.list_comments(conn, worker_env)
+        assert len(comments) == 1
+        # Author defaults to HERMES_PROFILE env we set in the fixture
+        assert comments[0].author == "test-worker"
+        assert comments[0].body == "hello thread"
+    finally:
+        conn.close()
+
+
+def test_comment_rejects_empty_body(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_comment({"task_id": worker_env, "body": "   "})
+    assert json.loads(out).get("error")
+
+
+def test_comment_custom_author(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_comment({
+        "task_id": worker_env, "body": "hi", "author": "custom-bot",
+    })
+    assert json.loads(out)["ok"]
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        comments = kb.list_comments(conn, worker_env)
+        assert comments[0].author == "custom-bot"
+    finally:
+        conn.close()
+
+
+def test_create_happy_path(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_create({
+        "title": "child task",
+        "assignee": "peer",
+        "parents": [worker_env],
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    assert d["task_id"]
+    assert d["status"] == "todo"  # parent isn't done yet
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        child = kb.get_task(conn, d["task_id"])
+        assert child.title == "child task"
+        assert child.assignee == "peer"
+    finally:
+        conn.close()
+
+
+def test_create_rejects_no_title(worker_env):
+    from tools import kanban_tools as kt
+    assert json.loads(kt._handle_create({"assignee": "x"})).get("error")
+    assert json.loads(kt._handle_create({"title": "   ", "assignee": "x"})).get("error")
+
+
+def test_create_rejects_no_assignee(worker_env):
+    from tools import kanban_tools as kt
+    assert json.loads(kt._handle_create({"title": "t"})).get("error")
+
+
+def test_create_rejects_non_list_parents(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_create({"title": "t", "assignee": "a", "parents": 42})
+    assert json.loads(out).get("error")
+
+
+def test_create_accepts_string_parent(worker_env):
+    """Convenience: a single parent id as string is coerced to [id]."""
+    from tools import kanban_tools as kt
+    out = kt._handle_create({
+        "title": "t", "assignee": "a", "parents": worker_env,
+    })
+    assert json.loads(out)["ok"]
+
+
+def test_create_accepts_skills_list(worker_env):
+    """Tool writes the per-task skills through to the kernel."""
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+    out = kt._handle_create({
+        "title": "skilled",
+        "assignee": "linguist",
+        "skills": ["translation", "github-code-review"],
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    with kb.connect() as conn:
+        task = kb.get_task(conn, d["task_id"])
+    assert task.skills == ["translation", "github-code-review"]
+
+
+def test_create_accepts_skills_string(worker_env):
+    """Convenience: a single skill name as string is coerced to [name]."""
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+    out = kt._handle_create({
+        "title": "one-skill",
+        "assignee": "a",
+        "skills": "translation",
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    with kb.connect() as conn:
+        task = kb.get_task(conn, d["task_id"])
+    assert task.skills == ["translation"]
+
+
+def test_create_rejects_non_list_skills(worker_env):
+    """skills: 42 must be rejected, not silently dropped."""
+    from tools import kanban_tools as kt
+    out = kt._handle_create({
+        "title": "t", "assignee": "a", "skills": 42,
+    })
+    assert json.loads(out).get("error")
+
+
+def test_link_happy_path(worker_env):
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        a = kb.create_task(conn, title="A", assignee="x")
+        b = kb.create_task(conn, title="B", assignee="x")
+    finally:
+        conn.close()
+    from tools import kanban_tools as kt
+    out = kt._handle_link({"parent_id": a, "child_id": b})
+    d = json.loads(out)
+    assert d["ok"] is True
+
+
+def test_link_rejects_self_reference(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_link({"parent_id": worker_env, "child_id": worker_env})
+    assert json.loads(out).get("error")
+
+
+def test_link_rejects_missing_args(worker_env):
+    from tools import kanban_tools as kt
+    assert json.loads(kt._handle_link({"parent_id": "x"})).get("error")
+    assert json.loads(kt._handle_link({"child_id": "y"})).get("error")
+
+
+def test_link_rejects_cycle(worker_env):
+    """A → B, then try to link B → A."""
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        a = kb.create_task(conn, title="A", assignee="x")
+        b = kb.create_task(conn, title="B", assignee="x", parents=[a])
+    finally:
+        conn.close()
+    from tools import kanban_tools as kt
+    out = kt._handle_link({"parent_id": b, "child_id": a})
+    assert json.loads(out).get("error")
+
+
+# ---------------------------------------------------------------------------
+# End-to-end: simulate a full worker lifecycle through the tools
+# ---------------------------------------------------------------------------
+
+def test_worker_lifecycle_through_tools(worker_env):
+    """Drive the full claim -> heartbeat -> comment -> complete lifecycle
+    exclusively through the tools, then verify the DB state matches what
+    the dispatcher/notifier expect."""
+    from tools import kanban_tools as kt
+
+    # 1. show — worker orientation
+    show = json.loads(kt._handle_show({}))
+    assert show["task"]["id"] == worker_env
+
+    # 2. heartbeat during long op
+    assert json.loads(kt._handle_heartbeat({"note": "warming up"}))["ok"]
+
+    # 3. comment for a future peer
+    assert json.loads(kt._handle_comment({
+        "task_id": worker_env,
+        "body": "note: using stdlib sqlite3 bindings",
+    }))["ok"]
+
+    # 4. spawn a child task for follow-up
+    child_out = json.loads(kt._handle_create({
+        "title": "write integration test",
+        "assignee": "qa",
+        "parents": [worker_env],
+    }))
+    assert child_out["ok"]
+
+    # 5. complete with structured handoff
+    comp = json.loads(kt._handle_complete({
+        "summary": "implemented + spawned QA follow-up",
+        "metadata": {"child_task": child_out["task_id"]},
+    }))
+    assert comp["ok"]
+
+    # Verify final state
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        parent = kb.get_task(conn, worker_env)
+        assert parent.status == "done"
+        assert parent.current_run_id is None
+        run = kb.latest_run(conn, worker_env)
+        assert run.outcome == "completed"
+        assert run.metadata == {"child_task": child_out["task_id"]}
+        # Child is todo (parent just finished, but recompute_ready may
+        # have promoted it — complete_task runs recompute internally).
+        child = kb.get_task(conn, child_out["task_id"])
+        assert child.status == "ready", (
+            f"child should be ready after parent done, got {child.status}"
+        )
+        # Comment is visible
+        assert len(kb.list_comments(conn, worker_env)) == 1
+        # Heartbeat event recorded
+        hb = [e for e in kb.list_events(conn, worker_env) if e.kind == "heartbeat"]
+        assert len(hb) == 1
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# System-prompt guidance injection
+# ---------------------------------------------------------------------------
+
+def test_kanban_guidance_not_in_normal_prompt(monkeypatch, tmp_path):
+    """A normal chat session (no HERMES_KANBAN_TASK) must NOT have
+    KANBAN_GUIDANCE in its system prompt."""
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    from pathlib import Path as _P
+    monkeypatch.setattr(_P, "home", lambda: tmp_path)
+
+    from run_agent import AIAgent
+    a = AIAgent(
+        api_key="test",
+        base_url="https://openrouter.ai/api/v1",
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+    )
+    prompt = a._build_system_prompt()
+    assert "You are a Kanban worker" not in prompt
+    assert "kanban_show()" not in prompt
+
+
+def test_kanban_guidance_in_worker_prompt(monkeypatch, tmp_path):
+    """A worker session (HERMES_KANBAN_TASK set) MUST have the full
+    lifecycle guidance in its system prompt."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    from pathlib import Path as _P
+    monkeypatch.setattr(_P, "home", lambda: tmp_path)
+
+    from run_agent import AIAgent
+    a = AIAgent(
+        api_key="test",
+        base_url="https://openrouter.ai/api/v1",
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+    )
+    prompt = a._build_system_prompt()
+    # Header phrase
+    assert "You are a Kanban worker" in prompt
+    # Lifecycle signals
+    assert "kanban_show()" in prompt
+    assert "kanban_complete" in prompt
+    assert "kanban_block" in prompt
+    assert "kanban_create" in prompt
+    # Anti-shell guidance
+    assert "Do not shell out" in prompt or "tools — they work" in prompt
+
+
+def test_kanban_guidance_prompt_size_bounded(monkeypatch, tmp_path):
+    """Sanity: the guidance block is under 4 KB so it doesn't blow
+    up the cached prompt."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    from pathlib import Path as _P
+    monkeypatch.setattr(_P, "home", lambda: tmp_path)
+
+    from agent.prompt_builder import KANBAN_GUIDANCE
+    assert 1_500 < len(KANBAN_GUIDANCE) < 4_096, (
+        f"KANBAN_GUIDANCE is {len(KANBAN_GUIDANCE)} chars — too short (missing?) or too long"
+    )