feat(gateway): roll-up AC-verification gate (guardrails spec 4/4)

A parent could complete / submit_up / escalate_to_ceo once its subtasks were
merely terminal — never checking whether the parent's acceptance criteria were
actually satisfied. That's how PR #175's half-built umbrella sailed to CEO
approval (escalate_to_ceo had no subtask/AC check at all).

- TaskService.uncovered_parent_acceptance_criteria(parent): parent ACs not
  covered by a COMPLETED child (via parent_ac_refs). Safe-by-construction —
  returns [] unless a child declares coverage, so it is INERT for tasks
  decomposed before coverage tracking and activates only once a PM maps
  children to parent criteria. Cancelled children do not count.
- _parent_acs_covered_envelope wired into all four roll-up gates: cell_pm_complete,
  main_pm_complete, submit_up, and escalate_to_ceo (the weakest — previously
  only journal:decision). isinstance guard keeps it inert under partial mocks.
- 4 new tests; 57 task + 89 gateway tests green.

Pairs with spec 2 (coverage at decompose-time forces the linkage this enforces).

roboco/services/gateway/choreographer/_impl.py

@@ -811,6 +811,41 @@ class Choreographer:
             context_briefing=await self._briefing_for(agent_id, task_id),
         )
 
+    async def _parent_acs_covered_envelope(
+        self,
+        agent_id: UUID,
+        task_id: UUID,
+        *,
+        context_phrase: str,
+    ) -> Envelope | None:
+        """Reject roll-up if any parent acceptance criterion is unsatisfied.
+
+        A parent AC is satisfied only when a COMPLETED child declared it via
+        ``covers_parent_criteria`` (→ ``parent_ac_refs``). Safe-by-construction:
+        ``uncovered_parent_acceptance_criteria`` returns nothing unless coverage
+        is declared on the children, so this is inert for tasks decomposed before
+        coverage tracking and bites only once a PM maps children to parent
+        criteria. The backstop that stops a half-built parent (criteria silently
+        dropped at decompose-time) from rolling up "done" — the PR #175 hole.
+        """
+        uncovered = await self.task.uncovered_parent_acceptance_criteria(task_id)
+        # isinstance keeps the gate inert under partial test mocks (an AsyncMock
+        # TaskService returns a truthy stub, not a concrete list) and on any
+        # unexpected return — enforce only on a real, non-empty list.
+        if not isinstance(uncovered, list) or not uncovered:
+            return None
+        listing = "; ".join(uncovered)
+        return Envelope.tracing_gap(
+            missing=["parent acceptance criteria not satisfied"],
+            remediate=(
+                f"{len(uncovered)} parent acceptance criteria are not covered by a "
+                f"completed subtask before {context_phrase}: {listing}. Delegate "
+                "(or reassign) subtasks covering them and let those pass QA + "
+                "complete first."
+            ),
+            context_briefing=await self._briefing_for(agent_id, task_id),
+        )
+
     def _verb_runner(self) -> VerbRunner:
         """Construct a VerbRunner bound to this Choreographer's services.
 
@@ -4342,6 +4377,10 @@ class Choreographer:
             pm_agent_id, task_id, context_phrase="bubbling up"
         ):
             return env
+        if env := await self._parent_acs_covered_envelope(
+            pm_agent_id, task_id, context_phrase="bubbling up"
+        ):
+            return env
         if not t.branch_name:
             return Envelope.invalid_state(
                 message="task has no branch; cannot open cell-level PR",
@@ -4604,8 +4643,13 @@ class Choreographer:
             )
         if env := await self._check_complete_gates(pm_agent_id, task_id, notes):
             return env
-        if env := await self._subtasks_not_terminal_envelope(
+        if env := (
+            await self._subtasks_not_terminal_envelope(
                 pm_agent_id, task_id, context_phrase="completing parent"
+            )
+            or await self._parent_acs_covered_envelope(
+                pm_agent_id, task_id, context_phrase="completing parent"
+            )
         ):
             return env
         if t.pr_number is None:
@@ -4885,8 +4929,13 @@ class Choreographer:
             main_pm_agent_id, root_task_id, notes
         ):
             return env
-        if env := await self._subtasks_not_terminal_envelope(
+        if env := (
+            await self._subtasks_not_terminal_envelope(
                 main_pm_agent_id, root_task_id, context_phrase="escalating to CEO"
+            )
+            or await self._parent_acs_covered_envelope(
+                main_pm_agent_id, root_task_id, context_phrase="escalating to CEO"
+            )
         ):
             return env
         return None
@@ -5261,9 +5310,15 @@ class Choreographer:
         # Verb-specific preflight: journal:decision presence (out of spec scope).
         # Delegates to _check_pm_decision_required which consumes
         # VERB_REQUIREMENTS["escalate_to_ceo"].
-        if env := await self._check_pm_decision_required(
+        # Verb-body preflight: journal:decision presence, then the parent-AC
+        # backstop (a root may not escalate to the CEO with parent ACs that no
+        # completed subtask covered — inert until coverage is declared).
+        env = await self._check_pm_decision_required(
             "escalate_to_ceo", agent_id, task_id, t
-        ):
+        ) or await self._parent_acs_covered_envelope(
+            agent_id, task_id, context_phrase="escalating to CEO"
+        )
+        if env:
             return await self._emit_rejection(
                 env.with_introspection(task=t, role=role_str),
                 agent_id=agent_id,

roboco/services/task.py

@@ -5671,6 +5671,41 @@ class TaskService(BaseService):
         statuses = result.scalars().all()
         return all(s in terminal for s in statuses)
 
+    async def uncovered_parent_acceptance_criteria(self, task_id: UUID) -> list[str]:
+        """Parent ACs not yet satisfied by a COMPLETED child — for the roll-up gate.
+
+        Safe-by-construction: returns ``[]`` (no enforcement) unless at least one
+        child declares ``parent_ac_refs``, so the gate is inert for tasks
+        decomposed before coverage tracking and activates automatically once a PM
+        declares which child covers which parent criterion. A criterion counts as
+        covered only when a child whose ``parent_ac_refs`` includes it has
+        COMPLETED (cancelled children do not count — their work did not pass QA).
+        Returns the uncovered criterion *texts* for a human-readable rejection.
+        """
+        parent = await self.get(task_id)
+        if not parent or not parent.acceptance_criteria:
+            return []
+        result = await self.session.execute(
+            select(TaskTable.status, TaskTable.parent_ac_refs).where(
+                TaskTable.parent_task_id == task_id
+            )
+        )
+        rows = list(result.all())
+        if not any((refs or []) for _status, refs in rows):
+            # Decomposition predates coverage tracking — do not enforce.
+            return []
+        covered: set[str] = set()
+        for status, refs in rows:
+            if status == TaskStatus.COMPLETED:
+                covered.update(refs or [])
+        ids = parent.acceptance_criteria_ids or []
+        texts = parent.acceptance_criteria or []
+        return [
+            texts[idx] if idx < len(texts) else ac_id
+            for idx, ac_id in enumerate(ids)
+            if ac_id not in covered
+        ]
+
     async def set_plan(
         self, task_id: UUID, plan: str | dict[str, Any]
     ) -> TaskTable | None:

tests/unit/services/test_task.py

@@ -621,6 +621,57 @@ async def test_create_generates_ac_ids_and_carries_parent_ac_refs() -> None:
     assert list(task.parent_ac_refs) == ["parent-ac-1", "parent-ac-2"]
 
 
+def _svc_with_children(parent: object, child_rows: list[tuple]) -> TaskService:
+    """TaskService whose get() returns `parent` and whose execute() yields the
+    (status, parent_ac_refs) child rows the coverage primitive selects."""
+    rows = MagicMock()
+    rows.all.return_value = child_rows
+    svc = TaskService(MagicMock(execute=AsyncMock(return_value=rows)))
+    _bind(svc, "get", AsyncMock(return_value=parent))
+    return svc
+
+
+@pytest.mark.asyncio
+async def test_uncovered_parent_acs_inert_without_declared_coverage() -> None:
+    # No child declares parent_ac_refs -> coverage tracking inactive -> the gate
+    # is inert (legacy/in-flight tasks are never blocked).
+    parent = _build_task(
+        acceptance_criteria=["a", "b"], acceptance_criteria_ids=["id-a", "id-b"]
+    )
+    svc = _svc_with_children(
+        parent, [(TaskStatus.COMPLETED, []), (TaskStatus.COMPLETED, [])]
+    )
+    assert await svc.uncovered_parent_acceptance_criteria(parent.id) == []
+
+
+@pytest.mark.asyncio
+async def test_uncovered_parent_acs_flags_unsatisfied_and_ignores_cancelled() -> None:
+    parent = _build_task(
+        acceptance_criteria=["crit a", "crit b", "crit c"],
+        acceptance_criteria_ids=["id-a", "id-b", "id-c"],
+    )
+    svc = _svc_with_children(
+        parent,
+        [
+            (TaskStatus.COMPLETED, ["id-a"]),  # covers crit a
+            (TaskStatus.CANCELLED, ["id-b"]),  # cancelled -> does NOT cover crit b
+        ],
+    )
+    assert await svc.uncovered_parent_acceptance_criteria(parent.id) == [
+        "crit b",
+        "crit c",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_uncovered_parent_acs_empty_when_all_covered() -> None:
+    parent = _build_task(
+        acceptance_criteria=["a", "b"], acceptance_criteria_ids=["id-a", "id-b"]
+    )
+    svc = _svc_with_children(parent, [(TaskStatus.COMPLETED, ["id-a", "id-b"])])
+    assert await svc.uncovered_parent_acceptance_criteria(parent.id) == []
+
+
 @pytest.mark.asyncio
 async def test_unblock_with_branch_resumes_in_progress() -> None:
     # A task claimed (has a branch) before it blocked resumes in_progress.