feat(gateway): CEO-triggered supersede of a reviewed external PR

The org takes over a reviewed external PR and finishes it itself:

- POST /api/tasks/{id}/supersede-external-pr (CEO-only) -> orchestrator
  .supersede_external_pr: confirms the review task (this CEO action authorizes
  running the contributor's code), cuts a roboco-owned branch off the fork head
  (create_branch_from_pr_head — the only point untrusted code enters a roboco
  branch), and creates the supersede umbrella.
- TaskService.create_supersede_umbrella: a planning task on the same repo,
  parented to the review task (contributor PR# stays reachable for close-on-land),
  carrying the pre-cut fork branch, handed to Main PM to delegate to a cell.
  confirmed_by_human=True.

From there the work rides the normal lifecycle (Main PM -> cell -> our PR ->
QA/doc/PM -> CEO). ruff + mypy clean (279); app builds with the new route.
Follow-up: close-on-land of the contributor PR + an adversarial review pass.

roboco/api/routes/tasks.py

@@ -642,6 +642,35 @@ async def get_awaiting_ceo_approval_tasks(
     return task_list_to_response(tasks)
 
 
+@router.post("/{task_id}/supersede-external-pr")
+async def supersede_external_pr(
+    task_id: UUID,
+    agent: CurrentAgentContext,
+) -> dict[str, Any]:
+    """CEO-authorized takeover of a reviewed external PR.
+
+    Confirms the review task and hands the contribution to the org: a
+    roboco-owned branch is cut from the contributor's fork head and a supersede
+    task is created for Main PM to delegate to a cell. This is the human
+    confirmation that authorizes fetching + running the contributor's code, so
+    it is CEO-only.
+    """
+    if agent.role != AgentRole.CEO:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="only the CEO may supersede an external PR",
+        )
+    from roboco.api.deps import get_orchestrator
+
+    result = await get_orchestrator().supersede_external_pr(task_id)
+    if not result.get("ok"):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(result.get("error", "supersede failed")),
+        )
+    return result
+
+
 @router.get("/lifecycle-transitions", response_model=dict[str, list[str]])
 async def get_lifecycle_transitions() -> dict[str, list[str]]:
     """Return the task lifecycle state graph as a JSON-serialisable dict.

roboco/runtime/orchestrator.py

@@ -4622,6 +4622,54 @@ Start by:
         assoc = (pr.get("author_association") or "").upper()
         return assoc not in trusted
 
+    async def supersede_external_pr(self, review_task_id: "UUID") -> dict[str, Any]:
+        """CEO-authorized takeover of a reviewed external PR.
+
+        Confirms the review task (this CEO action is the human confirmation that
+        authorizes running the contributor's code), cuts a roboco-owned branch
+        off the contributor's fork head (refs/pull/{n}/head — the only point
+        untrusted code enters a roboco branch), and creates the supersede
+        umbrella for Main PM to delegate to a cell. Returns a status dict.
+        """
+        from roboco.db import get_db_context
+        from roboco.services.git import GitService
+        from roboco.services.project import get_project_service
+        from roboco.services.task import get_task_service
+
+        async with get_db_context() as db:
+            task_service = get_task_service(db)
+            review = await task_service.get(review_task_id)
+            if review is None or getattr(review, "source", "") != "external_pr":
+                return {"ok": False, "error": "not an external-PR review task"}
+            if not review.project_id or not review.pr_number:
+                return {
+                    "ok": False,
+                    "error": "review task missing project or pr_number",
+                }
+            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)
+            # The CEO authorized fetching + finishing the contributor's code.
+            review.confirmed_by_human = True
+            await db.flush()
+            git = GitService(db)
+            system_id = _foundation.AGENTS["system"].uuid
+            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
+            await db.commit()
+        self._dispatch_wake.set()
+        return {"ok": True, "supersede_task_id": umbrella_id, "branch": branch_name}
+
     async def _rate_limit_probe_loop(self) -> None:
         """Background loop: probe rate-limited providers every ~30 seconds.
 

roboco/services/task.py

@@ -738,6 +738,62 @@ class TaskService(BaseService):
         self.log.info("External PR review complete", task_id=str(task_id))
         return task
 
+    async def create_supersede_umbrella(
+        self, *, review_task_id: UUID, branch_name: str, created_by: UUID
+    ) -> 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
+        contributor PR number stays reachable as ``review.pr_number`` for
+        close-on-land), handed to Main PM to delegate the code work to a cell.
+        ``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``
+        is True — the CEO authorized this supersede. Returns None if the review
+        task is missing or is not an external-PR review.
+        """
+        review = await self.get(review_task_id)
+        if review is None or getattr(review, "source", "") != "external_pr":
+            return None
+        pr_number = review.pr_number
+        req = TaskCreateRequest(
+            title=f"Supersede external PR #{pr_number}: finish + harden it ourselves",
+            description=(
+                f"The org reviewed external PR #{pr_number} and is taking it over. "
+                f"A roboco-owned branch ('{branch_name}') has been cut from the "
+                "contributor's commits. Delegate the work to the appropriate cell "
+                "to finish + harden it to our standards on that branch, open our "
+                "own PR, and merge it; the contributor PR is closed and linked on "
+                "land. Never push to the contributor's fork."
+            ),
+            acceptance_criteria=[
+                "The contributor's PR is superseded by our own merged PR",
+                "The work meets the project's quality standards",
+            ],
+            team=Team.MAIN_PM,
+            created_by=created_by,
+            task_type=TaskType.PLANNING,
+            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.
+        umbrella.branch_name = branch_name
+        await self.session.flush()
+        self.log.info(
+            "Supersede umbrella created",
+            task_id=str(umbrella.id),
+            review_task_id=str(review_task_id),
+            pr_number=pr_number,
+        )
+        return umbrella
+
     async def _inherit_parent_session(
         self,
         task_id: UUID,