feat(gateway): structured required_cells gate — reject i_am_idle on a dropped named cell

The companion to the prompt rule (60de3499): when the brief explicitly names
cells, the Main PM must create a subtask for each and not silently collapse one
into a neighbour. Records the named cells as a 'required_cells:' marker on the
parent's quick_context (no migration — same pattern as the other markers), and
adds a _pm_uncovered_required_cells_guard at i_am_idle that refuses to idle
while a named cell has no subtask. Inert when no parent carries the marker, so
legacy decompositions are never blocked (mirrors the AC-coverage guard).
TaskService.uncovered_required_cells + extract_required_cells + 7 unit tests.

roboco/services/gateway/choreographer/_impl.py

@@ -2893,6 +2893,10 @@ class Choreographer:
             return await self._emit_rejection(
                 guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
             )
+        if guard := await self._pm_uncovered_required_cells_guard(agent_id, briefing):
+            return await self._emit_rejection(
+                guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
+            )
         paused_ids = await self._auto_pause_in_progress_tasks(agent_id)
         await self.task.mark_agent_idle(agent_id)
         if paused_ids:
@@ -3059,6 +3063,46 @@ class Choreographer:
             )
         return None
 
+    async def _pm_uncovered_required_cells_guard(
+        self, agent_id: UUID, briefing: dict[str, Any]
+    ) -> Envelope | None:
+        """Refuse i_am_idle when a PM left an explicitly-named cell undelegated.
+
+        The brief / acceptance criteria / Board handoff can name specific cells
+        (recorded as a ``required_cells:`` marker on the parent). The Main PM
+        must create a subtask for each — its discretion covers only un-named
+        scope; a genuinely-unnecessary named cell must be confirmed via
+        escalate_up/dm, not silently dropped. This is the structured companion
+        to the prompt rule (commit 60de3499), firing at PM exit. Inert when no
+        parent carries the marker, so legacy decompositions are never blocked.
+        """
+        agent = await self.task.agent_for(agent_id)
+        if not agent or agent.role not in ("cell_pm", "main_pm"):
+            return None
+        assigned = await self.task.list_assigned_for_agent(agent_id)
+        for parent in assigned:
+            if str(parent.status) in self._TERMINAL_STATUSES:
+                continue
+            uncovered = await self.task.uncovered_required_cells(parent.id)
+            # isinstance keeps the gate inert under partial test mocks (an
+            # AsyncMock TaskService returns a truthy stub, not a concrete list).
+            if not isinstance(uncovered, list) or not uncovered:
+                continue
+            listing = ", ".join(uncovered)
+            return Envelope.invalid_state(
+                message=(
+                    f"task {parent.id} names cells with no subtask: {listing}; "
+                    "cannot idle while an explicitly-named cell is undelegated."
+                ),
+                remediate=(
+                    f"delegate a subtask for each named cell ({listing}); if one is "
+                    "genuinely unnecessary, confirm via escalate_up/dm rather than "
+                    "dropping it, then retry i_am_idle"
+                ),
+                context_briefing=briefing,
+            )
+        return None
+
     async def _auto_pause_in_progress_tasks(self, agent_id: UUID) -> list[str]:
         """Pause every in_progress task assigned to this agent.
 

roboco/services/task.py

@@ -331,6 +331,40 @@ def extract_original_developer(quick_context: str | None) -> str | None:
     return None
 
 
+_REQUIRED_CELLS_PREFIX = "required_cells:"
+
+
+def _normalize_cell(value: object) -> str:
+    """Normalize a team/cell token for comparison (e.g. backend, frontend, ux_ui)."""
+    raw = str(getattr(value, "value", value)).strip().lower()
+    return raw.replace("/", "_").replace("-", "_").replace(" ", "")
+
+
+def extract_required_cells(quick_context: str | None) -> list[str]:
+    """Cells the brief explicitly named, from a ``required_cells:`` marker line.
+
+    The Main PM must create a subtask for each named cell (it may not silently
+    collapse one into a neighbour — see commit 60de3499). The marker is a single
+    line, e.g. ``required_cells: backend, frontend, ux_ui``. Absent → no
+    constraint (the gate is inert). Returns normalized, de-duplicated cells in
+    marker order.
+    """
+    if not quick_context:
+        return []
+    for raw in quick_context.splitlines():
+        line = raw.strip()
+        if not line.lower().startswith(_REQUIRED_CELLS_PREFIX):
+            continue
+        body = line[len(_REQUIRED_CELLS_PREFIX) :]
+        seen: list[str] = []
+        for tok in body.split(","):
+            cell = _normalize_cell(tok)
+            if cell and cell not in seen:
+                seen.append(cell)
+        return seen
+    return []
+
+
 _SUPERSEDE_MARKER_PREFIX = "external_pr_supersede"
 
 
@@ -5206,6 +5240,23 @@ class TaskService(BaseService):
         )
         return list(result.scalars().all())
 
+    async def uncovered_required_cells(self, parent_task_id: UUID) -> list[str]:
+        """Named cells (parent's ``required_cells:`` marker) with no subtask.
+
+        Inert ([]) when the parent carries no marker — legacy / un-named
+        decompositions are never blocked. Otherwise returns each named cell that
+        has no child subtask on that team, in marker order.
+        """
+        parent = await self.get(parent_task_id)
+        if parent is None:
+            return []
+        required = extract_required_cells(parent.quick_context)
+        if not required:
+            return []
+        children = await self.get_subtasks(parent_task_id)
+        covered = {_normalize_cell(c.team) for c in children if c.team is not None}
+        return [cell for cell in required if cell not in covered]
+
     async def has_earlier_incomplete_code_sibling(self, task: TaskTable) -> bool:
         """True if a lower-sequence, non-terminal, same-assignee code sibling exists.
 

tests/unit/services/test_required_cells.py

@@ -0,0 +1,74 @@
+"""required_cells decomposition gate — marker parse + uncovered-cell coverage.
+
+The Main PM must create a subtask for each cell the brief explicitly names
+(recorded as a ``required_cells:`` marker on the parent's quick_context). The
+gate is inert when no marker is present, so legacy decompositions never block.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.task import TaskService, extract_required_cells
+
+# ---------------------------------------------------------------------------
+# extract_required_cells (marker parser)
+# ---------------------------------------------------------------------------
+
+
+def test_extract_required_cells_absent_is_empty() -> None:
+    assert extract_required_cells(None) == []
+    assert extract_required_cells("original_developer: abc\ndoc_notes: y") == []
+
+
+def test_extract_required_cells_parses_and_normalizes() -> None:
+    qc = "original_developer: abc\nrequired_cells: Backend, Frontend , UX/UI"
+    assert extract_required_cells(qc) == ["backend", "frontend", "ux_ui"]
+
+
+def test_extract_required_cells_dedups_in_order() -> None:
+    out = extract_required_cells("required_cells: backend, backend, frontend")
+    assert out == ["backend", "frontend"]
+
+
+# ---------------------------------------------------------------------------
+# uncovered_required_cells (service coverage check)
+# ---------------------------------------------------------------------------
+
+
+def _service(parent_qc: str | None, child_teams: list[str | None]) -> TaskService:
+    """A TaskService whose get()/get_subtasks() return a parent + these children."""
+    svc = TaskService(MagicMock())
+    parent = MagicMock(quick_context=parent_qc)
+    children = [MagicMock(team=t) for t in child_teams]
+    object.__setattr__(svc, "get", AsyncMock(return_value=parent))
+    object.__setattr__(svc, "get_subtasks", AsyncMock(return_value=children))
+    return svc
+
+
+@pytest.mark.asyncio
+async def test_uncovered_inert_without_marker() -> None:
+    svc = _service("doc_notes: x", ["backend"])
+    assert await svc.uncovered_required_cells(uuid4()) == []
+
+
+@pytest.mark.asyncio
+async def test_uncovered_flags_the_dropped_cell() -> None:
+    # Brief named backend+frontend+ux_ui; only backend+frontend got subtasks.
+    svc = _service("required_cells: backend, frontend, ux_ui", ["backend", "frontend"])
+    assert await svc.uncovered_required_cells(uuid4()) == ["ux_ui"]
+
+
+@pytest.mark.asyncio
+async def test_uncovered_empty_when_all_named_cells_covered() -> None:
+    svc = _service("required_cells: backend, frontend", ["frontend", "backend"])
+    assert await svc.uncovered_required_cells(uuid4()) == []
+
+
+@pytest.mark.asyncio
+async def test_uncovered_normalizes_child_team_form() -> None:
+    # Marker uses underscore, child team uses the slash form — they match.
+    svc = _service("required_cells: ux_ui", ["UX/UI"])
+    assert await svc.uncovered_required_cells(uuid4()) == []