From 65056612f9e0e8503f2f190195b72dacb4c3ec16 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 21 May 2026 13:27:51 -0700 Subject: [PATCH] fix(kanban): restrict graph mutation to software-engineer tier-1: option a strict kanban graph fanout control --- agent/prompt_builder.py | 24 +++-- tests/tools/test_kanban_tools.py | 169 ++++++++++++++++++++++--------- tools/kanban_tools.py | 66 ++++++++++-- 3 files changed, 189 insertions(+), 70 deletions(-) diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index 9c36d205ac5bb..5e3c505d5f51d 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -230,19 +230,21 @@ def _strip_yaml_frontmatter(content: str) -> str: "reviewer can approve+unblock or request changes. Reviewing-then-" "completing is more honest than auto-completing work that still needs " "eyes on it.\n" - "6. **If follow-up work appears, create it; don't do it.** Use " - "`kanban_create(title=..., assignee=, parents=[your-task-id])` " - "to spawn a child task for the appropriate specialist profile instead of " - "scope-creeping into the next thing.\n" + "6. **Do not grow the graph unless you are software-engineer.** Only " + "dispatcher workers running with `HERMES_PROFILE=software-engineer` may use " + "`kanban_create` or `kanban_link` to decompose implementation work into " + "child tasks. All other workers must record follow-up work in " + "`kanban_comment`, include it in completion metadata, or " + "`kanban_block(reason=...)` if the follow-up is required before completion.\n" "\n" - "## Orchestrator mode\n" + "## Software-engineer decomposition mode\n" "\n" - "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 you are `software-engineer` and your assigned task is decomposition, use " + "`kanban_create` to fan out into child tasks — one per specialist, each " + "with an explicit `assignee` and `parents=[...]` to express dependencies. " + "Use `kanban_link` only to repair or join that dependency topology. Then " + "`kanban_complete` your own task with a summary of the decomposition. Do NOT " + "execute the child work yourself; your job is routing, not implementation.\n" "\n" "## Do NOT\n" "\n" diff --git a/tests/tools/test_kanban_tools.py b/tests/tools/test_kanban_tools.py index 80b08377ab51c..dd0acd436712f 100644 --- a/tests/tools/test_kanban_tools.py +++ b/tests/tools/test_kanban_tools.py @@ -40,8 +40,32 @@ def test_kanban_tools_hidden_without_env_var(monkeypatch, tmp_path): def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path): - """Worker sessions get task lifecycle tools, not board-routing tools.""" + """Generic worker sessions get lifecycle tools, not graph-routing tools.""" monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake") + monkeypatch.setenv("HERMES_PROFILE", "test-worker") + 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} + kanban = {n for n in names if n and n.startswith("kanban_")} + expected = { + "kanban_show", "kanban_complete", "kanban_block", "kanban_heartbeat", + "kanban_comment", + } + assert kanban == expected, f"expected {expected}, got {kanban}" + + +def test_software_engineer_worker_gets_graph_tools(monkeypatch, tmp_path): + """Only the software-engineer worker profile sees task graph tools.""" + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake") + monkeypatch.setenv("HERMES_PROFILE", "software-engineer") home = tmp_path / ".hermes" home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) @@ -146,14 +170,13 @@ def test_kanban_tools_visible_with_toolset_config(monkeypatch, tmp_path): # Handler happy paths # --------------------------------------------------------------------------- -@pytest.fixture -def worker_env(monkeypatch, tmp_path): +def _make_worker_env(monkeypatch, tmp_path, *, profile: str): """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") + monkeypatch.setenv("HERMES_PROFILE", profile) monkeypatch.delenv("HERMES_SESSION_ID", raising=False) from pathlib import Path as _Path monkeypatch.setattr(_Path, "home", lambda: tmp_path) @@ -163,7 +186,7 @@ def worker_env(monkeypatch, tmp_path): kb.init_db() conn = kb.connect() try: - tid = kb.create_task(conn, title="worker-test", assignee="test-worker") + tid = kb.create_task(conn, title="worker-test", assignee=profile) kb.claim_task(conn, tid) finally: conn.close() @@ -171,6 +194,16 @@ def worker_env(monkeypatch, tmp_path): return tid +@pytest.fixture +def worker_env(monkeypatch, tmp_path): + return _make_worker_env(monkeypatch, tmp_path, profile="test-worker") + + +@pytest.fixture +def software_engineer_env(monkeypatch, tmp_path): + return _make_worker_env(monkeypatch, tmp_path, profile="software-engineer") + + def test_show_defaults_to_env_task_id(worker_env): from tools import kanban_tools as kt out = kt._handle_show({}) @@ -559,7 +592,7 @@ def test_complete_retry_with_empty_created_cards_succeeds(worker_env): conn.close() -def test_complete_retry_with_corrected_created_cards_succeeds(worker_env): +def test_complete_retry_with_corrected_created_cards_succeeds(software_engineer_env): """After a phantom rejection, retrying kanban_complete with a corrected created_cards list (phantom ids removed) must complete the task. Regression for #22923.""" @@ -569,7 +602,8 @@ def test_complete_retry_with_corrected_created_cards_succeeds(worker_env): # Create a real child via the tool so it gets the worker-profile # attribution the gate trusts. child = json.loads(kt._handle_create({ - "title": "real child", "assignee": "peer", + "title": "real child", "assignee": "coder", + "parents": [software_engineer_env], })) assert child["ok"] real_id = child["task_id"] @@ -591,7 +625,7 @@ def test_complete_retry_with_corrected_created_cards_succeeds(worker_env): conn = kb.connect() try: - assert kb.get_task(conn, worker_env).status == "done" + assert kb.get_task(conn, software_engineer_env).status == "done" finally: conn.close() @@ -747,13 +781,51 @@ def test_comment_schema_omits_author_override(): assert "author" not in props -def test_create_happy_path(worker_env): +def test_create_rejects_non_software_engineer_worker(worker_env): + """Option A topology: generic workers cannot fan out the task graph. + + Only the software-engineer decomposition role may create child cards from + inside a dispatcher-scoped worker. Other workers should comment/block + instead of spawning follow-up tasks. + """ + from hermes_cli import kanban_db as kb from tools import kanban_tools as kt + out = kt._handle_create({ - "title": "child task", - "assignee": "peer", + "title": "unauthorized child", + "assignee": "coder", "parents": [worker_env], }) + err = json.loads(out).get("error", "") + assert "software-engineer" in err + with kb.connect() as conn: + rows = kb.list_tasks(conn) + assert [t.title for t in rows] == ["worker-test"] + + +def test_link_rejects_non_software_engineer_worker(worker_env): + """Non software-engineer workers cannot mutate dependency topology.""" + from hermes_cli import kanban_db as kb + from tools import kanban_tools as kt + + with kb.connect() as conn: + a = kb.create_task(conn, title="A", assignee="x") + b = kb.create_task(conn, title="B", assignee="x") + + out = kt._handle_link({"parent_id": a, "child_id": b}) + err = json.loads(out).get("error", "") + assert "software-engineer" in err + with kb.connect() as conn: + assert b not in kb.child_ids(conn, a) + + +def test_create_happy_path(software_engineer_env): + from tools import kanban_tools as kt + out = kt._handle_create({ + "title": "child task", + "assignee": "coder", + "parents": [software_engineer_env], + }) d = json.loads(out) assert d["ok"] is True assert d["task_id"] @@ -763,12 +835,12 @@ def test_create_happy_path(worker_env): try: child = kb.get_task(conn, d["task_id"]) assert child.title == "child task" - assert child.assignee == "peer" + assert child.assignee == "coder" finally: conn.close() -def test_create_stamps_session_id_from_env(monkeypatch, worker_env): +def test_create_stamps_session_id_from_env(monkeypatch, software_engineer_env): """When the agent loop runs under ACP, the server propagates the originating chat session id via HERMES_SESSION_ID. ``kanban_create`` reads it and stamps the new task so clients can render a per-session @@ -778,8 +850,8 @@ def test_create_stamps_session_id_from_env(monkeypatch, worker_env): from hermes_cli import kanban_db as kb out = kt._handle_create({ "title": "from chat", - "assignee": "peer", - "parents": [worker_env], + "assignee": "coder", + "parents": [software_engineer_env], }) d = json.loads(out) assert d["ok"] is True @@ -791,7 +863,7 @@ def test_create_stamps_session_id_from_env(monkeypatch, worker_env): conn.close() -def test_create_session_id_arg_overrides_env(monkeypatch, worker_env): +def test_create_session_id_arg_overrides_env(monkeypatch, software_engineer_env): """An explicit ``session_id`` arg from the model wins over the env propagation. Edge case but exercised: a tool call could carry a different session id (e.g. cross-session linking) and the explicit @@ -801,8 +873,8 @@ def test_create_session_id_arg_overrides_env(monkeypatch, worker_env): from hermes_cli import kanban_db as kb out = kt._handle_create({ "title": "explicit override", - "assignee": "peer", - "parents": [worker_env], + "assignee": "coder", + "parents": [software_engineer_env], "session_id": "explicit-arg", }) d = json.loads(out) @@ -815,7 +887,7 @@ def test_create_session_id_arg_overrides_env(monkeypatch, worker_env): conn.close() -def test_create_session_id_absent_when_env_unset(monkeypatch, worker_env): +def test_create_session_id_absent_when_env_unset(monkeypatch, software_engineer_env): """No env var, no arg → session_id stays NULL. Important for backwards compatibility: pre-ACP-propagation hosts and CLI-driven creates must not accidentally inherit a stale id.""" @@ -824,8 +896,8 @@ def test_create_session_id_absent_when_env_unset(monkeypatch, worker_env): from hermes_cli import kanban_db as kb out = kt._handle_create({ "title": "no session", - "assignee": "peer", - "parents": [worker_env], + "assignee": "coder", + "parents": [software_engineer_env], }) d = json.loads(out) assert d["ok"] is True @@ -854,12 +926,12 @@ def test_create_rejects_non_list_parents(worker_env): assert json.loads(out).get("error") -def test_create_parses_triage_string_false(worker_env): +def test_create_parses_triage_string_false(software_engineer_env): from tools import kanban_tools as kt from hermes_cli import kanban_db as kb out = kt._handle_create({ "title": "not triage", - "assignee": "peer", + "assignee": "coder", "triage": "false", }) d = json.loads(out) @@ -872,12 +944,12 @@ def test_create_parses_triage_string_false(worker_env): conn.close() -def test_create_parses_triage_string_true(worker_env): +def test_create_parses_triage_string_true(software_engineer_env): from tools import kanban_tools as kt from hermes_cli import kanban_db as kb out = kt._handle_create({ "title": "needs triage", - "assignee": "peer", + "assignee": "coder", "triage": "true", }) d = json.loads(out) @@ -890,7 +962,7 @@ def test_create_parses_triage_string_true(worker_env): conn.close() -def test_create_rejects_bad_triage(worker_env): +def test_create_rejects_bad_triage(software_engineer_env): from tools import kanban_tools as kt out = kt._handle_create({ "title": "bad triage", @@ -900,22 +972,22 @@ def test_create_rejects_bad_triage(worker_env): assert "triage must be" in json.loads(out).get("error", "") -def test_create_accepts_string_parent(worker_env): +def test_create_accepts_string_parent(software_engineer_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, + "title": "t", "assignee": "coder", "parents": software_engineer_env, }) assert json.loads(out)["ok"] -def test_create_accepts_skills_list(worker_env): +def test_create_accepts_skills_list(software_engineer_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", + "assignee": "coder", "skills": ["translation", "github-code-review"], }) d = json.loads(out) @@ -925,13 +997,13 @@ def test_create_accepts_skills_list(worker_env): assert task.skills == ["translation", "github-code-review"] -def test_create_accepts_skills_string(worker_env): +def test_create_accepts_skills_string(software_engineer_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", + "assignee": "coder", "skills": "translation", }) d = json.loads(out) @@ -950,7 +1022,7 @@ def test_create_rejects_non_list_skills(worker_env): assert json.loads(out).get("error") -def test_link_happy_path(worker_env): +def test_link_happy_path(software_engineer_env): from hermes_cli import kanban_db as kb conn = kb.connect() try: @@ -976,7 +1048,7 @@ def test_link_rejects_missing_args(worker_env): assert json.loads(kt._handle_link({"child_id": "y"})).get("error") -def test_link_rejects_cycle(worker_env): +def test_link_rejects_cycle(software_engineer_env): """A → B, then try to link B → A.""" from hermes_cli import kanban_db as kb conn = kb.connect() @@ -1039,18 +1111,19 @@ def test_worker_lifecycle_through_tools(worker_env): "body": "note: using stdlib sqlite3 bindings", }))["ok"] - # 4. spawn a child task for follow-up - child_out = json.loads(kt._handle_create({ + # 4. generic workers do not spawn follow-up cards; they record handoff + # information in comments/metadata and then complete or block. + denied = json.loads(kt._handle_create({ "title": "write integration test", "assignee": "qa", "parents": [worker_env], })) - assert child_out["ok"] + assert denied.get("error") # 5. complete with structured handoff comp = json.loads(kt._handle_complete({ - "summary": "implemented + spawned QA follow-up", - "metadata": {"child_task": child_out["task_id"]}, + "summary": "implemented; QA follow-up noted in comments, not spawned", + "metadata": {"follow_up": "write integration test"}, })) assert comp["ok"] @@ -1063,13 +1136,9 @@ def test_worker_lifecycle_through_tools(worker_env): 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}" - ) + assert run.metadata == {"follow_up": "write integration test"} + # No child task was created by the denied follow-up attempt. + assert len([t for t in kb.list_tasks(conn) if t.title == "write integration test"]) == 0 # Comment is visible assert len(kb.list_comments(conn, worker_env)) == 1 # Heartbeat event recorded @@ -1142,6 +1211,8 @@ def test_kanban_guidance_in_worker_prompt(monkeypatch, tmp_path): assert "kanban_complete" in prompt assert "kanban_block" in prompt assert "kanban_create" in prompt + assert "HERMES_PROFILE=software-engineer" in prompt + assert "All other workers must record follow-up work" in prompt # Anti-shell guidance assert "Do not shell out" in prompt or "tools — they work" in prompt @@ -1170,9 +1241,9 @@ def test_kanban_guidance_prompt_size_bounded(monkeypatch, tmp_path): # destructive tools (kanban_complete, kanban_block, kanban_heartbeat, # kanban_unblock) 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. +# argument. Workers legitimately call kanban_show / kanban_comment on other +# tasks, while graph tools (kanban_create / kanban_link) are limited to the +# software-engineer decomposition role. # # Orchestrator profiles (no HERMES_KANBAN_TASK in env) are intentionally # exempt — their job is routing, and they sometimes close out child diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index 29b5618e6815b..b885cf90addae 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -44,6 +44,7 @@ KANBAN_LIST_DEFAULT_LIMIT = 50 KANBAN_LIST_MAX_LIMIT = 200 +SOFTWARE_ENGINEER_PROFILE = "software-engineer" def _profile_has_kanban_toolset() -> bool: @@ -90,6 +91,21 @@ def _check_kanban_orchestrator_mode() -> bool: return _profile_has_kanban_toolset() +def _check_kanban_graph_mode() -> bool: + """Graph mutation tools are visible only to orchestrators or software-engineer. + + All dispatcher-spawned workers need lifecycle tools, but only the + software-engineer decomposition role should be able to create child cards + or mutate dependency edges from inside a worker run. Orchestrator profiles + with the kanban toolset still see these tools outside task scope. + """ + if not _check_kanban_mode(): + return False + if not os.environ.get("HERMES_KANBAN_TASK"): + return True + return (os.environ.get("HERMES_PROFILE") or "").strip() == SOFTWARE_ENGINEER_PROFILE + + # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- @@ -222,6 +238,29 @@ def _require_orchestrator_tool(tool_name: str) -> Optional[str]: return None +def _require_software_engineer_graph_tool(tool_name: str) -> Optional[str]: + """Restrict worker-side graph mutation to software-engineer. + + Orchestrator contexts (no ``HERMES_KANBAN_TASK``) remain able to route + board work. Dispatcher-scoped workers, however, must not grow or rewire + the DAG unless they are the implementation decomposition role. This is + the hard runtime counterpart to the Mosaiq full-pipeline contract: normal + workers comment/block/complete; software-engineer creates the parallel + implementation fanout and join structure. + """ + if not os.environ.get("HERMES_KANBAN_TASK"): + return None + profile = (os.environ.get("HERMES_PROFILE") or "").strip() + if profile == SOFTWARE_ENGINEER_PROFILE: + return None + return tool_error( + f"{tool_name} is restricted to the {SOFTWARE_ENGINEER_PROFILE} " + "decomposition role for dispatcher-spawned workers. This worker " + f"is {profile or 'unknown'}; add a kanban_comment or kanban_block " + "instead of spawning or rewiring child tasks." + ) + + def _task_summary_dict(kb, conn, task) -> dict[str, Any]: """Compact task shape for board-listing tools.""" parents = kb.parent_ids(conn, task.id) @@ -637,11 +676,15 @@ def _handle_comment(args: dict, **kw) -> str: def _handle_create(args: dict, **kw) -> str: - """Create a child task. Orchestrator workers use this to fan out. + """Create a child task. Software-engineer workers use this to fan out. ``parents`` can be a list of task ids; dependency-gated promotion - works as usual. + works as usual. Dispatcher-scoped non-software-engineer workers are + denied by policy and should comment/block instead of growing the graph. """ + guard = _require_software_engineer_graph_tool("kanban_create") + if guard: + return guard title = args.get("title") if not title or not str(title).strip(): return tool_error("title is required") @@ -750,6 +793,9 @@ def _handle_unblock(args: dict, **kw) -> str: def _handle_link(args: dict, **kw) -> str: """Add a parent→child dependency edge after the fact.""" + guard = _require_software_engineer_graph_tool("kanban_link") + if guard: + return guard parent_id = args.get("parent_id") child_id = args.get("child_id") if not parent_id or not child_id: @@ -1048,11 +1094,10 @@ def _board_schema_prop() -> dict[str, str]: "name": "kanban_create", "description": ( "Create a new kanban task, optionally as a child of the current " - "one (pass the current task id in ``parents``). Used by " - "orchestrator workers to fan out — decompose work into child " - "tasks with specific assignees, link them into a pipeline, " - "then complete your own task. The dispatcher picks up the new " - "tasks on its next tick and spawns the assigned profiles." + "one (pass the current task id in ``parents``). Orchestrators can " + "route board work outside task scope. In dispatcher worker mode, " + "only the software-engineer decomposition role may create child " + "tasks; other workers should add a comment or block instead." ), "parameters": { "type": "object", @@ -1197,7 +1242,8 @@ def _board_schema_prop() -> dict[str, str]: "description": ( "Add a parent→child dependency edge after both tasks already " "exist. The child won't promote to 'ready' until all parents " - "are 'done'. Cycles and self-links are rejected." + "are 'done'. Cycles and self-links are rejected. In dispatcher " + "worker mode, only software-engineer may mutate dependency topology." ), "parameters": { "type": "object", @@ -1274,7 +1320,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_mode, emoji="➕", ) @@ -1292,6 +1338,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_mode, emoji="🔗", )