External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

[chore] orchestrator: refuse to spawn human-only roles (CEO/prompter/secretary)

Browse Source 
A live 2026-06-27 incident saw a CEO agent container spawned. Root cause:
_dispatch_a2a_work iterates every A2A/notification target and spawns it
with no human-role filter, and _is_agent_active('ceo') is always false
(the CEO is never a container), so the 'skip if active' check could never
protect the CEO. Any CEO-addressed notification (board handoff, escalation)
launched a CEO container — the system acting as the human CEO: a trust
violation. The CEO is the human operator; intake (prompter) and secretary
are human-driven chats launched through their own dedicated guarded paths
(_spawn_intake_container / _spawn_secretary_container), never spawn_agent.

Fix: a single chokepoint guard at the top of spawn_agent refuses
Role.CEO / PROMPTER / SECRETARY (raises AgentReadinessError + logs). This
structurally covers every dispatcher present and future, since they all go
through spawn_agent. Plus a defense-in-depth skip in _dispatch_a2a_work so
a human-role target never even calls in (avoids error-log spam; the
notification stays for the human to read in the panel).

Safe: the dedicated human-spawn paths do not route through spawn_agent.
Regression tests: spawn_agent refuses ceo/intake-1/secretary-1, does NOT
refuse a real agent; _dispatch_a2a_work skips CEO/intake/secretary targets
and still spawns real-agent + mixed-target cases.
This commit is contained in:
Renn F
2026-06-28 06:45:16 +02:00
parent 431f2c72e0
commit d31d6719cf
2 changed files with 210 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
roboco/runtime/orchestrator.py
+43 -1
View File
@@ -46,7 +46,7 @@ from roboco.agents_config import (
)
from roboco.config import settings
from roboco.foundation import identity as _foundation
from roboco.foundation.identity import CELL_TEAMS
from roboco.foundation.identity import CELL_TEAMS, Role, role_for_slug
from roboco.foundation.policy.agent_loop import DEFAULT_BUDGET as _AGENT_LOOP_BUDGET
from roboco.foundation.policy.batch import is_branchless_coordination
from roboco.models import AgentRole, Team
@@ -1853,6 +1853,33 @@ class AgentOrchestrator:
is auto-blocked before we raise so the dispatcher doesn't
keep retrying.
"""
# Human-only roles (ceo / prompter / secretary) are NEVER spawned by a
# dispatcher. The CEO is the human operator; intake (prompter) and
# secretary are human-driven interactive chats launched through their
# own deliberately-separate guarded paths (_spawn_intake_container /
# _spawn_secretary_container), NOT through this method. A dispatcher
# that spawns "any A2A/notification target" (e.g. _dispatch_a2a_work)
# could otherwise resolve a CEO-addressed notification to slug "ceo" and
# launch a CEO container — a trust violation (the system acting as the
# human CEO). This chokepoint guard is the single structural fix: no
# matter which dispatcher calls in, a human role never gets a container
# here. Safe because the dedicated human-spawn paths do not route
# through spawn_agent (see the _spawn_intake_container note at the top
# of this file).
_role = role_for_slug(agent_id)
if _role in (Role.CEO, Role.PROMPTER, Role.SECRETARY):
logger.error(
"spawn_agent refused for human-only role — dispatchers must never"
" spawn the CEO / prompter / secretary; these are human-driven",
agent_id=agent_id,
role=str(_role),
task_id=str(task_id) if task_id else None,
)
raise AgentReadinessError(
f"refused to spawn human-only role {_role!r} ({agent_id}) — the"
f" CEO is the human operator, not a container; intake and secretary"
f" launch through their dedicated paths, not spawn_agent"
)
# Pre-flight: refuse to spawn if the task isn't ready. Auto-block
# on refusal so the dispatcher doesn't keep spinning a container
# that will immediately fail (wasted image pull + startup tokens).
@@ -9973,6 +10000,21 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
# Resolve UUID to slug - to_agents contains UUIDs from database
agent_slug = self._resolve_agent_slug(str(agent_id))
# Human-only roles (CEO / prompter / secretary) are never
# dispatched — the CEO is the human operator and intake/
# secretary are human-driven chats with their own launch
# paths. Spawning a container for one is a trust violation
# (the system acting as the human CEO). The CEO being a
# notification target (board-review handoff, escalation, etc.)
# is expected; it is NOT a spawn signal. Skip — the
# notification stays for the human to read in the panel.
if role_for_slug(agent_slug) in (
Role.CEO,
Role.PROMPTER,
Role.SECRETARY,
):
continue
if self._is_agent_active(agent_slug):
# Agent is online - SDK handles A2A delivery directly
# No action needed here, SDK server receives messages
tests/unit/runtime/test_no_spawn_human_roles.py
+167
View File
@@ -0,0 +1,167 @@
"""Human-only roles (CEO / prompter / secretary) are NEVER spawned.
The CEO is the human operator, not a container; intake (prompter) and
secretary are human-driven interactive chats launched through their own
guarded paths (_spawn_intake_container / _spawn_secretary_container). A
live 2026-06-27 incident saw a CEO container spawned by _dispatch_a2a_work:
that dispatcher iterates every A2A/notification target and spawns it, and
`_is_agent_active("ceo")` is *always* false (the CEO is never a container),
so the "skip if active" check could never protect the CEO — any
CEO-addressed notification launched a CEO container (the system acting as
the human CEO: a trust violation).
The fix is a single chokepoint guard at the top of `spawn_agent` that
refuses Role.CEO / PROMPTER / SECRETARY, plus a defense-in-depth skip in
`_dispatch_a2a_work` so the dispatcher never even calls in for a human
target (avoids error-log spam — the notification stays for the human to
read in the panel). The chokepoint is structural: every dispatcher present
or future is covered, because they all go through `spawn_agent`.
"""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock
import pytest
from roboco.foundation.identity import Role, role_for_slug
from roboco.runtime.orchestrator import AgentOrchestrator, AgentReadinessError
from roboco.seeds.initial_data import AGENT_UUIDS
def _orch() -> AgentOrchestrator:
# The human-role guard is the first statement in spawn_agent and only
# consults the pure `role_for_slug` + the module logger — no self state
# — so a bare (un-initialized) orchestrator is sufficient to exercise it.
orch = object.__new__(AgentOrchestrator)
return orch
# ---------------------------------------------------------------------------
# spawn_agent chokepoint — refuses human-only roles
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"human_slug",
["ceo", "intake-1", "secretary-1"],
)
@pytest.mark.asyncio
async def test_spawn_agent_refuses_human_only_roles(human_slug: str) -> None:
"""spawn_agent must refuse the CEO / prompter / secretary — a dispatcher
must never launch a container for a human-only role."""
orch = _orch()
with pytest.raises(AgentReadinessError, match="human-only role"):
await orch.spawn_agent(human_slug)
# Sanity: the slug really is a human-only role (guards against the test
# silently passing because role_for_slug returned something unexpected).
assert role_for_slug(human_slug) in (Role.CEO, Role.PROMPTER, Role.SECRETARY)
@pytest.mark.asyncio
async def test_spawn_agent_does_not_refuse_real_agent() -> None:
"""A real (container-eligible) agent must NOT trip the human-role guard.
It may still be refused by the downstream readiness gate, but the
refusal must NOT carry the human-only-role message. This proves the
guard is scoped to human roles only and doesn't over-block the fleet.
"""
orch = _orch()
# Stub the readiness gate so we isolate the human-role guard: a real
# agent passes the guard and reaches (and is stopped by) readiness,
# which is a *different* refusal than the human-role one.
async def _ready(_aid: str, _tid: str | None) -> str | None:
return "stubbed-not-ready"
orch._readiness_gate = _ready # type: ignore[assignment]
with pytest.raises(AgentReadinessError) as exc_info:
await orch.spawn_agent("be-dev-1", task_id="t-1")
assert "human-only role" not in str(exc_info.value)
assert role_for_slug("be-dev-1") not in (Role.CEO, Role.PROMPTER, Role.SECRETARY)
# ---------------------------------------------------------------------------
# _dispatch_a2a_work — skips human-only targets (defense-in-depth)
# ---------------------------------------------------------------------------
def _a2a_orch(ceo_uuid: str) -> AgentOrchestrator:
"""A bare orchestrator with the a2a-dispatch collaborators stubbed."""
orch = object.__new__(AgentOrchestrator)
# _dispatch_a2a_work consults: _fetch_notifications, _resolve_agent_slug,
# _is_agent_active, spawn_agent. _resolve_agent_slug is pure (module
# UUID_TO_SLUG) so it works unstubbed; stub the rest.
orch.spawn_agent = AsyncMock() # type: ignore[method-assign]
orch._is_agent_active = MagicMock(return_value=False) # type: ignore[method-assign]
orch._fetch_notifications = AsyncMock( # type: ignore[method-assign]
return_value=[
{"id": "n1", "to_agents": [ceo_uuid], "body": "board handoff"},
]
)
return orch
@pytest.mark.asyncio
async def test_dispatch_a2a_skips_ceo_target() -> None:
"""A CEO-addressed A2A notification must NOT spawn a CEO container.
The CEO being a notification target (board-review handoff, escalation)
is expected; it is NOT a spawn signal. The notification stays for the
human to read in the panel. This is the live 2026-06-27 regression.
"""
ceo_uuid = AGENT_UUIDS["ceo"]
orch = _a2a_orch(ceo_uuid)
client = MagicMock()
await orch._dispatch_a2a_work(client)
orch.spawn_agent.assert_not_awaited() # type: ignore[attr-defined]
@pytest.mark.asyncio
async def test_dispatch_a2a_skips_intake_and_secretary_targets() -> None:
"""Intake (prompter) and secretary are human-driven chats — never spawned."""
for slug in ("intake-1", "secretary-1"):
orch = _a2a_orch(AGENT_UUIDS[slug])
await orch._dispatch_a2a_work(MagicMock())
orch.spawn_agent.assert_not_awaited() # type: ignore[attr-defined]
@pytest.mark.asyncio
async def test_dispatch_a2a_still_spawns_real_agent_target() -> None:
"""A real (container-eligible) A2A target is still dispatched — the
skip is scoped to human roles only and does not suppress real A2A."""
be_uuid = AGENT_UUIDS["be-dev-1"]
orch = _a2a_orch(be_uuid)
client = MagicMock()
await orch._dispatch_a2a_work(client)
orch.spawn_agent.assert_awaited_once() # type: ignore[attr-defined]
_args, kwargs = orch.spawn_agent.call_args # type: ignore[attr-defined]
assert kwargs.get("agent_id") == "be-dev-1"
@pytest.mark.asyncio
async def test_dispatch_a2a_mixed_targets_skips_only_human() -> None:
"""A notification addressed to both the CEO and a real agent spawns the
real agent once and never the CEO."""
ceo_uuid = AGENT_UUIDS["ceo"]
be_uuid = AGENT_UUIDS["be-dev-1"]
orch = object.__new__(AgentOrchestrator)
orch.spawn_agent = AsyncMock() # type: ignore[method-assign]
orch._is_agent_active = MagicMock(return_value=False) # type: ignore[method-assign]
orch._fetch_notifications = AsyncMock( # type: ignore[method-assign]
return_value=[{"id": "n1", "to_agents": [ceo_uuid, be_uuid]}]
)
client = MagicMock()
await orch._dispatch_a2a_work(client)
spawned = [c.kwargs.get("agent_id") for c in orch.spawn_agent.call_args_list]
assert "ceo" not in spawned
assert spawned == ["be-dev-1"]