feat(external-pr): CEO decision surface — backend (notify + list + dismiss)

A notification can't be the gate: the reviewer is read-only and the CEO
decides what happens next. Backend for a real PR-review decision queue:

- post_pr_review now notifies the CEO (send_external_pr_reviewed_notification,
  APPROVAL/HIGH, related_task_id) the moment a review lands — server-side
  best-effort (the reviewer has no notify verb).
- TaskService.list_external_pr_reviews_awaiting_decision(): completed
  external_pr reviews the CEO has neither superseded (confirmed_by_human) nor
  dismissed (quick_context dismissed=1 marker).
- TaskService.dismiss_external_pr_review(): CEO declines → marker → leaves queue.
- GET /api/tasks/external-pr-reviews (PM+/CEO) and CEO-only
  POST /api/tasks/{id}/dismiss-external-pr. Supersede already exists.

Panel queue wiring follows in the next commit.

roboco/api/routes/tasks.py

@@ -642,6 +642,30 @@ async def get_awaiting_ceo_approval_tasks(
     return task_list_to_response(tasks)
 
 
+@router.get("/external-pr-reviews", response_model=list[TaskResponse])
+async def get_external_pr_reviews(
+    db: DbSession,
+    agent: CurrentAgentContext,
+    permissions: PermissionServiceDep,
+) -> list[TaskResponse]:
+    """Inbound external PRs that were reviewed and await the CEO's decision.
+
+    The PR-review decision queue: completed external-PR review tasks the CEO has
+    neither superseded nor dismissed. Org-wide; visible to PMs and above.
+    """
+    can_view_all = permissions.can_perform_task_action(agent, TaskAction.VIEW_ALL)
+    is_pm = agent.role in (AgentRole.CELL_PM, AgentRole.MAIN_PM)
+    is_ceo = agent.role == AgentRole.CEO
+    if not (can_view_all or is_pm or is_ceo):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only PMs and management can view the PR-review queue",
+        )
+    service = get_task_service(db)
+    tasks = await service.list_external_pr_reviews_awaiting_decision()
+    return task_list_to_response(tasks)
+
+
 @router.post("/{task_id}/supersede-external-pr")
 async def supersede_external_pr(
     task_id: UUID,
@@ -671,6 +695,33 @@ async def supersede_external_pr(
     return result
 
 
+@router.post("/{task_id}/dismiss-external-pr")
+async def dismiss_external_pr(
+    task_id: UUID,
+    db: DbSession,
+    agent: CurrentAgentContext,
+) -> dict[str, Any]:
+    """CEO declines to act on a reviewed external PR — drop it from the queue.
+
+    The review stays on the GitHub PR; this only records that the CEO chose not
+    to supersede, so the PR-review decision queue stops surfacing it. CEO-only.
+    """
+    if agent.role != AgentRole.CEO:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="only the CEO may dismiss an external-PR review",
+        )
+    service = get_task_service(db)
+    task = await service.dismiss_external_pr_review(task_id)
+    if task is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="external-PR review task not found",
+        )
+    await db.commit()
+    return {"ok": True, "task_id": str(task_id)}
+
+
 @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/services/gateway/choreographer/pr_review.py

@@ -161,6 +161,22 @@ class PRReviewerMixin(_Base):
                 logger.exception(
                     "post_pr_review GitHub post failed", task_id=str(task_id)
                 )
+        # Surface the review to the CEO as an actionable decision (supersede /
+        # dismiss). The reviewer is read-only with no notify verb, so the server
+        # emits it. Best-effort — a notify failure must not fail the review.
+        if pr_number:
+            try:
+                from roboco.services.notification import NotificationService
+
+                await NotificationService().send_external_pr_reviewed_notification(
+                    task_id=str(task_id),
+                    pr_number=pr_number,
+                    pr_url=str(getattr(t, "pr_url", "") or ""),
+                )
+            except Exception:
+                logger.exception(
+                    "post_pr_review CEO notify failed", task_id=str(task_id)
+                )
         return Envelope.ok(
             status=str(t.status),
             task_id=str(task_id),

roboco/services/notification.py

@@ -285,6 +285,47 @@ class NotificationService:
             )
         )
 
