feat(kanban): allow orchestrator-only graph mutation policy

agent/agent_init.py

@@ -1476,9 +1476,10 @@ def init_agent(
     # Resolving the ~835-token block once here avoids re-running the
     # membership test + reference on every system-prompt rebuild
     # (init + each context compression).
-    from agent.prompt_builder import KANBAN_GUIDANCE
+    from agent.prompt_builder import kanban_guidance_for_tools
     agent._kanban_worker_guidance = (
-        KANBAN_GUIDANCE if "kanban_show" in agent.valid_tool_names else ""
+        kanban_guidance_for_tools(set(agent.valid_tool_names))
+        if "kanban_show" in agent.valid_tool_names else ""
     )
 
     # Check tool requirements

agent/prompt_builder.py

@@ -322,6 +322,50 @@ def _strip_yaml_frontmatter(content: str) -> str:
     "cross-agent handoffs that outlive one API loop."
 )
 
+
+def kanban_guidance_for_tools(valid_tool_names: set[str]) -> str:
+    """Match worker instructions to the graph-mutation tools actually exposed."""
+    if "kanban_create" in valid_tool_names and "kanban_link" in valid_tool_names:
+        return KANBAN_GUIDANCE
+    guidance = KANBAN_GUIDANCE.replace(
+        "6. **If follow-up work appears, create it; don't do it.** Use "
+        "`kanban_create(title=..., assignee=<right-profile>, parents=[your-task-id])` "
+        "to spawn a child task for the appropriate specialist profile instead of "
+        "scope-creeping into the next thing.\n",
+        "6. **If follow-up work appears, request it; don't create it.** Add a "
+        "`followup-request:` comment to your own task with the proposed owner, "
+        "scope, acceptance criteria, and dependency. The orchestrator is the sole "
+        "graph writer; do not create cards or dependency links.\n",
+    )
+    guidance = guidance.replace(
+        "If your task is itself a decomposition task (e.g. a planner profile given "
+        "a high-level goal), use `kanban_create` to fan out into child tasks — one "
+        "per specialist, each with an explicit `assignee` and `parents=[...]` to "
+        "express dependencies. Then `kanban_complete` your own task with a summary "
+        "of the decomposition. Do NOT execute the work yourself; your job is "
+        "routing, not implementation.\n",
+        "If your task uncovers a decomposition, record the proposed graph as a "
+        "`followup-request:` comment and complete/block with that handoff. Do NOT "
+        "fan out directly; the orchestrator deduplicates and owns graph mutation.\n",
+    )
+    guidance = guidance.replace(
+        "- **Created cards.** List ids in `kanban_complete(created_cards=[...])` "
+        "ONLY when captured from a successful `kanban_create` return — never invent "
+        "or paste ids; the kernel rejects the completion on any phantom id.\n",
+        "- **Follow-up requests.** Put proposed work in a structured "
+        "`followup-request:` comment; never claim or invent child-card ids.\n",
+    )
+    guidance = guidance.replace(
+        "- **Orchestrating: discover profiles first.** The dispatcher SILENTLY "
+        "drops a card with an unknown assignee (it sits in `ready` forever). Ground "
+        "every assignee in a real profile (`hermes profile list`, or ask the user), "
+        "and express dependencies via `parents=[...]` on `kanban_create`, not prose.\n",
+        "- **Routing requests.** Name a known profile when possible, but leave actual "
+        "card creation and dependency wiring to the orchestrator.\n",
+    )
+    return guidance
+
+
 TOOL_USE_ENFORCEMENT_GUIDANCE = (
     "# Tool-use enforcement\n"
     "You MUST use your tools to take action — do not describe what you would do "

agent/system_prompt.py

@@ -34,7 +34,6 @@
     DEFAULT_AGENT_IDENTITY,
     GOOGLE_MODEL_OPERATIONAL_GUIDANCE,
     HERMES_AGENT_HELP_GUIDANCE,
-    KANBAN_GUIDANCE,
     MEMORY_GUIDANCE,
     OPENAI_MODEL_EXECUTION_GUIDANCE,
     PARALLEL_TOOL_CALL_GUIDANCE,
@@ -47,6 +46,7 @@
     TOOL_USE_ENFORCEMENT_GUIDANCE,
     TOOL_USE_ENFORCEMENT_MODELS,
     drain_truncation_warnings,
+    kanban_guidance_for_tools,
 )
 from agent.runtime_cwd import resolve_context_cwd
 from hermes_constants import get_hermes_home
