fix(kanban): restrict board routing tools to orchestrators

Adapted from PR #20568 commit ce3518578 (Eric Litovsky / @kallidean).
Adds two-tier gating for the kanban tool surface so dispatcher-spawned
workers see only task-lifecycle tools (show/complete/block/heartbeat/
comment/create/link) while orchestrator profiles with `toolsets: [kanban]`
also see board-routing tools (kanban_list, kanban_unblock).

Workers shouldn't be enumerating or unblocking the board — they should
close their own task via the lifecycle tools. Hiding board-routing tools
from worker schemas keeps the worker focused and the toolset-isolation
contract honest.

Plus inherited from the same upstream commit:
- 50/200 row bound on kanban_list with `truncated` + `next_limit` metadata.
- Belt-and-suspenders runtime guard `_require_orchestrator_tool()` inside
  the orchestrator handlers in case a stale registration ever routes a
  worker to one of them.
- Tests for the new gate, the stricter bound, and the fact that even a
  worker with `toolsets: [kanban]` in config still doesn't see board
  routing.

Co-authored-by: Eric Litovsky <elitovsky@zenproject.net>

tests/tools/test_kanban_tools.py

@@ -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()
@@ -50,6 +50,59 @@ def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path):
     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_show", "kanban_complete", "kanban_block", "kanban_heartbeat",
+        "kanban_comment", "kanban_create", "kanban_link",
+    }
+    assert kanban == expected, f"expected {expected}, got {kanban}"
+
+
+def test_worker_with_kanban_toolset_still_hides_board_routing(monkeypatch, tmp_path):
+    """Task scope wins over profile config for board-routing tools.
+
+    Even if a worker process happens to also have ``toolsets: [kanban]``
+    in its config, the HERMES_KANBAN_TASK env var means it's a focused
+    worker and must not see kanban_list / kanban_unblock.
+    """
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
+    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_")}
+    assert {
+        "kanban_list",
+        "kanban_unblock",
+    }.isdisjoint(kanban), (
+        f"Board-routing tools leaked into worker schema: "
+        f"{kanban & {'kanban_list', 'kanban_unblock'}}"
+    )
+
+
+def test_kanban_tools_visible_with_toolset_config(monkeypatch, tmp_path):
+    """Orchestrator profiles with toolsets: [kanban] see all kanban 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}
@@ -63,28 +116,6 @@ def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path):
     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)
-    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_")}
-    assert {
-        "kanban_list",
-        "kanban_unblock",
-    }.issubset(kanban)
-
-
 # ---------------------------------------------------------------------------
 # Handler happy paths
 # ---------------------------------------------------------------------------
@@ -138,8 +169,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:
@@ -168,19 +200,22 @@ 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")
 
 
-def test_list_parses_include_archived_string_false(worker_env):
+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:
@@ -200,7 +235,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:
@@ -220,7 +256,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", "")
@@ -641,7 +678,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")
@@ -899,7 +937,13 @@ def test_worker_can_comment_on_foreign_task(worker_env):
 
 
 def test_worker_unblock_rejects_foreign_task_id(worker_env):
-    """A worker cannot unblock a task that isn't its own."""
+    """A worker cannot unblock any task — kanban_unblock is orchestrator-only.
+
+    The check fires before the per-task ownership check, so the error
+    surface is the orchestrator-only refusal rather than the
+    cross-task-ownership refusal. Either is fine — the property we're
+    pinning is "worker cannot mutate foreign task via kanban_unblock".
+    """
     from hermes_cli import kanban_db as kb
     conn = kb.connect()
     try:
@@ -911,7 +955,10 @@ 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", "")
+    err = d.get("error", "")
+    assert "orchestrator-only" in err or "refusing to mutate" in err, (
+        f"expected worker-rejection error, got {err}"
+    )
 
     conn = kb.connect()
     try: