fix(kanban): restrict board routing tools

tests/tools/test_kanban_tools.py

@@ -1,8 +1,8 @@
 """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 the kanban set.
+  - A normal chat session sees zero kanban tools; dispatcher-spawned
+    workers see lifecycle tools; orchestrator profiles see board tools.
   - Each handler's happy path.
   - Error paths (missing required args, bad metadata type, etc).
 """
@@ -40,7 +40,7 @@ def test_kanban_tools_hidden_without_env_var(monkeypatch, tmp_path):
 
 
 def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path):
-    """Worker sessions (HERMES_KANBAN_TASK set) must have kanban tools."""
+    """Worker sessions get task lifecycle tools, not board-routing tools."""
     monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
     home = tmp_path / ".hermes"
     home.mkdir()
@@ -55,17 +55,15 @@ def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path):
     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_list",
         "kanban_show", "kanban_complete", "kanban_block", "kanban_heartbeat",
         "kanban_comment", "kanban_create", "kanban_link",
-        "kanban_assign", "kanban_unblock", "kanban_archive",
     }
     assert kanban == expected, f"expected {expected}, got {kanban}"
 
 
-def test_kanban_tools_visible_with_toolset_config(monkeypatch, tmp_path):
-    """Orchestrator profiles with toolsets: [kanban] see the same tools."""
-    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+def test_worker_with_kanban_toolset_still_hides_board_routing(monkeypatch, tmp_path):
+    """Task scope wins over profile config for board-routing tools."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
     home = tmp_path / ".hermes"
     home.mkdir()
     (home / "config.yaml").write_text("toolsets:\n  - kanban\n")
@@ -84,7 +82,32 @@ def test_kanban_tools_visible_with_toolset_config(monkeypatch, tmp_path):
         "kanban_assign",
         "kanban_unblock",
         "kanban_archive",
-    }.issubset(kanban)
+    }.isdisjoint(kanban)
+
+
+def test_kanban_tools_visible_with_toolset_config(monkeypatch, tmp_path):
+    """Orchestrator profiles with toolsets: [kanban] see all board tools."""
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text("toolsets:\n  - kanban\n")
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    import tools.kanban_tools  # ensure registered
+    from tools.registry import invalidate_check_fn_cache, registry
+    from toolsets import resolve_toolset
+
+    invalidate_check_fn_cache()
+    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_list",
+        "kanban_show", "kanban_complete", "kanban_block", "kanban_heartbeat",
+        "kanban_comment", "kanban_create", "kanban_link",
+        "kanban_assign", "kanban_unblock", "kanban_archive",
+    }
+    assert kanban == expected, f"expected {expected}, got {kanban}"
 
 
 # ---------------------------------------------------------------------------
@@ -140,8 +163,9 @@ def test_show_explicit_task_id(worker_env):
     assert d["task"]["id"] == other
 
 
-def test_list_filters_tasks(worker_env):
+def test_list_filters_tasks(monkeypatch, worker_env):
     """kanban_list gives orchestrators filtered board discovery."""
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from hermes_cli import kanban_db as kb
     conn = kb.connect()
     try:
@@ -170,19 +194,81 @@ def test_list_filters_tasks(worker_env):
     assert tenant_ids == [c]
 
 
-def test_list_rejects_invalid_status(worker_env):
+def test_list_rejects_invalid_status(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from tools import kanban_tools as kt
     out = kt._handle_list({"status": "not-a-state"})
     assert "status must be one of" in json.loads(out).get("error", "")
 
 
-def test_list_rejects_bad_limit(worker_env):
+def test_list_rejects_bad_limit(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from tools import kanban_tools as kt
     assert json.loads(kt._handle_list({"limit": "nope"})).get("error")
     assert json.loads(kt._handle_list({"limit": 0})).get("error")
+    assert json.loads(kt._handle_list({"limit": 201})).get("error")
 
 
-def test_list_parses_include_archived_string_false(worker_env):
+def test_list_defaults_limit_and_reports_truncation(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        for i in range(55):
+            kb.create_task(conn, title=f"task {i:02d}", assignee="factory")
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    d = json.loads(kt._handle_list({"assignee": "factory"}))
+    assert d["count"] == 50
+    assert d["limit"] == 50
+    assert d["truncated"] is True
+    assert d["next_limit"] == 100
+
+
+def test_list_treats_null_limit_as_default(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        kb.create_task(conn, title="task", assignee="factory")
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    d = json.loads(kt._handle_list({"assignee": "factory", "limit": None}))
+    assert d["count"] == 1
+    assert d["limit"] == 50
+    assert d["truncated"] is False
+
+
+def test_list_honors_explicit_limit_without_truncation(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        for i in range(3):
+            kb.create_task(conn, title=f"task {i}", assignee="factory")
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    d = json.loads(kt._handle_list({"assignee": "factory", "limit": 10}))
+    assert d["count"] == 3
+    assert d["limit"] == 10
+    assert d["truncated"] is False
+    assert d["next_limit"] is None
+
+
+def test_worker_list_is_orchestrator_only(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_list({})
+    assert "orchestrator-only" in json.loads(out).get("error", "")
+
+
+def test_list_parses_include_archived_string_false(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from hermes_cli import kanban_db as kb
     conn = kb.connect()
     try:
@@ -202,7 +288,8 @@ def test_list_parses_include_archived_string_false(worker_env):
     assert archived not in ids
 
 
-def test_list_parses_include_archived_string_true(worker_env):
+def test_list_parses_include_archived_string_true(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from hermes_cli import kanban_db as kb
     conn = kb.connect()
     try:
@@ -222,7 +309,8 @@ def test_list_parses_include_archived_string_true(worker_env):
     assert archived in ids
 
 
-def test_list_rejects_bad_include_archived(worker_env):
+def test_list_rejects_bad_include_archived(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from tools import kanban_tools as kt
     out = kt._handle_list({"include_archived": "sometimes"})
     assert "include_archived must be" in json.loads(out).get("error", "")
@@ -588,13 +676,15 @@ def test_assign_can_unassign(monkeypatch, worker_env):
     assert d["assignee"] is None
 
 
-def test_assign_rejects_missing_args(worker_env):
+def test_assign_rejects_missing_args(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from tools import kanban_tools as kt
     assert json.loads(kt._handle_assign({"assignee": "x"})).get("error")
     assert json.loads(kt._handle_assign({"task_id": worker_env})).get("error")
 
 
-def test_assign_rejects_running_claimed_task(worker_env):
+def test_assign_rejects_running_claimed_task(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from tools import kanban_tools as kt
     out = kt._handle_assign({"task_id": worker_env, "assignee": "reviewer"})
     d = json.loads(out)
@@ -624,7 +714,8 @@ def test_unblock_happy_path(monkeypatch, worker_env):
         conn.close()
 
 
-def test_unblock_rejects_non_blocked_task(worker_env):
+def test_unblock_rejects_non_blocked_task(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
     from tools import kanban_tools as kt
     out = kt._handle_unblock({"task_id": worker_env})
     assert json.loads(out).get("error")
@@ -811,12 +902,11 @@ def test_kanban_guidance_prompt_size_bounded(monkeypatch, tmp_path):
 # ---------------------------------------------------------------------------
 #
 # A worker process has HERMES_KANBAN_TASK set to its own task id. The
-# destructive tools (kanban_complete, kanban_block, kanban_heartbeat,
-# kanban_assign, kanban_unblock, kanban_archive) must refuse to operate
-# on any OTHER task id, even if the caller supplies an explicit `task_id`
-# argument. Workers legitimately call kanban_show / kanban_list /
-# kanban_comment / kanban_create / kanban_link on other tasks, so those
-# are unrestricted.
+# lifecycle tools (kanban_complete, kanban_block, kanban_heartbeat) must
+# refuse to operate on any OTHER task id, even if the caller supplies an
+# explicit `task_id` argument. Board-routing tools (kanban_list,
+# kanban_assign, kanban_unblock, kanban_archive) are orchestrator-only and
+# hidden from dispatcher-spawned workers entirely.
 #
 # Orchestrator profiles (no HERMES_KANBAN_TASK in env) are intentionally
 # exempt — their job is routing, and they sometimes close out child
@@ -889,8 +979,8 @@ def test_worker_heartbeat_rejects_foreign_task_id(worker_env):
     assert "refusing to mutate" in d.get("error", "")
 
 
-def test_worker_assign_rejects_foreign_task_id(worker_env):
-    """A worker cannot reassign a task that isn't its own."""
+def test_worker_assign_is_orchestrator_only(worker_env):
+    """A worker cannot use the assignment router, even on its own task."""
     from hermes_cli import kanban_db as kb
     conn = kb.connect()
     try:
