[F059] self-heal: hold fix tasks for CEO Approve-&-Start (restore dispatch gate)

The module docstring promised self-heal fix tasks 'wait for the CEO's
Approve-&-Start', but _originate created them confirmed_by_human=True and the
orchestrator dispatched them at once — a self-heal fix that re-broke CI would
trigger another cycle, open another auto-dispatched fix, and loop with no CEO
gate on dispatch.

Restore the documented gate:
* _originate opens the task confirmed_by_human=False (held for the CEO).
* The orchestrator holds a self-heal task out of both the PM and dev dispatch
  paths until confirmed_by_human flips True.
* approve_and_start (the CEO's start gate) sets confirmed_by_human=True so the
  held task finally dispatches (idempotent for board/intake tasks already True).
* list_pending_for_agent scopes the give_me_work hold to self-heal
  (source != self_heal OR confirmed_by_human) so an already-alive PM can't grab
  it pre-approval — while ordinary delegated subtasks (confirmed_by_human=False
  by default, where the delegation IS the authorization) still dispatch.

The 'never self-deploys' guarantee (no merge) is unchanged.

tests/unit/runtime/test_self_heal_ceo_gate.py

@@ -0,0 +1,283 @@
+"""F059: self-heal fix tasks must WAIT for the CEO's Approve-&-Start.
+
+The module docstring promises the loop 'only NOTIFIES and, at most, OPENS a
+PENDING task ... the task waits for the CEO's Approve-&-Start and terminates at
+awaiting_ceo_approval'. The implementation did the opposite: it created the
+task ``confirmed_by_human=True`` and the orchestrator dispatched it at once —
+a self-heal fix that re-broke CI would trigger another self-heal cycle, open
+another auto-dispatched fix, and loop with no CEO gate on dispatch.
+
+The fix restores the documented gate:
+* ``_originate`` opens the task ``confirmed_by_human=False`` (held for the CEO).
+* The orchestrator holds a self-heal task out of dispatch until the CEO
+  approves it (``confirmed_by_human`` flips True via ``approve_and_start``).
+* ``give_me_work`` (``list_pending_for_agent``) never offers a held task to an
+  already-alive agent.
+* ``approve_and_start`` is the CEO's start gate — it flips ``confirmed_by_human``
+  True so the held task finally dispatches.
+
+The 'never self-deploys' guarantee (no merge) is unchanged; only the dispatch
+gate is restored.
+"""
+
+from __future__ import annotations
+
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+import roboco.services.self_heal_engine as _self_heal_mod
+from roboco.foundation import identity as _foundation
+from roboco.models.base import TaskStatus, TaskType
+from roboco.runtime.orchestrator import AgentOrchestrator
+from roboco.services.self_heal_engine import RegressionObservation, SelfHealEngine
+from roboco.services.task import (
+    SELF_HEAL_SOURCE,
+    TaskCreateRequest,
+    TaskService,
+)
+from sqlalchemy.dialects import postgresql
+
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+
+
+def _task(
+    tid: str,
+    source: str,
+    *,
+    assigned_to: str | None = None,
+    confirmed: bool = True,
+) -> dict[str, Any]:
+    return {
+        "id": tid,
+        "source": source,
+        "assigned_to": assigned_to,
+        "confirmed_by_human": confirmed,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Orchestrator: a held self-heal task (not yet CEO-confirmed) is NOT dispatched
+# ---------------------------------------------------------------------------
+
+
+def _pm_stub(tasks: list[dict[str, Any]]) -> MagicMock:
+    stub = MagicMock()
+    stub._fetch_tasks = AsyncMock(return_value=tasks)
+    stub._is_task_handled_this_tick = MagicMock(return_value=False)
+    stub._resolve_agent_slug = MagicMock(return_value="main-pm")
+    stub._BOARD_AGENTS = frozenset()
+    stub._route_unassigned_pm_task = AsyncMock()
+    stub._handle_pm_assigned_task = AsyncMock()
+    stub._handle_board_assigned_task = AsyncMock()
+    return stub
+
+
+@pytest.mark.asyncio
+async def test_held_self_heal_task_is_not_dispatched() -> None:
+    """A self-heal fix task the CEO has NOT yet approved (confirmed_by_human=False)
+    is held out of the assigned-PM dispatch path — no autonomous dispatch."""
+    tasks = [
+        _task("A", SELF_HEAL_SOURCE, assigned_to="main-pm", confirmed=False),
+    ]
+    stub = _pm_stub(tasks)
+    client: Any = MagicMock()
+    await AgentOrchestrator._dispatch_pm_work(cast("AgentOrchestrator", stub), client)
+
+    stub._handle_pm_assigned_task.assert_not_awaited()
+    stub._route_unassigned_pm_task.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ceo_approved_self_heal_task_dispatches() -> None:
+    """Once the CEO has approved the self-heal task (confirmed_by_human=True via
+    approve_and_start), it dispatches through the assigned-PM path normally."""
+    tasks = [
+        _task("A", SELF_HEAL_SOURCE, assigned_to="main-pm", confirmed=True),
+        _task("B", SELF_HEAL_SOURCE, assigned_to="main-pm", confirmed=False),
+    ]
+    stub = _pm_stub(tasks)
+    client: Any = MagicMock()
+    await AgentOrchestrator._dispatch_pm_work(cast("AgentOrchestrator", stub), client)
+
+    handled = [c.args[0]["id"] for c in stub._handle_pm_assigned_task.await_args_list]
+    assert handled == ["A"]  # only the CEO-approved one
+
+
+@pytest.mark.asyncio
+async def test_non_self_heal_tasks_dispatch_regardless_of_confirmation() -> None:
+    """The hold is specific to self-heal — an ordinary confirmed-by-human task
+    is NOT gated by this skip (regression guard for the board/intake flow)."""
+    tasks = [
+        _task("A", "manual", assigned_to="main-pm", confirmed=False),
+        _task("B", "manual", assigned_to="main-pm", confirmed=True),
+    ]
+    stub = _pm_stub(tasks)
+    client: Any = MagicMock()
+    await AgentOrchestrator._dispatch_pm_work(cast("AgentOrchestrator", stub), client)
+
+    handled = [c.args[0]["id"] for c in stub._handle_pm_assigned_task.await_args_list]
+    assert set(handled) == {"A", "B"}
+
+
+# ---------------------------------------------------------------------------
+# _originate: the opened task is held for the CEO (confirmed_by_human=False)
+# ---------------------------------------------------------------------------
+
+
+def _originate_engine(captured: dict[str, Any]) -> SelfHealEngine:
+    """An engine whose task/project services are mocks that capture the create
+    request — so the held-for-CEO invariant is unit-testable without a DB."""
+    session = MagicMock()
+
+    task_svc = MagicMock()
+    task_svc.list_open_self_heal_tasks = AsyncMock(return_value=[])
+    task_svc.create = AsyncMock(
+        side_effect=lambda req: (captured.setdefault("req", req), MagicMock(id="t1"))[1]
+    )
+
+    project_svc = MagicMock()
+    project_svc.get_by_slug = AsyncMock(return_value=MagicMock(id="p1"))
+
+    engine = SelfHealEngine.__new__(SelfHealEngine)  # skip __init__ (no source)
+    engine.session = session
+    engine.log = MagicMock()
+    engine._source = MagicMock()
+    return engine, task_svc, project_svc, _self_heal_mod
+
+
+@pytest.mark.asyncio
+async def test_originate_opens_task_held_for_ceo(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The opened self-heal fix task is ``confirmed_by_human=False`` — held for
+    the CEO's Approve-&-Start, NOT auto-confirmed for autonomous dispatch."""
+    captured: dict[str, Any] = {}
+    engine, task_svc, project_svc, mod = _originate_engine(captured)
+    monkeypatch.setattr(mod, "get_task_service", lambda _s: task_svc)
+    monkeypatch.setattr(mod, "get_project_service", lambda _s: project_svc)
+    monkeypatch.setattr(mod, "markers", MagicMock())  # set_self_heal_fingerprint no-op
+
+    obs = [
+        RegressionObservation(
+            fingerprint="fp1",
+            signal_name="ci:roboco",
+            repo_hint="roboco",
+            summary="x",
+            detail="d",
+            raw_ref="r",
+        )
+    ]
+    # Bypass the session.flush (MagicMock session) — _originate awaits it.
+    engine.session.flush = AsyncMock()  # type: ignore[assignment]
+
+    count = await engine._originate(obs)
+
+    assert count == 1
+    req: TaskCreateRequest = captured["req"]
+    assert req.confirmed_by_human is False  # held for the CEO — the F059 fix
+    assert req.status == TaskStatus.PENDING
+    assert req.source == SELF_HEAL_SOURCE
+    assert req.task_type == TaskType.CODE
+
+
+# ---------------------------------------------------------------------------
+# approve_and_start: the CEO's start gate lifts the hold
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_approve_and_start_lifts_the_ceo_hold(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """``approve_and_start`` is the CEO's start gate — it flips
+    ``confirmed_by_human`` True so a held self-heal task finally dispatches."""
+    svc = TaskService(MagicMock())
+    svc.session.flush = AsyncMock()  # type: ignore[assignment]
+    task = MagicMock()
+    task.status = TaskStatus.PENDING
+    task.team = MagicMock()
+    task.team.value = "main_pm"
+    task.board_review_complete = True
+    task.confirmed_by_human = False  # held self-heal task
+    task.assigned_to = "someone-else"
+    task.task_type = MagicMock()
+    task.task_type.value = "code"
+    svc.get = AsyncMock(return_value=task)
+    main_pm = MagicMock(id=MAIN_PM_UUID)
+    agent_svc = MagicMock()
+    agent_svc.get_by_slug = AsyncMock(return_value=main_pm)
+    monkeypatch.setattr(
+        "roboco.services.agent.get_agent_service",
+        MagicMock(return_value=agent_svc),
+    )
+    monkeypatch.setattr(svc, "_activate_batch_root_subtasks", AsyncMock())
+    monkeypatch.setattr(svc, "_emit_task_event", AsyncMock())
+    monkeypatch.setattr(
+        "roboco.services.task.main_pm_cannot_own_code", lambda *_args, **_kwargs: False
+    )
+
+    await svc.approve_and_start(_foundation.AGENTS["main-pm"].uuid, notes=None)
+
+    assert task.confirmed_by_human is True  # the hold lifts on CEO approval
+
+
+# ---------------------------------------------------------------------------
+# list_pending_for_agent: give_me_work never offers a held task
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_list_pending_for_agent_excludes_held_self_heal() -> None:
+    """A held self-heal fix task is never offered via give_me_work — an
+    already-alive PM can't grab it before the CEO approves.
+
+    Asserted at the SQL layer: the query scopes the hold to self-heal
+    (``source != 'self_heal' OR confirmed_by_human``), so the database drops a
+    held self-heal task before the agent ever sees the list.
+    """
+    svc = TaskService(MagicMock())
+    result = MagicMock()
+    result.scalars.return_value.all.return_value = []
+    svc.session.execute = AsyncMock(return_value=result)
+    svc.unmet_dependency_ids = AsyncMock(return_value=[])  # type: ignore[assignment]
+
+    await svc.list_pending_for_agent(MAIN_PM_UUID)
+
+    stmt = svc.session.execute.await_args.args[0]
+    compiled = str(
+        stmt.compile(
+            dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}
+        )
+    )
+    assert "self_heal" in compiled  # scoped to self-heal, not a universal gate
+    assert "confirmed_by_human" in compiled
+
+
+@pytest.mark.asyncio
+async def test_list_pending_for_agent_still_offers_delegated_subtask() -> None:
+    """Regression guard (F059): the hold is scoped to self-heal. A delegated
+    subtask (source != self_heal, confirmed_by_human=False — the default for
+    PM-delegated work, where the delegation IS the authorization to start) must
+    STILL be offered via give_me_work. A universal confirmed_by_human filter
+    would starve devs of all delegated work."""
+    svc = TaskService(MagicMock())
+
+    delegated = MagicMock()
+    delegated.source = "manual"  # not self_heal
+    delegated.confirmed_by_human = False  # the delegation default
+    delegated.dependency_ids = []
+
+    result = MagicMock()
+    result.scalars.return_value.all.return_value = [delegated]
+    svc.session.execute = AsyncMock(return_value=result)
+    svc.unmet_dependency_ids = AsyncMock(return_value=[])  # type: ignore[assignment]
+
+    available = await svc.list_pending_for_agent(MAIN_PM_UUID)
+
+    assert delegated in available
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-q"])

tests/unit/runtime/test_self_heal_dispatch.py

@@ -1,8 +1,11 @@
-"""Self-heal fix tasks dispatch autonomously through the PM dispatcher.
+"""Self-heal fix tasks dispatch through the PM dispatcher only after the CEO
+approves them (F059).
 
-The loop opens a ``source='self_heal'`` task confirmed + assigned to the Main PM
-agent, so the dispatcher routes it through the assigned-PM path like any other
-task — there is no CEO Approve-&-Start hold (that gate is the Intake/board flow).
+The loop opens a ``source='self_heal'`` task assigned to the Main PM agent but
+HELD (``confirmed_by_human=False``) for the CEO's Approve-&-Start. The dispatcher
+holds an unconfirmed self-heal task out of the assigned-PM path; once the CEO's
+``approve_and_start`` flips ``confirmed_by_human`` True, it routes through the
+assigned-PM path like any other PM task. Unassigned tasks route normally.
 """
 
 from __future__ import annotations
@@ -12,17 +15,32 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from roboco.runtime.orchestrator import AgentOrchestrator
+from roboco.services.task import SELF_HEAL_SOURCE
 
 
-def _task(tid: str, source: str, assigned_to: str | None = None) -> dict[str, Any]:
-    return {"id": tid, "source": source, "assigned_to": assigned_to}
+def _task(
+    tid: str,
+    source: str,
+    *,
+    assigned_to: str | None = None,
+    confirmed: bool | None = None,
+) -> dict[str, Any]:
+    task: dict[str, Any] = {"id": tid, "source": source, "assigned_to": assigned_to}
+    if confirmed is not None:
+        task["confirmed_by_human"] = confirmed
+    return task
 
 
 @pytest.mark.asyncio
-async def test_self_heal_task_dispatches_through_the_assigned_pm_path() -> None:
+async def test_ceo_approved_self_heal_task_dispatches_through_assigned_pm_path() -> (
+    None
+):
+    """A self-heal task the CEO has approved (confirmed_by_human=True) is handed
+    to the assigned-PM path — the CEO's gate has lifted."""
     tasks = [
-        _task("A", "self_heal", assigned_to="main-pm"),  # assigned → assigned-PM path
-        _task("B", "self_heal"),  # unassigned self-heal → routing
+        _task(
+            "A", SELF_HEAL_SOURCE, assigned_to="main-pm", confirmed=True
+        ),  # CEO-approved → assigned-PM path
         _task("C", "manual"),  # ordinary unassigned → routing
     ]
     stub = MagicMock()
@@ -37,10 +55,40 @@ async def test_self_heal_task_dispatches_through_the_assigned_pm_path() -> None:
     client: Any = MagicMock()
     await AgentOrchestrator._dispatch_pm_work(cast("AgentOrchestrator", stub), client)
 
-    # The assigned self-heal task is handed to the assigned-PM path (spawned),
-    # NOT held — it dispatches without any CEO approval.
     handled = [c.args[0]["id"] for c in stub._handle_pm_assigned_task.await_args_list]
     assert handled == ["A"]
-    # Unassigned tasks (self-heal or not) route normally.
     routed = [c.args[1]["id"] for c in stub._route_unassigned_pm_task.await_args_list]
-    assert set(routed) == {"B", "C"}
+    assert routed == ["C"]
+
+
+@pytest.mark.asyncio
+async def test_held_self_heal_task_is_not_dispatched() -> None:
+    """A self-heal task the CEO has NOT yet approved (confirmed_by_human=False)
+    is held — neither the assigned-PM path nor routing touches it."""
+    tasks = [
+        _task(
+            "A", SELF_HEAL_SOURCE, assigned_to="main-pm", confirmed=False
+        ),  # held → skip
+        _task("B", SELF_HEAL_SOURCE, confirmed=False),  # held + unassigned → skip
+        _task("C", "manual"),  # ordinary unassigned → routing still happens
+    ]
+    stub = MagicMock()
+    stub._fetch_tasks = AsyncMock(return_value=tasks)
+    stub._is_task_handled_this_tick = MagicMock(return_value=False)
+    stub._resolve_agent_slug = MagicMock(return_value="main-pm")
+    stub._BOARD_AGENTS = frozenset()
+    stub._route_unassigned_pm_task = AsyncMock()
+    stub._handle_pm_assigned_task = AsyncMock()
+    stub._handle_board_assigned_task = AsyncMock()
+
+    client: Any = MagicMock()
+    await AgentOrchestrator._dispatch_pm_work(cast("AgentOrchestrator", stub), client)
+
+    stub._handle_pm_assigned_task.assert_not_awaited()
+    # Only the non-self-heal unassigned task routes; the held self-heal one does not.
+    routed = [c.args[1]["id"] for c in stub._route_unassigned_pm_task.await_args_list]
+    assert routed == ["C"]
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-q"])

tests/unit/services/test_self_heal_originate_db.py

@@ -3,8 +3,9 @@
 The loop opens a fix task only when ``self_heal_originate_enabled``, dedupes one
 open task per regression fingerprint, honors the per-cycle and rolling open-task
 caps, and creates the task PENDING + assigned to the Main PM agent +
-``confirmed_by_human=True`` so it dispatches autonomously (no CEO Approve-&-Start).
-Crucially the loop NEVER calls start / approve / merge / deploy — asserted here.
+``confirmed_by_human=False`` so it is HELD for the CEO's Approve-&-Start (it
+does not dispatch autonomously). Crucially the loop NEVER calls start / approve
+/ merge / deploy — asserted here.
 """
 
 from __future__ import annotations
@@ -148,9 +149,12 @@ async def test_originate_creates_pending_main_pm_assigned_task(
     task = open_tasks[0]
     assert task.status == TaskStatus.PENDING
     # Assigned to the Main PM agent up front (not just team=main_pm) so that, once
-    # the CEO confirms it, the orchestrator dispatches it straight to that agent.
+    # the CEO approves it, the orchestrator dispatches it straight to that agent.
     assert task.assigned_to == MAIN_PM_UUID
-    assert task.confirmed_by_human is True  # auto-confirmed → dispatches autonomously
+    # F059: held for the CEO's Approve-&-Start — NOT auto-confirmed. The
+    # orchestrator + give_me_work keep it out of dispatch until the CEO
+    # approves it (approve_and_start flips this True).
+    assert task.confirmed_by_human is False
     assert task.team == Team.MAIN_PM
     assert task.source == "self_heal"
     assert task.acceptance_criteria  # non-empty (AC-guardrail)
@@ -240,17 +244,20 @@ async def test_loop_never_starts_or_approves(
 
 
 @pytest.mark.asyncio
-async def test_originated_task_is_confirmed_for_autonomous_dispatch(
+async def test_originated_task_is_held_for_ceo_approve_and_start(
     db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """The opened task is confirmed up front, so the PM dispatcher picks it up
-    without any CEO Approve-&-Start (that gate is the Intake/board flow)."""
+    """The opened task is HELD (confirmed_by_human=False) for the CEO's
+    Approve-&-Start — the PM dispatcher does NOT pick it up autonomously; the
+    CEO must approve_and_start it first (F059)."""
     await _seed_project(db_session)
     _enable(monkeypatch)
     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 True  # dispatches autonomously
-    assert task.status == TaskStatus.PENDING  # ready for the PM dispatcher
-    assert task.assigned_to == MAIN_PM_UUID  # straight to the Main PM agent
+    assert task.confirmed_by_human is False  # held for the CEO — no autonomous dispatch
+    assert task.status == TaskStatus.PENDING  # not advanced by the loop
+    assert (
+        task.assigned_to == MAIN_PM_UUID
+    )  # straight to the Main PM agent once approved