+    async def send_external_pr_reviewed_notification(
+        self,
+        task_id: str,
+        pr_number: int,
+        pr_url: str,
+        from_agent: str | None = None,
+        to_ceo: str = "ceo",
+    ) -> None:
+        """Tell the CEO an inbound external PR has been reviewed — their call.
+
+        The PR reviewer is read-only: it posts one change-request and stops. The
+        CEO is the gate on what happens next (supersede the PR — the org takes it
+        over and finishes it — or dismiss it). A passive ping is not enough, so
+        this emits a formal APPROVAL notification carrying ``related_task_id`` so
+        the panel's PR-review decision queue can surface it as an actionable
+        signal. Emitted server-side as ``system`` (the reviewer has no notify
+        verb).
+        """
+        logger.info(
+            "Sending external-PR-reviewed notification to CEO",
+            task_id=task_id,
+            pr_number=pr_number,
+            to_ceo=to_ceo,
+        )
+        body = (
+            f"External PR #{pr_number} has been reviewed and a change-request "
+            f"posted ({pr_url}).\n\nYour call: supersede it (the org takes the "
+            "contribution over and finishes it to our standards) or dismiss it."
+        )
+        await self._create_notification(
+            CreateNotificationParams(
+                notification_type=NotificationType.APPROVAL,
+                priority=NotificationPriority.HIGH,
+                from_agent=from_agent or "system",
+                to_agents=[to_ceo],
+                subject=f"External PR #{pr_number} reviewed — your decision",
+                body=body,
+                related_task_id=task_id,
+            )
+        )
+
     async def send_ack_notification(
         self,
         *,

roboco/services/task.py

@@ -720,6 +720,42 @@ class TaskService(BaseService):
         await self.session.flush()
         return task
 
+    async def list_external_pr_reviews_awaiting_decision(self) -> list[TaskTable]:
+        """Completed external-PR reviews still awaiting the CEO's decision.
+
+        A review is awaiting decision once the reviewer has posted (status
+        COMPLETED) and the CEO has neither superseded it (supersede sets
+        ``confirmed_by_human=True``) nor dismissed it (a ``dismissed=1`` marker
+        in quick_context). This backs the panel's PR-review decision queue.
+        """
+        result = await self.session.execute(
+            select(TaskTable).where(
+                TaskTable.source == "external_pr",
+                TaskTable.status == TaskStatus.COMPLETED,
+                TaskTable.confirmed_by_human.is_(False),
+            )
+        )
+        return [
+            t
+            for t in result.scalars().all()
+            if "dismissed=1" not in (t.quick_context or "").split()
+        ]
+
+    async def dismiss_external_pr_review(self, task_id: UUID) -> TaskTable | None:
+        """CEO declines to act on a reviewed external PR — drop it from the queue.
+
+        Appends a ``dismissed=1`` marker to quick_context so the review leaves
+        ``list_external_pr_reviews_awaiting_decision``. Returns None if the task
+        is missing or is not an external-PR review.
+        """
+        task = await self.get(task_id)
+        if task is None or getattr(task, "source", "") != "external_pr":
+            return None
+        if "dismissed=1" not in (task.quick_context or "").split():
+            task.quick_context = f"{task.quick_context or ''} dismissed=1".strip()
+        await self.session.flush()
+        return task
+
     async def pr_review_claim(
         self, reviewer_agent_id: UUID, task_id: UUID
     ) -> TaskTable | None:

tests/unit/services/test_external_pr_ingest.py

@@ -61,3 +61,39 @@ async def test_multiple_old_shas_still_rereviews_new() -> None:
     svc = _service(["external_pr_head=abc", "external_pr_head=def"])
     assert await svc.external_review_task_exists(uuid4(), 170, "ghi") is False
     assert await svc.external_review_task_exists(uuid4(), 170, "def") is True
+
+
+def _bind(svc: TaskService, name: str, value: object) -> None:
+    object.__setattr__(svc, name, value)
+
+
+@pytest.mark.asyncio
+async def test_list_awaiting_decision_excludes_dismissed() -> None:
+    pending = MagicMock(quick_context="external_pr_head=abc")
+    dismissed = MagicMock(quick_context="external_pr_head=def dismissed=1")
+    svc = _service([pending, dismissed])
+    out = await svc.list_external_pr_reviews_awaiting_decision()
+    assert out == [pending]
+
+
+@pytest.mark.asyncio
+async def test_dismiss_marks_and_is_idempotent() -> None:
+    task = MagicMock(source="external_pr", quick_context="external_pr_head=abc")
+    session = MagicMock()
+    session.flush = AsyncMock()
+    svc = TaskService(session)
+    _bind(svc, "get", AsyncMock(return_value=task))
+    await svc.dismiss_external_pr_review(uuid4())
+    assert "dismissed=1" in task.quick_context.split()
+    await svc.dismiss_external_pr_review(uuid4())  # idempotent
+    assert task.quick_context.split().count("dismissed=1") == 1
+
+
+@pytest.mark.asyncio
+async def test_dismiss_rejects_non_external_pr() -> None:
+    task = MagicMock(source="code")
+    session = MagicMock()
+    session.flush = AsyncMock()
+    svc = TaskService(session)
+    _bind(svc, "get", AsyncMock(return_value=task))
+    assert await svc.dismiss_external_pr_review(uuid4()) is None