fix(supersede): resolve adversarial-review findings (depth, dedup, gate)

A second adversarial review of the supersede flow found a feature-breaking HIGH
plus correctness gaps:

- HIGH: parenting the umbrella to the review task burned a MAX_TASK_DEPTH level
  (review->umbrella->cell-PM->dev = depth 4 > 3), so the dev code task could
  never be created and the work never reached a cell. Fix: the umbrella is now a
  ROOT task; the contributor PR# + review link ride quick_context (also
  simplifies close-on-land — no parent walk).
- MED: no dedup — a repeat CEO trigger created duplicate umbrellas / two racing
  PRs. Fix: find_supersede_umbrella() makes the trigger idempotent (returns the
  existing umbrella).
- LOW: supersede worked on un-reviewed/cancelled review tasks. Fix: require
  review.status == COMPLETED (review-first).
- LOW: a partial failure could orphan a pushed fork branch. Fix: create the
  umbrella before the push, and log the branch so any orphan is discoverable.

ruff + mypy clean (279).

roboco/runtime/orchestrator.py

@@ -4632,6 +4632,7 @@ Start by:
         umbrella for Main PM to delegate to a cell. Returns a status dict.
         """
         from roboco.db import get_db_context
+        from roboco.models.base import TaskStatus
         from roboco.services.git import GitService
         from roboco.services.project import get_project_service
         from roboco.services.task import get_task_service
@@ -4646,26 +4647,51 @@ Start by:
                     "ok": False,
                     "error": "review task missing project or pr_number",
                 }
+            # Review-first: only supersede a PR the org has actually reviewed.
+            if review.status != TaskStatus.COMPLETED:
+                return {
+                    "ok": False,
+                    "error": "review not complete — review the PR first",
+                }
             project = await get_project_service(db).get(cast("UUID", review.project_id))
             if project is None:
                 return {"ok": False, "error": "project not found"}
             pr_number = int(review.pr_number)
+            project_id = cast("UUID", review.project_id)
+            # Idempotent: a repeat call returns the existing umbrella — no second
+            # branch cut, no duplicate cell takeover.
+            existing = await task_service.find_supersede_umbrella(project_id, pr_number)
+            if existing is not None:
+                return {
+                    "ok": True,
+                    "supersede_task_id": str(existing.id),
+                    "branch": existing.branch_name,
+                    "already_superseded": True,
+                }
+            system_id = _foundation.AGENTS["system"].uuid
+            branch_name = f"feature/main_pm/supersede-pr-{pr_number}"
             # The CEO authorized fetching + finishing the contributor's code.
             review.confirmed_by_human = True
-            await db.flush()
+            # Create the umbrella BEFORE the push: a create failure then can't
-            git = GitService(db)
+            # orphan a pushed branch. Only a commit failure after the push could
-            system_id = _foundation.AGENTS["system"].uuid
+            # (rare) — the branch is logged so an orphan stays discoverable.
-            workspace = await git.get_workspace(project.slug, agent_id=system_id)
-            branch_name = f"feature/main_pm/supersede-pr-{pr_number}"
-            await git.create_branch_from_pr_head(
-                workspace, project.slug, pr_number, branch_name
-            )
             umbrella = await task_service.create_supersede_umbrella(
                 review_task_id=review_task_id,
                 branch_name=branch_name,
                 created_by=system_id,
             )
             umbrella_id = str(umbrella.id) if umbrella is not None else None
+            git = GitService(db)
+            workspace = await git.get_workspace(project.slug, agent_id=system_id)
+            logger.warning(
+                "supersede: cutting roboco branch off untrusted fork PR head",
+                branch=branch_name,
+                pr_number=pr_number,
+                project=project.slug,
+            )
+            await git.create_branch_from_pr_head(
+                workspace, project.slug, pr_number, branch_name
+            )
             await db.commit()
         self._dispatch_wake.set()
         return {"ok": True, "supersede_task_id": umbrella_id, "branch": branch_name}

roboco/services/task.py

@@ -743,9 +743,11 @@ class TaskService(BaseService):
     ) -> TaskTable | None:
         """Create the supersede coordination task for a reviewed external PR.
 
-        A planning task on the same repo, parented to the review task (so the
+        A ROOT planning task on the same repo (NOT parented to the review task —
-        contributor PR number stays reachable as ``review.pr_number`` for
+        that would burn a MAX_TASK_DEPTH level and block the cell->dev
-        close-on-land), handed to Main PM to delegate the code work to a cell.
+        decomposition), handed to Main PM to delegate the code work to a cell.
+        The contributor PR number + review-task id are carried in
+        ``quick_context`` (so close-on-land and dedup don't need a parent walk).
         ``branch_name`` is the roboco-owned branch already cut from the
         contributor's fork head, so the delegated code subtask builds on the
         contributor's commits (we never push to the fork). ``confirmed_by_human``
@@ -776,15 +778,18 @@ class TaskService(BaseService):
             nature=TaskNature.TECHNICAL,
             estimated_complexity=Complexity.MEDIUM,
             project_id=cast("UUID", review.project_id),
-            parent_task_id=review_task_id,
             source="external_pr_supersede",
             confirmed_by_human=True,
         )
         umbrella = await self.create(req)
         # Carry the pre-cut fork branch so the delegated code subtask cuts off
         # the contributor's commits (via _resolve_base_branch's parent-branch
-        # rule), not the default branch.
+        # rule), not the default branch. The marker links back to the review +
+        # contributor PR for dedup and close-on-land (no parent link needed).
         umbrella.branch_name = branch_name
+        umbrella.quick_context = (
+            f"external_pr_supersede pr={pr_number} review={review_task_id}"
+        )
         await self.session.flush()
         self.log.info(
             "Supersede umbrella created",
@@ -794,6 +799,28 @@ class TaskService(BaseService):
         )
         return umbrella
 
+    async def find_supersede_umbrella(
+        self, project_id: UUID, pr_number: int
+    ) -> TaskTable | None:
+        """The existing (non-cancelled) supersede umbrella for this PR, or None.
+
+        Idempotency for the supersede trigger — a repeat CEO call must not cut a
+        second branch or spawn a second umbrella. Matches the ``quick_context``
+        marker exactly (``pr={n} review=`` won't false-match pr=50 for pr=5).
+        """
+        result = await self.session.execute(
+            select(TaskTable).where(
+                TaskTable.project_id == project_id,
+                TaskTable.source == "external_pr_supersede",
+                TaskTable.status != TaskStatus.CANCELLED,
+            )
+        )
+        needle = f"pr={pr_number} review="
+        for task in result.scalars().all():
+            if needle in (task.quick_context or ""):
+                return task
+        return None
+
     async def _inherit_parent_session(
         self,
         task_id: UUID,