From eec35c03573ecbaa235d98e167e2dd725ef90031 Mon Sep 17 00:00:00 2001 From: Renn F Date: Wed, 17 Jun 2026 16:20:23 +0200 Subject: [PATCH] fix(orchestrator): un-deadlock a CEO-rejected coordination root MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A coordination root (team=main_pm, product-linked, no repo) the CEO sends back lands in needs_revision, but the dev dispatcher skips it (not a cell team) and the closure path only handles paused parents — so it sat in needs_revision forever. (NOT a foundation-spec gap: the spec already allows needs_revision -> claimed for any role.) - _dispatch_revision_coordination_roots: re-spawn the owning PM for a needs_revision coordination root so it re-coordinates the revision (registered in the dispatch loop after PM closure) - _readiness_check_role_for_status: widen the dev-owned states (needs_revision, verifying) to also accept cell_pm/main_pm for coordination roots — a pure widening; normal code tasks stay dev/doc-only - 16 unit tests (dispatcher decision + readiness widening) --- roboco/runtime/orchestrator.py | 48 +++++++++- .../test_readiness_role_status_match.py | 57 ++++++++++++ .../test_revision_coordination_dispatch.py | 87 +++++++++++++++++++ 3 files changed, 189 insertions(+), 3 deletions(-) create mode 100644 tests/unit/runtime/test_revision_coordination_dispatch.py diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index b7a834b3..51a58fdc 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -2330,7 +2330,7 @@ class AgentOrchestrator: @staticmethod def _readiness_check_role_for_status( - agent_id: str, role: str, status: str + agent_id: str, role: str, status: str, *, is_coordination: bool = False ) -> str | None: """Verify agent role matches the role expected for the task status. @@ -2339,7 +2339,10 @@ class AgentOrchestrator: developer/documenter to defang the bug where QA got respawned on a `needs_revision` task via the crash-restart path and immediately hit ``role 'qa' may not claim from status - 'needs_revision'`` at the gateway. + 'needs_revision'`` at the gateway. A coordination root (no code; product + fan-out owned by a PM) is the exception: it has no dev, so a CEO-rejected + one returns to its PM — the dev-owned states also accept the PM roles for + it (a pure widening; nothing currently allowed is blocked). """ role_mismatch: dict[str, str | set[str]] = { "awaiting_qa": "qa", @@ -2354,6 +2357,8 @@ class AgentOrchestrator: required = role_mismatch.get(status) if required is None: return None + if is_coordination and status in ("needs_revision", "verifying"): + required = set(required) | {"cell_pm", "main_pm"} ok = role in required if isinstance(required, set) else role == required if ok: return None @@ -2379,7 +2384,9 @@ class AgentOrchestrator: # the readiness and stuck-detection paths agree. if _branch_is_expected(task) and not task.get("branch_name"): return f"state={status} but branch_name is unset" - return self._readiness_check_role_for_status(agent_id, role, status) + return self._readiness_check_role_for_status( + agent_id, role, status, is_coordination=_is_coordination_task(task) + ) @staticmethod async def _readiness_check_git_token(project_slug: str | None) -> str | None: @@ -6217,6 +6224,10 @@ Start now: evidence(task_id="{task_id}") dispatchers = [ ("pm_work", self._dispatch_pm_work(client)), ("pm_closure_work", self._dispatch_pm_closure_work(client)), + ( + "revision_coordination", + self._dispatch_revision_coordination_roots(client), + ), ("dev_work", self._dispatch_dev_work(client)), ("qa_work", self._dispatch_qa_work(client)), ("pr_review_work", self._dispatch_pr_review_work(client)), @@ -6723,6 +6734,37 @@ Start now: evidence(task_id="{task_id}") await self._route_unassigned_pm_task(client, task) + async def _dispatch_revision_coordination_roots( + self, client: httpx.AsyncClient + ) -> None: + """Re-spawn the owning PM for a CEO-rejected coordination root. + + A coordination root (team=main_pm, product-linked, no repo) the CEO sends + back lands in ``needs_revision``. The dev dispatcher skips it (not a cell + team) and the closure path only handles paused parents, so without this + it would sit in needs_revision forever — the deadlock. Respawn its PM so + it re-coordinates the revision. Cell/code needs_revision tasks are left + to the dev dispatcher; this handles only coordination roots. + """ + tasks = await self._fetch_tasks(client, "needs_revision") + for task in tasks: + if self._is_task_handled_this_tick(task.get("id")): + continue + if not _is_coordination_task(task): + continue + owner = task.get("assigned_to") or task.get("claimed_by") + agent_slug = self._resolve_agent_slug(owner) if owner else None + if not agent_slug or self._is_agent_active(agent_slug): + continue + if get_agent_role(agent_slug) not in ("cell_pm", "main_pm"): + continue + await self.spawn_agent( + agent_id=agent_slug, + task_id=task["id"], + initial_prompt=self._get_prompt_for_agent(agent_slug, task), + git_context=self._task_git_context(task), + ) + @staticmethod def _all_descendants_terminal(descendants: list[dict[str, Any]]) -> bool: """Every descendant in a closure-complete state?""" diff --git a/tests/unit/runtime/test_readiness_role_status_match.py b/tests/unit/runtime/test_readiness_role_status_match.py index 4892d905..33e07ec8 100644 --- a/tests/unit/runtime/test_readiness_role_status_match.py +++ b/tests/unit/runtime/test_readiness_role_status_match.py @@ -73,3 +73,60 @@ def test_unmapped_status_allows_any_role() -> None: ) is None ), f"role={role} status={status} should not be rejected by this gate" + + +# --------------------------------------------------------------------------- +# Coordination roots — a CEO-rejected coordination root returns to its PM (#5). +# --------------------------------------------------------------------------- + + +def test_pm_on_needs_revision_coordination_allowed() -> None: + """A coordination root in needs_revision belongs to its PM, not a dev.""" + for role in ("main_pm", "cell_pm"): + assert ( + AgentOrchestrator._readiness_check_role_for_status( + agent_id="main-pm", + role=role, + status="needs_revision", + is_coordination=True, + ) + is None + ) + + +def test_pm_on_verifying_coordination_allowed() -> None: + assert ( + AgentOrchestrator._readiness_check_role_for_status( + agent_id="main-pm", role="main_pm", status="verifying", is_coordination=True + ) + is None + ) + + +def test_dev_on_needs_revision_coordination_still_allowed() -> None: + """Widening is additive — developer is still accepted.""" + assert ( + AgentOrchestrator._readiness_check_role_for_status( + agent_id="be-dev-1", + role="developer", + status="needs_revision", + is_coordination=True, + ) + is None + ) + + +def test_qa_on_needs_revision_coordination_still_blocked() -> None: + """The widening only adds the PM roles — QA is still a misroute.""" + reason = AgentOrchestrator._readiness_check_role_for_status( + agent_id="be-qa", role="qa", status="needs_revision", is_coordination=True + ) + assert reason is not None + + +def test_pm_on_needs_revision_noncoordination_still_blocked() -> None: + """A normal (code) needs_revision task is still dev/doc-only for a PM.""" + reason = AgentOrchestrator._readiness_check_role_for_status( + agent_id="be-pm", role="cell_pm", status="needs_revision", is_coordination=False + ) + assert reason is not None diff --git a/tests/unit/runtime/test_revision_coordination_dispatch.py b/tests/unit/runtime/test_revision_coordination_dispatch.py new file mode 100644 index 00000000..e8918405 --- /dev/null +++ b/tests/unit/runtime/test_revision_coordination_dispatch.py @@ -0,0 +1,87 @@ +"""_dispatch_revision_coordination_roots — un-deadlock a CEO-rejected root (#5). + +A coordination root (team=main_pm, product-linked, no repo) the CEO sends back +lands in needs_revision. The dev dispatcher skips it (not a cell team) and the +closure path only handles paused parents, so without this dispatcher it sits +forever. This re-spawns its owning PM so it re-coordinates the revision. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +import roboco.runtime.orchestrator as orch_mod +from roboco.runtime.orchestrator import AgentOrchestrator + + +def _orch( + tasks: list[dict[str, Any]], *, slug: str, active: bool +) -> tuple[AgentOrchestrator, AsyncMock]: + """A bare orchestrator with its dispatch helpers mocked; returns (orch, spawn).""" + orch = object.__new__(AgentOrchestrator) + spawn = AsyncMock() + object.__setattr__(orch, "_fetch_tasks", AsyncMock(return_value=tasks)) + object.__setattr__( + orch, "_is_task_handled_this_tick", MagicMock(return_value=False) + ) + object.__setattr__(orch, "_resolve_agent_slug", MagicMock(return_value=slug)) + object.__setattr__(orch, "_is_agent_active", MagicMock(return_value=active)) + object.__setattr__(orch, "_get_prompt_for_agent", MagicMock(return_value="p")) + object.__setattr__(orch, "_task_git_context", MagicMock(return_value=None)) + object.__setattr__(orch, "spawn_agent", spawn) + return orch, spawn + + +def _task() -> dict[str, Any]: + return { + "id": "t1", + "status": "needs_revision", + "assigned_to": "u1", + "team": "main_pm", + } + + +@pytest.mark.asyncio +async def test_respawns_pm_for_rejected_coordination_root( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(orch_mod, "_is_coordination_task", lambda _t: True) + orch, spawn = _orch([_task()], slug="main-pm", active=False) + await orch._dispatch_revision_coordination_roots(MagicMock()) + spawn.assert_awaited_once() + call = spawn.await_args + assert call is not None + assert call.kwargs["agent_id"] == "main-pm" + assert call.kwargs["task_id"] == "t1" + + +@pytest.mark.asyncio +async def test_skips_non_coordination_needs_revision( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # A normal (code) needs_revision task → left to the dev dispatcher. + monkeypatch.setattr(orch_mod, "_is_coordination_task", lambda _t: False) + orch, spawn = _orch([_task()], slug="be-dev-1", active=False) + await orch._dispatch_revision_coordination_roots(MagicMock()) + spawn.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_skips_when_pm_already_active( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(orch_mod, "_is_coordination_task", lambda _t: True) + orch, spawn = _orch([_task()], slug="main-pm", active=True) + await orch._dispatch_revision_coordination_roots(MagicMock()) + spawn.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_skips_non_pm_owner(monkeypatch: pytest.MonkeyPatch) -> None: + # A coordination root owned by a non-PM role → role guard skips it. + monkeypatch.setattr(orch_mod, "_is_coordination_task", lambda _t: True) + orch, spawn = _orch([_task()], slug="be-dev-1", active=False) + await orch._dispatch_revision_coordination_roots(MagicMock()) + spawn.assert_not_awaited()