feat(self-heal): wire the dormant orchestrator loop

Register _self_heal_loop alongside the other background loops (created in
start, cancelled in stop). It returns immediately unless self_heal_enabled, so
a standard deployment adds zero behaviour and makes no CI call; when on it runs
one engine cycle per interval and commits any opened fix task. A test pins the
default-off dormancy (no sleep / CI / DB when disabled).

roboco/runtime/orchestrator.py

@@ -623,6 +623,7 @@ class AgentOrchestrator:
         self._rate_limit_probe_task: asyncio.Task | None = None
         self._strategy_engine_task: asyncio.Task | None = None
         self._external_pr_poll_task: asyncio.Task | None = None
+        self._self_heal_task: asyncio.Task | None = None
         # Tracks which providers have already received a CEO notification
         # during the current rate-limit episode.  Cleared when the probe
         # succeeds and the rate limit is lifted (tracker.clear() path).
@@ -707,6 +708,7 @@ class AgentOrchestrator:
         self._rate_limit_probe_task = asyncio.create_task(self._rate_limit_probe_loop())
         self._strategy_engine_task = asyncio.create_task(self._strategy_engine_loop())
         self._external_pr_poll_task = asyncio.create_task(self._external_pr_poll_loop())
+        self._self_heal_task = asyncio.create_task(self._self_heal_loop())
 
         logger.info(
             "Orchestrator started",
@@ -749,6 +751,11 @@ class AgentOrchestrator:
             with contextlib.suppress(asyncio.CancelledError):
                 await self._external_pr_poll_task
 
+        if self._self_heal_task:
+            self._self_heal_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._self_heal_task
+
         # Stop all agents
         for agent_id in list(self._instances.keys()):
             await self.stop_agent(agent_id)
@@ -4661,6 +4668,32 @@ Start by:
             except Exception:
                 logger.exception("external-PR poll cycle failed")
 
+    async def _self_heal_loop(self) -> None:
+        """Engine 4: watch RoboCo's OWN CI, surface regressions, open fix tasks.
+
+        Dormant by default — returns immediately unless ``self_heal_enabled``, so
+        a standard deployment makes no CI call and adds zero behaviour. It only
+        NOTIFIES the CEO and (behind ``self_heal_originate_enabled``) opens a
+        PENDING fix task into RoboCo's own lifecycle; it never starts, merges, or
+        deploys. The per-cycle session commits any opened task here.
+        """
+        if not settings.self_heal_enabled:
+            return
+        from roboco.db import get_db_context
+        from roboco.services.self_heal_engine import get_self_heal_engine
+
+        interval = settings.self_heal_interval_seconds
+        while self._running:
+            try:
+                await asyncio.sleep(interval)
+                async with get_db_context() as db:
+                    await get_self_heal_engine(db).run_cycle()
+                    await db.commit()
+            except asyncio.CancelledError:
+                break
+            except Exception:
+                logger.exception("self-heal cycle failed")
+
     @staticmethod
     def _repo_key(git_url: str) -> str:
         """Normalized repo identity (case/.git/trailing-slash insensitive)."""

tests/unit/runtime/test_self_heal_loop_dormant.py

@@ -0,0 +1,26 @@
+"""The self-heal orchestrator loop is fully dormant when disabled (the default).
+
+With ``self_heal_enabled`` off, ``_self_heal_loop`` must return immediately —
+no sleep, no CI call, no DB — so a standard deployment behaves exactly as today.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import types
+from typing import cast
+
+import pytest
+from roboco.config import settings as cfg
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+@pytest.mark.asyncio
+async def test_self_heal_loop_returns_immediately_when_disabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(cfg, "self_heal_enabled", False)
+    stub = cast("AgentOrchestrator", types.SimpleNamespace(_running=True))
+    # Gated off → returns at once. If the gate were missing it would sleep the
+    # full interval and this wait_for would time out.
+    await asyncio.wait_for(AgentOrchestrator._self_heal_loop(stub), timeout=1.0)