fix(self-heal): hold an unconfirmed fix task out of dispatch until CEO approval

Adversarial review found the load-bearing invariant broken at the dispatch
layer: _dispatch_pm_work skipped only PR_REVIEW_SOURCES, so a PENDING
team=main_pm self_heal task (assigned_to=None, confirmed_by_human=False) was
routed to Main PM and spawned BEFORE the CEO approved it — the "never start
until you approve" promise didn't hold.

Fix: the PM dispatcher now also skips source='self_heal' while
confirmed_by_human is False (before the assigned/unassigned split, so it holds
either way); the task still shows in the panel so the CEO can see and approve
it. approve_and_start flips confirmed_by_human=True (the CEO's start IS the
human confirmation), so it dispatches normally afterward. Other sources are
unaffected.

Tests: a unit test that the dispatcher holds an unconfirmed self_heal task but
routes a confirmed one and ordinary tasks, plus a DB test that approve_and_start
flips the gate. (The readiness gate was deliberately not used — a blocker there
marks the task `blocked`; the dispatch skip leaves it cleanly PENDING.)

roboco/runtime/orchestrator.py

@@ -58,7 +58,7 @@ from roboco.models.runtime import (
     WaitingRecord,
 )
 from roboco.seeds.initial_data import AGENT_UUIDS
-from roboco.services.task import PR_REVIEW_SOURCES
+from roboco.services.task import PR_REVIEW_SOURCES, SELF_HEAL_SOURCE
 
 logger = structlog.get_logger()
 
@@ -6757,6 +6757,17 @@ Start now: evidence(task_id="{task_id}")
             # PM hierarchy never routes or spawns them.
             if task.get("source") in PR_REVIEW_SOURCES:
                 continue
+            # A self-heal fix task is opened by the loop but must stay INERT until
+            # the CEO Approve-&-Starts it (which flips confirmed_by_human). Until
+            # then the PM hierarchy must not route, assign, or spawn it — even
+            # though it's already team=main_pm. It still appears in the panel so
+            # the CEO can see and approve it. This guard sits before the
+            # assigned-vs-unassigned split, so it holds whether or not the task
+            # carries an assignee.
+            if task.get("source") == SELF_HEAL_SOURCE and not task.get(
+                "confirmed_by_human"
+            ):
+                continue
             assigned_to = task.get("assigned_to")
             if assigned_to:
                 if self._resolve_agent_slug(assigned_to) in self._BOARD_AGENTS:

roboco/services/task.py

@@ -4444,6 +4444,14 @@ class TaskService(BaseService):
         # team and does not affect dispatch (which routes by assignee, not team).
         task.team = cast("Any", Team.MAIN_PM)
 
+        # A self-heal fix task is opened unconfirmed and held OUT of dispatch
+        # until here (the PM dispatcher skips source='self_heal' while
+        # confirmed_by_human is False). The CEO's Approve-&-Start IS that human
+        # confirmation, so flip the gate now and the task dispatches normally.
+        # Other sources don't carry this hold and are unaffected.
+        if getattr(task, "source", "") == SELF_HEAL_SOURCE:
+            task.confirmed_by_human = True
+
         if notes:
             existing = task.quick_context or ""
             entry = f"approve_and_start_notes:{notes}"

tests/unit/runtime/test_self_heal_dispatch_inert.py

@@ -0,0 +1,44 @@
+"""The PM dispatcher holds an unconfirmed self-heal task OUT of dispatch.
+
+The load-bearing invariant: a source='self_heal' task the loop opened must NOT
+be routed / claimed / spawned while confirmed_by_human is False — it sits inert
+until the CEO Approve-&-Starts it (which flips the flag). Once confirmed it
+routes like any other task. Mirrors how PR-review tasks are skipped.
+"""
+
+from __future__ import annotations
+
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+def _task(tid: str, source: str, confirmed: bool) -> dict[str, Any]:
+    return {
+        "id": tid,
+        "source": source,
+        "confirmed_by_human": confirmed,
+        "assigned_to": None,
+    }
+
+
+@pytest.mark.asyncio
+async def test_unconfirmed_self_heal_task_is_held_out_of_dispatch() -> None:
+    tasks = [
+        _task("A", "self_heal", False),  # held — must NOT route until approved
+        _task("B", "self_heal", True),  # CEO-approved → routes
+        _task("C", "manual", False),  # ordinary task → routes
+    ]
+    stub = MagicMock()
+    stub._fetch_tasks = AsyncMock(return_value=tasks)
+    stub._is_task_handled_this_tick = MagicMock(return_value=False)
+    stub._route_unassigned_pm_task = AsyncMock()
+
+    client: Any = MagicMock()
+    await AgentOrchestrator._dispatch_pm_work(cast("AgentOrchestrator", stub), client)
+
+    routed = [c.args[1]["id"] for c in stub._route_unassigned_pm_task.await_args_list]
+    assert "A" not in routed  # the unconfirmed self-heal task stays inert
+    assert set(routed) == {"B", "C"}

tests/unit/services/test_self_heal_originate_db.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 from unittest.mock import AsyncMock
+from uuid import UUID, uuid4
 
 import pytest
 from roboco.config import settings as cfg
@@ -208,3 +209,43 @@ async def test_loop_never_starts_or_approves(
     open_tasks = await get_task_service(db_session).list_open_self_heal_tasks()
     assert len(open_tasks) == ONE
     assert open_tasks[0].status == TaskStatus.PENDING  # never advanced by the loop
+
+
+@pytest.mark.asyncio
+async def test_ceo_approve_and_start_flips_the_confirmation_gate(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The opened task is unconfirmed (held out of dispatch); the CEO's
+    approve_and_start flips confirmed_by_human=True so it can then dispatch."""
+    await _seed_project(db_session)
+    # approve_and_start reassigns to the main-pm agent — seed it.
+    db_session.add(
+        AgentTable(
+            id=uuid4(),
+            name="Main PM",
+            slug="main-pm",
+            role=AgentRole.MAIN_PM,
+            team=Team.MAIN_PM,
+            status=AgentStatus.ACTIVE,
+            model_config={},
+            system_prompt="pm",
+            capabilities=[],
+            permissions={},
+            metrics={},
+        )
+    )
+    await db_session.flush()
+    _enable(monkeypatch)
+    # approve_and_start emits a stream event; stub it out (no bus in the test).
+    monkeypatch.setattr(TaskService, "_emit_task_event", AsyncMock())
+
+    await SelfHealEngine(
+        db_session, source=_FakeSource([_breach("ci:roboco")])
+    ).run_cycle()
+    task = (await get_task_service(db_session).list_open_self_heal_tasks())[0]
+    assert task.confirmed_by_human is False  # inert: held out of dispatch
+
+    started = await TaskService(db_session).approve_and_start(UUID(str(task.id)))
+    assert started is not None
+    assert started.confirmed_by_human is True  # gate flipped → now dispatchable
+    assert started.status == TaskStatus.PENDING  # reassignment, not a transition