fix(dispatch): restart-safe PM review turns

A leaf task in awaiting_pm_review had no periodic pickup: the closure
dispatcher bailed on childless tasks and skipped PR-bearing review
tasks as already-promoted, assuming the submit-time PM session was
still alive — an assumption every restart breaks. Proven live on the
docs-sync leaf after the 0.25.0 redeploy, which also dependency-blocked
its sibling dev task. Childless awaiting_pm_review tasks now flow to
the PM's review turn, and the merge turn respawns its PM when none is
active.

roboco/runtime/orchestrator.py

@@ -12090,11 +12090,15 @@ Start now: evidence(task_id="{task_id}")
 
     @staticmethod
     def _already_promoted_for_closure(task: dict[str, Any]) -> bool:
-        """Skip closure respawn when PR+status show task has moved up."""
+        """Skip closure respawn when PR+status show task has moved past the PM.
+
+        ``awaiting_pm_review`` is deliberately NOT in the skip set: it is the
+        PM's own merge/review turn, and the session that submitted may be gone
+        (restart, idle-reap) — the ``_is_agent_active`` check below already
+        prevents double-spawning while a PM is genuinely alive."""
         return bool(
             task.get("pr_number")
-            and task.get("status")
-            in ("awaiting_pm_review", "awaiting_ceo_approval", "completed")
+            and task.get("status") in ("awaiting_ceo_approval", "completed")
         )
 
     @staticmethod
@@ -12314,7 +12318,15 @@ Start now: evidence(task_id="{task_id}")
             return
 
         descendants = await self._fetch_all_descendants(client, task_id)
-        if not descendants or not self._all_descendants_terminal(descendants):
+        # A childless task in awaiting_pm_review IS the PM's review turn —
+        # its dev→qa→doc stages ran on the task itself. Without this, a leaf
+        # review stranded by a restart (the submit-time PM session gone) had
+        # no periodic pickup at all — proven live on the docs-sync leaf
+        # after the 0.25.0 redeploy.
+        is_leaf_review = not descendants and task.get("status") == "awaiting_pm_review"
+        if not is_leaf_review and (
+            not descendants or not self._all_descendants_terminal(descendants)
+        ):
             return
         if self._already_promoted_for_closure(task):
             return

tests/unit/runtime/test_pm_closure_auto_resume.py

@@ -161,3 +161,38 @@ async def test_auto_recover_blocked_swallows_errors() -> None:
 
     # Must not raise.
     await orch._auto_recover_blocked_parent(client, "p")
+
+
+@pytest.mark.asyncio
+async def test_childless_awaiting_pm_review_reaches_closure() -> None:
+    """A leaf task in awaiting_pm_review (dev->qa->doc ran on the task itself)
+    is the PM's review turn — it must flow past the descendants gate, or a
+    restart-stranded review has no periodic pickup at all."""
+    orch = _ready_orch()
+    orch._fetch_all_descendants = AsyncMock(return_value=[])
+    orch._closure_handled_without_pm = AsyncMock(return_value=(True, None))
+    task = {"id": "t1", "status": "awaiting_pm_review", "team": "frontend"}
+    await orch._maybe_spawn_pm_closure(MagicMock(), task)
+    orch._closure_handled_without_pm.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_childless_in_progress_still_bails() -> None:
+    """A childless claimed/in_progress task is a dev's work, not PM closure
+    material — the descendants gate must still bail."""
+    orch = _ready_orch()
+    orch._fetch_all_descendants = AsyncMock(return_value=[])
+    orch._closure_handled_without_pm = AsyncMock(return_value=(True, None))
+    task = {"id": "t1", "status": "in_progress", "team": "frontend"}
+    await orch._maybe_spawn_pm_closure(MagicMock(), task)
+    orch._closure_handled_without_pm.assert_not_awaited()
+
+
+def test_promoted_skip_excludes_the_pm_merge_turn() -> None:
+    """awaiting_pm_review IS the PM's turn — a PR-bearing task there must not
+    be skipped as already-promoted (the submit-time PM session may be gone)."""
+    promoted = AgentOrchestrator._already_promoted_for_closure
+    assert not promoted({"pr_number": 5, "status": "awaiting_pm_review"})
+    assert promoted({"pr_number": 5, "status": "awaiting_ceo_approval"})
+    assert promoted({"pr_number": 5, "status": "completed"})
+    assert not promoted({"pr_number": None, "status": "completed"})