fix(task): drop cell_pm→main_pm auto-escalation in complete (#178)

`service.complete()` ran a two-tier approval chain
(`_apply_complete_approval_chain`): when a Cell PM completed an
`awaiting_pm_review` task, `_handle_cell_pm_escalation` silently
reassigned `task.assigned_to = main_pm.id` (+ `claimed_by`) and kept
the task in `awaiting_pm_review` for a second-tier review by Main PM.

That model is incompatible with the gateway's `main_pm_complete`,
which explicitly rejects any non-root task
(`if t.parent_task_id is not None: return invalid_state("main_pm
complete only operates on root tasks")` —
choreographer/_impl.py:3860). Result: the leaf got handed to main-pm
with no verb that could advance it → permanent wedge.

Observed end-to-end this session (smoke run 02:25–02:35):
- 02:25:56 leaf → awaiting_pm_review (correctly assigned to be-pm via
  notify_pm_of_docs_complete).
- 02:26:42 be-pm cell_pm_complete REJECTED tracing_gap journal:reflect
  (proves leaf IS assigned to be-pm).
- ~02:27 (silent — success-path is INFO, filtered): be-pm wrote the
  reflect note + retried → choreographer cell_pm_complete → git.pr_merge
  → service.cell_pm_complete → service.complete(agent=be-pm) →
  _apply_complete_approval_chain → _handle_cell_pm_escalation →
  task.assigned_to = main_pm.id (no `task.reassigned` audit because the
  event goes to `_emit_task_event(EventType.TASK_ESCALATED_TO_MAIN_PM)`,
  not the gateway audit log).
- 02:27:30 _dispatch_pm_review_work (orchestrator) saw leaf with
  assigned_to=main-pm → spawned main-pm against the leaf (target_id
  in audit_log confirms it: `target_id=f3bdd585 agent_slug=main-pm`).
- 02:28:21+ be-pm cell_pm_complete → not_authorized "not assigned to
  you". main-pm main_pm_complete → invalid_state "only operates on
  root tasks". Closure dispatcher cycled both PMs to budget-reap.

Fix: remove the cell_pm branch from `_apply_complete_approval_chain`.
Cell PM completing a non-root awaiting_pm_review task now transitions
it to COMPLETED (the gateway model). Cell→main escalation, when
intended, uses the dedicated `submit_up` verb on the cell-level parent,
not `complete`. The main_pm → CEO branch for root parents stays.
Removed the now-dead `_handle_cell_pm_escalation` helper and the now-
unused `agent_id` parameter on `_apply_complete_approval_chain`.

Tests inverted: `test_complete_cell_pm_escalates_to_main_pm` →
`test_complete_cell_pm_does_not_escalate_to_main_pm` (asserts
status=COMPLETED, assigned_to != main_pm.id). The no-Main-PM-fallback
test trivially still passes (the path is now the only path).
Lifecycle test comment updated. make quality green.

roboco/services/task.py

@@ -2970,41 +2970,6 @@ class TaskService(BaseService):
         is_own_task = agent_id and task.assigned_to == agent_id
         return task.status == TaskStatus.IN_PROGRESS and bool(is_own_task)
 
-    async def _handle_cell_pm_escalation(
-        self, task: TaskTable, task_id: UUID, agent_id: UUID | None
-    ) -> TaskTable | None:
-        """Handle Cell PM escalation to Main PM. Returns task if escalated."""
-        main_pm_result = await self.session.execute(
-            select(AgentTable)
-            .where(AgentTable.role == AgentRole.MAIN_PM)
-            .order_by(AgentTable.created_at)
-            .limit(1)
-        )
-        main_pm = main_pm_result.scalar_one_or_none()
-        if not main_pm:
-            self.log.warning(
-                "No Main PM found - proceeding with completion", task_id=str(task_id)
-            )
-            return None
-
-        task.assigned_to = cast("Any", main_pm.id)
-        task.claimed_by = cast("Any", main_pm.id)
-        await self.session.flush()
-        await self._emit_task_event(
-            EventType.TASK_ESCALATED_TO_MAIN_PM,
-            task_id,
-            {
-                "main_pm_id": str(main_pm.id),
-                "cell_pm_id": str(agent_id) if agent_id else None,
-            },
-        )
-        self.log.info(
-            "Cell PM approved - escalating to Main PM",
-            task_id=str(task_id),
-            main_pm_id=str(main_pm.id),
-        )
-        return task
-
     async def _validate_completion_prerequisites(
         self, task: TaskTable, task_id: UUID, agent_id: UUID | None
     ) -> list[TaskTable] | None:
@@ -3064,18 +3029,27 @@ class TaskService(BaseService):
         self,
         task: TaskTable,
         task_id: UUID,
-        agent_id: UUID | None,
         completing_agent_role: str | None,
         all_descendants: list[TaskTable],
     ) -> TaskTable | None:
-        """Run the Cell PM → Main PM → CEO chain; return escalated task or None."""
+        """Run the PM-completion approval chain; return escalated task or None.
+
+        #178: the cell_pm branch was removed (it reassigned every
+        ``awaiting_pm_review`` task from the cell PM to the main PM and
+        kept it in ``awaiting_pm_review`` — a legacy two-step "cell PM
+        approves, main PM approves" review chain). The gateway model
+        actually in use forbids ``main_pm_complete`` on any non-root
+        task (``parent_task_id IS NOT NULL`` → invalid_state), so a
+        leaf or cell-level task reassigned that way was permanently
+        wedged — main PM had no verb to advance it. Cell PM completing
+        a non-root task now just transitions it to COMPLETED (the
+        gateway model); the cell→main escalation, when intended,
+        happens via the dedicated ``submit_up`` verb, not via
+        ``complete``. The main_pm → CEO escalation for root parents
+        stays — it's the still-correct second tier.
+        """
         if task.status != TaskStatus.AWAITING_PM_REVIEW:
             return None
-
-        if completing_agent_role == "cell_pm":
-            escalated = await self._handle_cell_pm_escalation(task, task_id, agent_id)
-            if escalated:
-                return escalated
         is_root_parent = all_descendants and not task.parent_task_id
         if completing_agent_role == "main_pm" and is_root_parent:
             self.log.info(
@@ -3132,10 +3106,15 @@ class TaskService(BaseService):
         """
         Mark task as completed (PM only).
 
-        Approval hierarchy:
+        Approval model (post-#178 — matches the gateway invariant
-        1. Cell PM reviews → reassigns to Main PM (same awaiting_pm_review state)
+        ``main_pm_complete`` rejects any non-root task):
-        2. Main PM reviews leaf task → completes
+
-        3. Main PM reviews parent task (all descendants terminal) → escalates to CEO
+        - Cell PM completes a non-root task → COMPLETED. The cell→main
+          escalation, when intended, is the cell PM's ``submit_up``
+          verb on the cell-level parent, not ``complete``.
+        - Main PM completes a leaf/non-root task → COMPLETED.
+        - Main PM completes a root parent (descendants all terminal)
+          → escalates to CEO (``awaiting_ceo_approval``).
         """
         task = await self.get(task_id)
         if not task:
@@ -3149,7 +3128,7 @@ class TaskService(BaseService):
             return None
 
         escalated = await self._apply_complete_approval_chain(
-            task, task_id, agent_id, completing_agent_role, all_descendants
+            task, task_id, completing_agent_role, all_descendants
         )
         if escalated:
             return escalated

tests/integration/test_lifecycle_real_db.py

@@ -619,18 +619,13 @@ async def test_pm_complete_simple_task(
 ) -> None:
     """awaiting_pm_review → cell_pm complete → completed.
 
-    With no Main PM seeded, ``_handle_cell_pm_escalation`` short-circuits
+    Post-#178: cell PM completing a non-root awaiting_pm_review task
-    and the task transitions straight to COMPLETED instead of being
+    always transitions straight to COMPLETED — there is no longer a
-    handed up to the Main PM. This is the "simple task" path the plan
+    cell→main escalation in ``complete`` (the old branch is gone; the
-    calls out — no parent task, no Main PM, no CEO escalation.
+    cell→main hand-off, when intended, uses ``submit_up``). The
-
+    MAIN_PM deletion below is now a historical artifact (kept because
-    Test isolation note: ``test_groups_routes.py`` exercises the
+    other tests in this file rely on the same isolation pattern); it
-    groups POST endpoint, which commits via ``db.commit()`` and
+    no longer affects this test's outcome.
-    persists a MAIN_PM agent across sessions in the test DB. We
-    delete any pre-existing MAIN_PM rows at the top of this test so
-    the cell PM completion is the only one in play. The delete runs
-    inside this test's session and is unwound by the conftest's
-    rollback, so committed state in the shared DB is untouched.
     """
     await db_session.execute(
         delete(AgentTable).where(AgentTable.role == AgentRole.MAIN_PM)

tests/integration/test_task_service_transitions.py

@@ -1090,19 +1090,23 @@ async def test_complete_in_progress_for_own_task(
 
 
 @pytest.mark.asyncio
-async def test_complete_cell_pm_escalates_to_main_pm(
+async def test_complete_cell_pm_does_not_escalate_to_main_pm(
     task_setup: dict, db_session: AsyncSession
 ) -> None:
-    """When a Cell PM completes a task, it gets escalated to Main PM."""
+    """#178: cell_pm completing an awaiting_pm_review non-root task
+    transitions it to COMPLETED and does NOT reassign to main_pm.
+
+    Pre-#178 the cell_pm branch of ``_apply_complete_approval_chain``
+    reassigned every awaiting_pm_review task to main_pm and kept it in
+    awaiting_pm_review for a second-tier review — but the gateway's
+    ``main_pm_complete`` rejects every non-root task
+    (``parent_task_id IS NOT NULL`` → invalid_state), so main_pm had
+    no verb to advance it. Result: the leaf was permanently wedged
+    (observed end-to-end this session). The fix removes the cell_pm
+    escalation branch; cell PM now completes non-root tasks directly,
+    and cell→main escalation, when intended, uses ``submit_up``.
+    """
     svc = task_setup["svc"]
-    # Strip leaked Main PMs so the picked Main PM is the one we're seeding.
-    # `_handle_cell_pm_escalation` orders by created_at and would otherwise
-    # pick a leaked-from-prior-test Main PM with an older timestamp.
-    await db_session.execute(
-        AgentTable.__table__.update()
-        .where(AgentTable.role == AgentRole.MAIN_PM)
-        .values(role=AgentRole.SYSTEM)
-    )
     cell_pm = AgentTable(
         id=uuid4(),
         name="CellPM",
@@ -1133,31 +1137,25 @@ async def test_complete_cell_pm_escalates_to_main_pm(
     await db_session.flush()
     task = await svc.create(_req(task_setup))
     task.status = TaskStatus.AWAITING_PM_REVIEW
+    task.assigned_to = cell_pm.id
     await db_session.flush()
     out = await svc.complete(task.id, agent_id=cell_pm.id)
     assert out is not None
-    # Escalated — task reassigned to main_pm
+    assert out.status == TaskStatus.COMPLETED
-    assert out.assigned_to == main_pm.id
+    assert out.assigned_to != main_pm.id
-    assert out.status == TaskStatus.AWAITING_PM_REVIEW
 
 
 @pytest.mark.asyncio
-async def test_complete_cell_pm_no_main_pm_falls_through(
+async def test_complete_cell_pm_no_main_pm_completes(
     task_setup: dict, db_session: AsyncSession
 ) -> None:
-    """If no Main PM exists, escalation returns None and chain falls through.
+    """#178: cell_pm completing awaiting_pm_review transitions to
-
+    COMPLETED regardless of whether a Main PM exists. (Pre-#178 this
-    Other tests in earlier modules may have committed Main PMs that the
+    test guarded the "no Main PM → escalation returns None → falls
-    rollback fixture can't undo (commits stick). Delete any inside this
+    through to completion" fallback; post-#178 the cell_pm escalation
-    test's transaction so the rollback restores them at teardown — within
+    branch is gone entirely, so this path is the only path.)
-    this test they appear absent.
     """
     svc = task_setup["svc"]
-    await db_session.execute(
-        AgentTable.__table__.update()
-        .where(AgentTable.role == AgentRole.MAIN_PM)
-        .values(role=AgentRole.SYSTEM)
-    )
     cell_pm = AgentTable(
         id=uuid4(),
         name="CellPM",