@@ -55,6 +55,16 @@
 logger = logging.getLogger(__name__)
 
 
+def _resolve_kanban_guidance(agent: Any) -> str:
+    """Return cached guidance or derive a tool-aware fallback."""
+    guidance = getattr(agent, "_kanban_worker_guidance", None)
+    if guidance is not None:
+        return guidance
+    if "kanban_show" not in agent.valid_tool_names:
+        return ""
+    return kanban_guidance_for_tools(set(agent.valid_tool_names))
+
+
 def _ra():
     """Lazy reference to the ``run_agent`` module.
 
@@ -235,12 +245,9 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
     # dispatcher spawned this process (kanban_show check_fn gates on
     # HERMES_KANBAN_TASK env var). Normal chat sessions never see
     # this block. Resolved once at __init__ (see _kanban_worker_guidance).
-    _kanban_guidance = getattr(agent, "_kanban_worker_guidance", None)
+    _kanban_guidance = _resolve_kanban_guidance(agent)
     if _kanban_guidance:
         tool_guidance.append(_kanban_guidance)
-    elif _kanban_guidance is None and "kanban_show" in agent.valid_tool_names:
-        # Fallback for code paths that bypass agent_init (rare).
-        tool_guidance.append(KANBAN_GUIDANCE)
     if tool_guidance:
         stable_parts.append(" ".join(tool_guidance))
 

hermes_cli/config_defaults.py

@@ -2340,6 +2340,11 @@
         # behaviour — e.g. for a profile that prefers explicit
         # ``kanban_notify-subscribe`` calls per task.
         "auto_subscribe_on_create": True,
+        # Allow dispatcher-scoped workers to create cards and add dependency
+        # links. True preserves the historical worker surface. Set false when
+        # one or more unscoped profiles with the `kanban` toolset own graph
+        # mutation; restricted workers use `followup-request:` comments instead.
+        "worker_graph_mutations": True,
         # Run the dispatcher inside the gateway process. On by default —
         # the cost is ~300µs every `dispatch_interval_seconds` when idle,
         # and gateway is the supervisor users already have. Set to false

tests/agent/test_prompt_builder.py

@@ -5,6 +5,7 @@
 import logging
 import os
 import sys
+from types import SimpleNamespace
 
 import pytest
 
@@ -55,6 +56,19 @@ def test_session_search_guidance_is_simple_cross_session_recall(self):
         assert "relevant cross-session context exists" in SESSION_SEARCH_GUIDANCE
         assert "recent turns of the current session" not in SESSION_SEARCH_GUIDANCE
 
+    def test_kanban_fallback_guidance_matches_restricted_tool_schema(self):
+        """Rare callers bypassing agent_init must still get tool-aware guidance."""
+        from agent.system_prompt import _resolve_kanban_guidance
+
+        agent = SimpleNamespace(
+            valid_tool_names={"kanban_show", "kanban_comment", "kanban_complete"},
+        )
+
+        guidance = _resolve_kanban_guidance(agent)
+
+        assert "followup-request:" in guidance
+        assert "use `kanban_create` to fan out" not in guidance
+
 
 # =========================================================================
 # Context injection scanning

tests/hermes_cli/test_config.py

@@ -1422,11 +1422,27 @@ def test_default_config_kanban_block_not_dropped_by_duplicate_key():
     kanban = DEFAULT_CONFIG["kanban"]
     # From the first (dropped) block:
     assert kanban.get("auto_subscribe_on_create") is True
+    # Existing installs remain permissive unless operators opt into
+    # orchestrator-only graph mutation.
+    assert kanban.get("worker_graph_mutations") is True
     # From the second block:
     assert "dispatch_in_gateway" in kanban
     assert "auto_decompose" in kanban
 
 
+def test_load_config_honors_worker_graph_mutation_override(tmp_path, monkeypatch):
+    """The policy is a normal ``kanban`` config value, not a stale side table."""
+    (tmp_path / "config.yaml").write_text(
+        "kanban:\n  worker_graph_mutations: false\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+    config = load_config()
+
+    assert config["kanban"]["worker_graph_mutations"] is False
+
+
 def test_default_config_has_no_duplicate_top_level_keys():
     """Guard against any duplicate key silently shadowing a default."""
     import ast

tests/tools/test_kanban_tools.py

@@ -40,6 +40,51 @@ def test_kanban_tools_hidden_without_env_var(monkeypatch, tmp_path):
     )
 
 
+def test_worker_graph_tools_remain_visible_by_default(monkeypatch, tmp_path):
+    """The opt-in restriction must not change existing worker schemas."""
+    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 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}
+
+    assert {"kanban_create", "kanban_link"} <= names
+
+
+def test_worker_graph_policy_hides_schema_and_updates_prompt(monkeypatch, tmp_path):
+    """Restricted workers see neither graph tools nor instructions to call them."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text(
+        "kanban:\n  worker_graph_mutations: false\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    import tools.kanban_tools  # ensure registered
+    from agent.prompt_builder import kanban_guidance_for_tools
+    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}
+    guidance = kanban_guidance_for_tools(names)
+
+    assert {"kanban_create", "kanban_link"}.isdisjoint(names)
+    assert {"kanban_show", "kanban_comment", "kanban_complete"} <= names
+    assert "followup-request:" in guidance
+    assert "use `kanban_create` to fan out" not in guidance
+
+
 # ---------------------------------------------------------------------------
 # Handler happy paths
 # ---------------------------------------------------------------------------