@@ -901,7 +991,7 @@ def test_worker_assign_rejects_foreign_task_id(worker_env):
     from tools import kanban_tools as kt
     out = kt._handle_assign({"task_id": other, "assignee": "reviewer"})
     d = json.loads(out)
-    assert "refusing to mutate" in d.get("error", "")
+    assert "orchestrator-only" in d.get("error", "")
 
     conn = kb.connect()
     try:
@@ -910,8 +1000,8 @@ def test_worker_assign_rejects_foreign_task_id(worker_env):
         conn.close()
 
 
-def test_worker_unblock_rejects_foreign_task_id(worker_env):
-    """A worker cannot unblock a task that isn't its own."""
+def test_worker_unblock_is_orchestrator_only(worker_env):
+    """A worker cannot reopen blocked tasks through the router surface."""
     from hermes_cli import kanban_db as kb
     conn = kb.connect()
     try:
@@ -923,7 +1013,7 @@ def test_worker_unblock_rejects_foreign_task_id(worker_env):
     from tools import kanban_tools as kt
     out = kt._handle_unblock({"task_id": other})
     d = json.loads(out)
-    assert "refusing to mutate" in d.get("error", "")
+    assert "orchestrator-only" in d.get("error", "")
 
     conn = kb.connect()
     try:
@@ -932,8 +1022,8 @@ def test_worker_unblock_rejects_foreign_task_id(worker_env):
         conn.close()
 
 
-def test_worker_archive_rejects_foreign_task_id(worker_env):
-    """A worker cannot archive a task that isn't its own."""
+def test_worker_archive_is_orchestrator_only(worker_env):
+    """A worker cannot archive tasks instead of completing or blocking."""
     from hermes_cli import kanban_db as kb
     conn = kb.connect()
     try:
@@ -944,7 +1034,7 @@ def test_worker_archive_rejects_foreign_task_id(worker_env):
     from tools import kanban_tools as kt
     out = kt._handle_archive({"task_id": other})
     d = json.loads(out)
-    assert "refusing to mutate" in d.get("error", "")
+    assert "orchestrator-only" in d.get("error", "")
 
     conn = kb.connect()
     try: