fix(dispatch): route a code task that would land on main-pm to a cell PM instead

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

@@ -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(

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:
+async def _resolve(routing: str, team: str | None, **over: Any) -> str | None:
-    task: dict[str, Any] = {"id": "t1", "team": team}
+    task: dict[str, Any] = {"id": "t1", "team": team, **over}
-    return _orch()._get_routing_target(routing, task)
+    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:
+@pytest.mark.asyncio
-    assert _resolve("dev", "backend") == "be-dev-1"
+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:
+@pytest.mark.asyncio
-    assert _resolve("board", None) == "product-owner"
+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:
+@pytest.mark.asyncio
-    assert _resolve("main_pm", None) == "main-pm"
+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:
+@pytest.mark.asyncio
-    assert _resolve("cell_pm", "frontend") == "fe-pm"
+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:
+@pytest.mark.asyncio
-    assert _resolve("cell_pm", None) == "main-pm"
+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:
+@pytest.mark.asyncio
-    assert _resolve("dev", None) == "main-pm"
+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 await _resolve("dev", "fullstack") == "main-pm"
-    assert _resolve("dev", "system") == "main-pm"
+    assert await _resolve("dev", "system") == "main-pm"
 
 
-def test_unknown_routing_falls_back_to_main_pm() -> None:
+@pytest.mark.asyncio
-    assert _resolve("frobnicate", "backend") == "main-pm"
+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"