@@ -69,6 +114,86 @@ def worker_env(monkeypatch, tmp_path):
     return tid
 
 
+def test_worker_graph_policy_rejects_create_and_link_without_db_mutation(worker_env):
+    """Handler checks enforce the policy even if a caller bypasses schema hiding."""
+    from hermes_constants import get_hermes_home
+    from hermes_cli import config as cfg_mod
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    (get_hermes_home() / "config.yaml").write_text(
+        "kanban:\n  worker_graph_mutations: false\n",
+        encoding="utf-8",
+    )
+    setattr(cfg_mod, "_cached_config", None)
+
+    conn = kb.connect()
+    try:
+        other = kb.create_task(conn, title="existing", assignee="peer")
+        before_tasks = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]
+    finally:
+        conn.close()
+
+    created = json.loads(kt._handle_create({"title": "forbidden", "assignee": "peer"}))
+    linked = json.loads(kt._handle_link({"parent_id": worker_env, "child_id": other}))
+
+    assert "followup-request" in created.get("error", "")
+    assert "followup-request" in linked.get("error", "")
+    conn = kb.connect()
+    try:
+        assert conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0] == before_tasks
+        assert conn.execute(
+            "SELECT parent_id FROM task_links WHERE child_id=?", (other,)
+        ).fetchall() == []
+    finally:
+        conn.close()
+
+
+def test_orchestrator_retains_graph_mutations_when_worker_policy_is_disabled(
+    monkeypatch, tmp_path
+):
+    """The worker restriction must not remove the configured control plane."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text(
+        "toolsets:\n  - kanban\nkanban:\n  worker_graph_mutations: false\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setenv("HERMES_PROFILE", "orchestrator")
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from pathlib import Path as _Path
+    monkeypatch.setattr(_Path, "home", lambda: tmp_path)
+
+    from hermes_cli import config as cfg_mod
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+    from tools.registry import invalidate_check_fn_cache, registry
+    from toolsets import resolve_toolset
+
+    setattr(cfg_mod, "_cached_config", None)
+    invalidate_check_fn_cache()
+    schema = registry.get_definitions(set(resolve_toolset("kanban")), quiet=True)
+    names = {s["function"].get("name") for s in schema if "function" in s}
+    assert {"kanban_create", "kanban_link"} <= names
+
+    kb._INITIALIZED_PATHS.clear()
+    kb.init_db()
+    conn = kb.connect()
+    try:
+        parent = kb.create_task(conn, title="parent", assignee="orchestrator")
+    finally:
+        conn.close()
+
+    created = json.loads(kt._handle_create({"title": "child", "assignee": "dev"}))
+    assert created.get("ok") is True
+    linked = json.loads(kt._handle_link({
+        "parent_id": parent,
+        "child_id": created["task_id"],
+    }))
+    assert linked.get("ok") is True
+
+
 def test_show_defaults_to_env_task_id(worker_env):
     from tools import kanban_tools as kt
     out = kt._handle_show({})

tools/kanban_tools.py

@@ -135,6 +135,42 @@ def _check_kanban_orchestrator_mode() -> bool:
     return _profile_has_kanban_toolset()
 
 
+def _worker_graph_mutations_enabled() -> bool:
+    """Resolve the task-worker graph policy, preserving the legacy default."""
+    try:
+        return bool(cfg_get(
+            load_config(), "kanban", "worker_graph_mutations", default=True,
+        ))
+    except Exception:
+        # A config read failure must not silently revoke a previously available
+        # worker capability. load_config itself reports malformed config.
+        return True
+
+
+def _check_kanban_graph_mutation_mode() -> bool:
+    """Expose create/link to orchestrators and, when allowed, task workers."""
+    if _is_delegated_child_context():
+        return False
+    if os.environ.get("HERMES_KANBAN_TASK") and _is_dispatcher_owned_worker():
+        return _worker_graph_mutations_enabled()
+    return _profile_has_kanban_toolset()
+
+
+def _enforce_worker_graph_mutation_policy() -> Optional[str]:
+    """Enforce schema policy again at the handler security boundary."""
+    if (
+        os.environ.get("HERMES_KANBAN_TASK")
+        and _is_dispatcher_owned_worker()
+        and not _worker_graph_mutations_enabled()
+    ):
+        return tool_error(
+            "kanban graph mutations are reserved for an orchestrator on this "
+            "installation; add a followup-request: comment to your current task "
+            "instead of creating or linking cards"
+        )
+    return None
+
+
 # ---------------------------------------------------------------------------
 # Shared helpers
 # ---------------------------------------------------------------------------
@@ -1349,6 +1385,9 @@ def _handle_create(args: dict, **kw) -> str:
     delegated_err = _reject_delegated_child_mutation("kanban_create")
     if delegated_err:
         return delegated_err
+    policy_err = _enforce_worker_graph_mutation_policy()
+    if policy_err:
+        return policy_err
     title = args.get("title")
     if not title or not str(title).strip():
         return tool_error("title is required")
@@ -1642,6 +1681,9 @@ def _handle_link(args: dict, **kw) -> str:
     delegated_err = _reject_delegated_child_mutation("kanban_link")
     if delegated_err:
         return delegated_err
+    policy_err = _enforce_worker_graph_mutation_policy()
+    if policy_err:
+        return policy_err
     parent_id = args.get("parent_id")
     child_id = args.get("child_id")
     if not parent_id or not child_id:
@@ -2453,7 +2495,7 @@ def _board_schema_prop() -> dict[str, str]:
     toolset="kanban",
     schema=KANBAN_CREATE_SCHEMA,
     handler=_handle_create,
-    check_fn=_check_kanban_mode,
+    check_fn=_check_kanban_graph_mutation_mode,
     emoji="➕",
 )
 
@@ -2471,6 +2513,6 @@ def _board_schema_prop() -> dict[str, str]:
     toolset="kanban",
     schema=KANBAN_LINK_SCHEMA,
     handler=_handle_link,
-    check_fn=_check_kanban_mode,
+    check_fn=_check_kanban_graph_mutation_mode,
     emoji="🔗",
 )