fix(orchestrator): dispatch board agents for assigned board-team tasks

No dispatcher ever spawned board roles (product-owner / head-marketing) —
_handle_pm_assigned_task gates on _PM_AGENTS and there was no board path —
so a task assigned to the Product Owner sat pending forever (surfaced by the
first board-led run). Board roles advise: triage / note / say / escalate_to_ceo
/ i_am_idle, with NO verb to claim, plan, delegate, or complete. So a respawn
cannot advance the task and would just loop.

Add _handle_board_assigned_task: spawn the assigned board agent exactly ONCE
(tracked in _board_dispatched) with a review prompt that steers it to its real
verbs (record requirements via note, discuss via say, then i_am_idle). The
board review is recorded; the CEO then reassigns the task to Main PM for
delegation (the handoff stays CEO-mediated, by design — board roles cannot
delegate). _dispatch_pm_work routes board-assigned tasks here.
This commit is contained in:
Renn F
2026-05-25 02:48:13 +02:00
parent bc5e016d6d
commit c1d0eefd20
2 changed files with 176 additions and 1 deletions
+86 -1
View File
@@ -499,6 +499,11 @@ class AgentOrchestrator:
# is in a loop — without this gate the orchestrator re-spawns every # is in a loop — without this gate the orchestrator re-spawns every
# tick forever (seen in production on 2026-04-22). # tick forever (seen in production on 2026-04-22).
self._pm_respawn_tracker: dict[tuple[str, str], dict[str, Any]] = {} self._pm_respawn_tracker: dict[tuple[str, str], dict[str, Any]] = {}
# Board agents (Product Owner / Head of Marketing) get exactly ONE
# review pass per assigned task: they have no verb to claim, plan,
# delegate, or complete, so a respawn cannot advance the task and would
# just loop. Tracks (agent_slug, task_id) already dispatched.
self._board_dispatched: set[tuple[str, str]] = set()
# Stale-claim reaper config. Wave C3 (2026-05-12): sourced from # Stale-claim reaper config. Wave C3 (2026-05-12): sourced from
# stale_claim_reap_seconds (default 600) rather than # stale_claim_reap_seconds (default 600) rather than
# claim_stale_seconds (default 180). The two settings are now # claim_stale_seconds (default 180). The two settings are now
@@ -4027,6 +4032,15 @@ Start now: evidence(task_id="{task_id}")
} }
) )
# Board reviewers. They advise — review + record requirements + escalate —
# but do not build or delegate. Dispatched once per assigned board task.
_BOARD_AGENTS: ClassVar[frozenset[str]] = frozenset(
{
"product-owner",
"head-marketing",
}
)
# Use foundation's default; keep the local name for back-compat. # Use foundation's default; keep the local name for back-compat.
_PM_RESPAWN_MAX_UNPRODUCTIVE = _AGENT_LOOP_BUDGET.pm_respawn_max_unproductive _PM_RESPAWN_MAX_UNPRODUCTIVE = _AGENT_LOOP_BUDGET.pm_respawn_max_unproductive
@@ -4165,6 +4179,38 @@ Start now: evidence(task_id="{task_id}")
git_context=self._task_git_context(task), git_context=self._task_git_context(task),
) )
async def _handle_board_assigned_task(
self, task: dict[str, Any], assigned_to: str
) -> None:
"""Spawn a board agent (Product Owner / Head of Marketing) ONCE to
review an assigned board task.
Board roles advise: they can triage, record notes, discuss, and
escalate_to_ceo, but have NO verb to claim, plan, delegate, or complete.
So a respawn cannot advance the task — it would just loop. The board
reviews and records requirements; the CEO then reassigns the task to
Main PM for delegation to the cells. Dispatch is therefore one-shot per
(agent, task).
"""
agent_slug = self._resolve_agent_slug(assigned_to)
if agent_slug not in self._BOARD_AGENTS or self._is_agent_active(agent_slug):
return
key = (agent_slug, str(task.get("id")))
if key in self._board_dispatched:
return
self._board_dispatched.add(key)
logger.info(
"Spawning board agent for review",
task_id=task.get("id"),
agent_id=agent_slug,
)
await self.spawn_agent(
agent_id=agent_slug,
task_id=task["id"],
initial_prompt=self._build_board_prompt(task),
git_context=self._task_git_context(task),
)
def _pm_spawn_prompt( def _pm_spawn_prompt(
self, routing: str, agent_id: str, task: dict[str, Any] self, routing: str, agent_id: str, task: dict[str, Any]
) -> str: ) -> str:
@@ -4247,7 +4293,10 @@ Start now: evidence(task_id="{task_id}")
continue continue
assigned_to = task.get("assigned_to") assigned_to = task.get("assigned_to")
if assigned_to: if assigned_to:
await self._handle_pm_assigned_task(task, assigned_to) if self._resolve_agent_slug(assigned_to) in self._BOARD_AGENTS:
await self._handle_board_assigned_task(task, assigned_to)
else:
await self._handle_pm_assigned_task(task, assigned_to)
continue continue
await self._route_unassigned_pm_task(client, task) await self._route_unassigned_pm_task(client, task)
@@ -5566,6 +5615,42 @@ TEAM: {team}
5. give_me_work() / triage() for the next item, or i_am_idle(). 5. give_me_work() / triage() for the next item, or i_am_idle().
Never `commit`, never write code, never run `git`. PMs coordinate. Never `commit`, never write code, never run `git`. PMs coordinate.
"""
def _build_board_prompt(self, task: dict[str, Any]) -> str:
"""Prompt for a board agent (Product Owner / Head of Marketing) to
review and SHAPE a strategic task. Board roles advise — they do not
build, code, or delegate."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
description = task.get("description", "No description")
return f"""\
You are on the Board. This strategic task is assigned to YOU for review.
TASK: {task_id}
TITLE: {title}
DESCRIPTION: {description}
YOUR ROLE: review and shape this work. You do NOT build, code, claim, or
delegate — those verbs are not yours. Your deliverable is a recorded review.
== WHAT TO DO ==
1. triage()
— see your board-level work and context.
2. note(text="<the product requirements and acceptance criteria you expect, the
scope, the must-haves, and what 'done' looks like>",
scope='decision', task_id="{task_id}")
— this recorded review is how the CEO and Main PM act on your input.
3. say(...) in your board channel to flag UX, positioning, or risk concerns
(Head of Marketing: weigh in on UX + how the feature is positioned).
4. i_am_idle()
— when your review is recorded. The CEO routes the task to Main PM for
delegation to the cells; you do NOT hand it off yourself.
Do NOT attempt to claim, plan, complete, or delegate — the gateway will reject
those, and a substantive recorded note IS your job here.
""" """
def _build_marketing_prompt(self, task: dict[str, Any]) -> str: def _build_marketing_prompt(self, task: dict[str, Any]) -> str:
+90
View File
@@ -0,0 +1,90 @@
"""Board agents (Product Owner / Head of Marketing) must be dispatched for
assigned board-team tasks and only ONCE.
Before this, no dispatcher spawned board roles (only PMs, devs, QA, doc, and
marketing were wired), so a task assigned to the Product Owner sat `pending`
forever. Board roles also have no verb to claim/plan/delegate/complete, so a
respawn cannot advance the task dispatch is one-shot per (agent, task); the
CEO reassigns to Main PM after the board review is recorded.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock, patch
from uuid import uuid4
import pytest
from roboco.runtime.orchestrator import AgentOrchestrator
def _make_orch() -> AgentOrchestrator:
orch = AgentOrchestrator.__new__(AgentOrchestrator)
orch._instances = {}
orch._board_dispatched = set()
return orch
def _board_task(assigned_to: str) -> dict[str, Any]:
return {
"id": str(uuid4()),
"status": "pending",
"team": "board",
"title": "Strategic feature",
"description": "A board-level task to review and shape.",
"assigned_to": assigned_to,
}
@pytest.mark.asyncio
async def test_board_agent_spawned_once_for_assigned_board_task() -> None:
orch = _make_orch()
task = _board_task("product-owner")
with (
patch.object(orch, "_is_agent_active", return_value=False),
patch.object(orch, "_task_git_context", return_value=None),
patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn,
):
await orch._handle_board_assigned_task(task, "product-owner")
# Second tick: task is still pending (board has no progression verb) —
# must NOT respawn (no loop).
await orch._handle_board_assigned_task(task, "product-owner")
spawn.assert_awaited_once()
assert spawn.await_args.kwargs["agent_id"] == "product-owner"
assert spawn.await_args.kwargs["task_id"] == task["id"]
@pytest.mark.asyncio
async def test_board_handler_skips_when_agent_active() -> None:
orch = _make_orch()
task = _board_task("head-marketing")
with (
patch.object(orch, "_is_agent_active", return_value=True),
patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn,
):
await orch._handle_board_assigned_task(task, "head-marketing")
spawn.assert_not_awaited()
@pytest.mark.asyncio
async def test_board_handler_ignores_non_board_assignee() -> None:
orch = _make_orch()
task = _board_task("be-pm")
with (
patch.object(orch, "_is_agent_active", return_value=False),
patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn,
):
await orch._handle_board_assigned_task(task, "be-pm")
spawn.assert_not_awaited()
def test_board_review_prompt_uses_board_verbs_only() -> None:
"""The prompt must steer board agents to their real verbs (triage / note /
say / i_am_idle) and away from claim/plan/delegate they do not have."""
orch = _make_orch()
prompt = orch._build_board_prompt(_board_task("product-owner"))
assert "triage()" in prompt
assert "note(" in prompt
assert "i_am_idle()" in prompt
assert "do NOT" in prompt.lower() or "do not" in prompt.lower()