From b19856dcbd6504e65b507df11bdaac0a367231bc Mon Sep 17 00:00:00 2001 From: Renn F Date: Sat, 1 Aug 2026 01:36:39 +0200 Subject: [PATCH] fix(dispatch): route a code task that would land on main-pm to a cell PM instead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The routing layer's never-strand fallbacks — and the strategic classifier's own cross-cell/high-complexity/teamless branches — could hand a code-typed pending task to main-pm, but the dispatcher's pre-claim route refuses that by org invariant (MAIN_PM_NO_CODE: main_pm+code must not coexist; intake coerces such roots to planning), so the task looped claim-reject every ~30s tick forever instead of failing loud. Zero live instances in three days of logs — preventive, found adversarially while fixing the creator-skip hold. _get_routing_target now wraps the pure resolution: a code task whose resolved target is main-pm redirects via _cell_pm_for_stranded_code_task — nearest ancestor cell team wins (parent-chain walk mirroring _resolve_pm_for_review), parentless/unresolvable defaults to backend. Non-code routing is byte-for-byte unchanged; board precedence untouched; the claim-route guard and the lifecycle spec's broader i_will_plan allowance (cell PMs planning code parents, needs_revision recovery) are both left exactly as designed. Gate: full make quality green (suite passed through coverage; xenon/ bandit/pip-audit/deptry/alembic/import-linter/foundation-check exit 0). --- roboco/runtime/orchestrator.py | 90 ++++++++++- .../runtime/test_routing_target_fallback.py | 152 +++++++++++++++--- 2 files changed, 216 insertions(+), 26 deletions(-) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 5d2d1725..7d5aafca 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -11670,9 +11670,15 @@ Start by: "ux_ui": "ux-pm", } - def _get_routing_target(self, routing: str, task: dict[str, Any]) -> str | None: + def _resolve_routing_target(self, routing: str, task: dict[str, Any]) -> str | None: """ - Resolve a routing decision to a specific agent slug. + Resolve a routing decision to a specific agent slug (never-strand + fallbacks default to main-pm — code-task safety lives in the async + wrapper ``_get_routing_target``: main_pm+code must not coexist by + org invariant, so the dispatcher's pre-claim route refuses it + (``_raise_if_main_pm_code_claim``, needs_revision recovery exempt); + the lifecycle spec's broader ``i_will_plan`` claim allowance covers + cell PMs planning code parents, not this dispatch path). Args: routing: One of "board", "main_pm", "cell_pm", "dev", "marketing" @@ -11722,6 +11728,84 @@ Start by: ) return "main-pm" + @staticmethod + def _is_code_task(task: dict[str, Any]) -> bool: + return str(task.get("task_type") or "").lower() == "code" + + async def _nearest_cell_team(self, task: dict[str, Any]) -> str | None: + """Walk the parent chain for the nearest ancestor's cell team. + + Mirrors ``TaskService._resolve_pm_for_review`` (task.py:6493) but + keys on ``team`` rather than ``assigned_to`` — this runs before any + assignment exists. Best-effort: any lookup failure returns None so + the caller falls back to the deterministic default cell. + """ + parent_id = task.get("parent_task_id") + if not parent_id: + return None + from uuid import UUID + + from roboco.db.base import get_session_factory + from roboco.services.task import get_task_service + + cell_teams = frozenset(t.value for t in CELL_TEAMS) + try: + session_factory = get_session_factory() + async with session_factory() as db: + task_svc = get_task_service(db) + while parent_id: + parent = await task_svc.get(UUID(str(parent_id))) + if not parent: + return None + parent_team = getattr(parent.team, "value", parent.team) + if parent_team in cell_teams: + return str(parent_team) + parent_id = parent.parent_task_id + except Exception as exc: + logger.warning( + "cell-team parent walk failed; using default cell", + task_id=task.get("id"), + error=str(exc), + ) + return None + + async def _cell_pm_for_stranded_code_task(self, task: dict[str, Any]) -> str: + """Redirect a code task that would strand on main-pm to a cell PM. + + The dispatcher's pre-claim (``_raise_if_main_pm_code_claim``, the + MAIN_PM_NO_CODE org invariant: main_pm+code must not coexist — + intake coerces such roots to planning) refuses a Main-PM claim of a + pending code task, so a stranded one loops claim-reject forever at + ~30s cadence instead of failing loud. A code task belongs to a + cell: resolve the nearest ancestor cell team first; a parentless/ + unresolvable chain defaults to backend (ponytail: no signal to pick + a better default among the three cells). + """ + team = await self._nearest_cell_team(task) + target = self._TEAM_PM_MAP.get(team or "", "be-pm") + logger.warning( + "code task would strand on main-pm (MAIN_PM_NO_CODE); " + "redirecting to cell pm", + task_id=task.get("id"), + resolved_team=team, + agent_id=target, + ) + return target + + async def _get_routing_target( + self, routing: str, task: dict[str, Any] + ) -> str | None: + """Resolve a routing decision to a specific agent slug. + + Wraps ``_resolve_routing_target``: when that would strand a code + task on main-pm (a claim MAIN_PM_NO_CODE refuses forever), redirects + to a cell PM instead — see ``_cell_pm_for_stranded_code_task``. + """ + target = self._resolve_routing_target(routing, task) + if target == "main-pm" and self._is_code_task(task): + return await self._cell_pm_for_stranded_code_task(task) + return target + def _build_main_pm_triage_prompt( self, task: dict[str, Any], *, bounced_block: str = "" ) -> str: @@ -14069,7 +14153,7 @@ Start now: evidence(task_id="{task_id}") if await self._pending_claim_blocked(task.get("id")): return routing = self._classify_task_routing(task) - agent_id = self._get_routing_target(routing, task) + agent_id = await self._get_routing_target(routing, task) if not agent_id: logger.warning( diff --git a/tests/unit/runtime/test_routing_target_fallback.py b/tests/unit/runtime/test_routing_target_fallback.py index 4a1ce046..026f23f4 100644 --- a/tests/unit/runtime/test_routing_target_fallback.py +++ b/tests/unit/runtime/test_routing_target_fallback.py @@ -5,12 +5,22 @@ None leaves an ownerless pending task dormant, because no dispatcher re-spawns an unrouted task. Tasks that can't be placed on a cell (no team, or a non-cell team like ``fullstack`` / ``system``) and any unrecognized routing fall back to main-pm, which triages them. + +EXCEPT for a code-typed task: main-pm is claim-illegal for ``code`` +(MAIN_PM_NO_CODE, roboco/services/task.py:9506 — a Main PM coordinates, it +never owns code) — routing such a task to main-pm is a guaranteed permanent +claim-reject loop. The fallback instead redirects to a cell PM: the nearest +ancestor's cell team, or a deterministic default (backend / be-pm) with no +usable parent chain. """ from __future__ import annotations from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import uuid4 +import pytest from roboco.runtime.orchestrator import AgentOrchestrator @@ -20,9 +30,34 @@ def _orch() -> AgentOrchestrator: return orch -def _resolve(routing: str, team: str | None) -> str | None: - task: dict[str, Any] = {"id": "t1", "team": team} - return _orch()._get_routing_target(routing, task) +async def _resolve(routing: str, team: str | None, **over: Any) -> str | None: + task: dict[str, Any] = {"id": "t1", "team": team, **over} + return await _orch()._get_routing_target(routing, task) + + +class _CM: + """Async context manager stand-in for ``session_factory()``.""" + + async def __aenter__(self) -> MagicMock: + return MagicMock() + + async def __aexit__(self, *_a: Any) -> bool: + return False + + +def _patch_parent_chain(parents: list[MagicMock]) -> Any: + """Patch the direct-DB parent-chain lookup used by ``_nearest_cell_team``. + + ``parents`` are returned in order from successive ``task_svc.get`` calls, + mirroring one hop per ancestor. + """ + svc = MagicMock() + svc.get = AsyncMock(side_effect=parents) + factory = MagicMock(return_value=_CM()) + return ( + patch("roboco.db.base.get_session_factory", return_value=factory), + patch("roboco.services.task.get_task_service", return_value=svc), + ) # --------------------------------------------------------------------------- @@ -30,49 +65,120 @@ def _resolve(routing: str, team: str | None) -> str | None: # --------------------------------------------------------------------------- -def test_dev_on_cell_team_selects_cell_agent() -> None: - assert _resolve("dev", "backend") == "be-dev-1" +@pytest.mark.asyncio +async def test_dev_on_cell_team_selects_cell_agent() -> None: + assert await _resolve("dev", "backend") == "be-dev-1" -def test_board_routes_to_product_owner() -> None: - assert _resolve("board", None) == "product-owner" +@pytest.mark.asyncio +async def test_board_routes_to_product_owner() -> None: + assert await _resolve("board", None) == "product-owner" -def test_main_pm_routes_to_main_pm() -> None: - assert _resolve("main_pm", None) == "main-pm" +@pytest.mark.asyncio +async def test_main_pm_routes_to_main_pm() -> None: + assert await _resolve("main_pm", None) == "main-pm" -def test_cell_pm_on_team_routes_to_cell_pm() -> None: - assert _resolve("cell_pm", "frontend") == "fe-pm" +@pytest.mark.asyncio +async def test_cell_pm_on_team_routes_to_cell_pm() -> None: + assert await _resolve("cell_pm", "frontend") == "fe-pm" -def test_cell_pm_without_team_falls_back_to_main_pm() -> None: - assert _resolve("cell_pm", None) == "main-pm" +@pytest.mark.asyncio +async def test_cell_pm_without_team_falls_back_to_main_pm() -> None: + assert await _resolve("cell_pm", None) == "main-pm" # --------------------------------------------------------------------------- -# Fallbacks — never None (no dormancy) +# Fallbacks — never None (no dormancy); non-code tasks land on main-pm # --------------------------------------------------------------------------- -def test_dev_without_team_falls_back_to_main_pm() -> None: - assert _resolve("dev", None) == "main-pm" +@pytest.mark.asyncio +async def test_dev_without_team_falls_back_to_main_pm() -> None: + assert await _resolve("dev", None) == "main-pm" -def test_dev_on_non_cell_team_falls_back_to_main_pm() -> None: +@pytest.mark.asyncio +async def test_dev_on_non_cell_team_falls_back_to_main_pm() -> None: # fullstack / system are valid Team values with no cell agent pool. - assert _resolve("dev", "fullstack") == "main-pm" - assert _resolve("dev", "system") == "main-pm" + assert await _resolve("dev", "fullstack") == "main-pm" + assert await _resolve("dev", "system") == "main-pm" -def test_unknown_routing_falls_back_to_main_pm() -> None: - assert _resolve("frobnicate", "backend") == "main-pm" +@pytest.mark.asyncio +async def test_unknown_routing_falls_back_to_main_pm() -> None: + assert await _resolve("frobnicate", "backend") == "main-pm" -def test_no_routing_ever_returns_none() -> None: +@pytest.mark.asyncio +async def test_no_routing_ever_returns_none() -> None: """Every (routing, team) combination resolves to some agent — never None.""" routings = ["board", "main_pm", "marketing", "cell_pm", "dev", "bogus"] teams: list[str | None] = [None, "backend", "fullstack", "system", "marketing"] for routing in routings: for team in teams: - assert _resolve(routing, team) is not None, (routing, team) + assert await _resolve(routing, team) is not None, (routing, team) + + +# --------------------------------------------------------------------------- +# Code tasks: main-pm is claim-illegal — redirect to a cell PM instead. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_code_task_no_team_resolves_parent_cell_pm() -> None: + """team=None, but the parent task names team=backend -> be-pm, not main-pm.""" + parent = MagicMock(team="backend", parent_task_id=None) + patches = _patch_parent_chain([parent]) + with patches[0], patches[1]: + result = await _resolve( + "dev", None, task_type="code", parent_task_id=str(uuid4()) + ) + assert result == "be-pm" + + +@pytest.mark.asyncio +async def test_code_task_fullstack_team_no_parent_defaults_to_backend() -> None: + """team="fullstack" (non-cell), no parent -> deterministic default (be-pm).""" + result = await _resolve("dev", "fullstack", task_type="code") + assert result == "be-pm" + + +@pytest.mark.asyncio +async def test_non_code_task_missing_team_still_routes_to_main_pm() -> None: + """The redirect is code-only — a planning/admin task may legally sit on main-pm.""" + result = await _resolve("dev", None, task_type="planning") + assert result == "main-pm" + + +@pytest.mark.asyncio +async def test_code_task_with_proper_cell_team_unchanged() -> None: + """A code task that already resolves to a real cell agent is untouched.""" + assert await _resolve("dev", "backend", task_type="code") == "be-dev-1" + + +@pytest.mark.asyncio +async def test_code_task_walks_multiple_parent_hops() -> None: + """Grandparent names the cell team; the immediate parent has none.""" + grandparent = MagicMock(team="frontend", parent_task_id=None) + parent = MagicMock(team=None, parent_task_id=uuid4()) + patches = _patch_parent_chain([parent, grandparent]) + with patches[0], patches[1]: + result = await _resolve( + "dev", None, task_type="code", parent_task_id=str(uuid4()) + ) + assert result == "fe-pm" + + +@pytest.mark.asyncio +async def test_code_task_parent_lookup_failure_defaults_to_backend() -> None: + """A DB error walking the parent chain must not strand the task either.""" + with patch( + "roboco.db.base.get_session_factory", side_effect=RuntimeError("db down") + ): + result = await _resolve( + "dev", None, task_type="code", parent_task_id=str(uuid4()) + ) + assert result == "be-pm"