feat(self-heal): regression engine — detect + notify the CEO (dormant)

The detect side of the self-healing loop, modeled on the strategy engine: a
pure assess() turns breaching telemetry samples into RegressionObservations
(with a stable per-signal fingerprint for later dedupe), and run_cycle() is a
no-op unless self_heal_enabled and otherwise only sends the CEO one
ack-notification per regression. Detect + notify only — it never originates,
starts, merges, or deploys; the telemetry source is injectable for testing.
6 unit tests.

roboco/services/self_heal_engine.py

@@ -0,0 +1,113 @@
+"""Production self-healing engine ("engine 4") — dormant by default.
+
+RoboCo heals ITSELF. This watches RoboCo's OWN repo for a regression (a failing
+CI run on its default branch, via the telemetry source) and, when it sees one,
+surfaces it to the CEO — and, behind a second opt-in, opens a PENDING fix task
+into RoboCo's own delivery lifecycle and STOPS. It is deliberately conservative:
+
+* **Default OFF.** ``self_heal_enabled`` is False, so the orchestrator loop never
+  starts and the existing system is completely unaffected.
+* **Never self-deploys.** Even fully enabled, the loop only NOTIFIES and, at
+  most, OPENS a PENDING task. It never starts, approves, merges, or deploys work
+  — every downstream step stays a human/CEO decision (the task waits for the
+  CEO's Approve-&-Start and terminates at ``awaiting_ceo_approval``).
+* **Repo-singular.** It targets only RoboCo's own project
+  (``self_heal_project_slug``); the org's repo-agnostic delivery work is separate.
+* **Bounded + deduped.** One pass per interval; at most one open fix task per
+  signal fingerprint; per-cycle and rolling open-task caps; the notification
+  layer's purpose-dedup suppresses repeat CEO pings until acknowledged.
+
+This slice is detect + notify only; task origination is layered on next, behind
+the second flag.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from roboco.config import settings
+from roboco.services.base import BaseService
+from roboco.services.notification import NotificationService
+from roboco.services.telemetry import get_ci_telemetry_source
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from roboco.services.telemetry import TelemetrySource
+
+
+@dataclass(frozen=True)
+class RegressionObservation:
+    """One regression the engine detected in RoboCo's own repo."""
+
+    fingerprint: str  # stable per signal — the dedupe key for open fix tasks
+    signal_name: str
+    repo_hint: str
+    summary: str
+    detail: str
+    raw_ref: str
+
+
+def _fingerprint(signal_name: str) -> str:
+    """Stable short hash of the signal (which already encodes the repo)."""
+    return hashlib.sha256(signal_name.encode("utf-8")).hexdigest()[:16]
+
+
+class SelfHealEngine(BaseService):
+    """Detect a regression in RoboCo's own repo; surface it (and later open a fix)."""
+
+    service_name = "self_heal_engine"
+
+    def __init__(
+        self, session: AsyncSession, source: TelemetrySource | None = None
+    ) -> None:
+        super().__init__(session)
+        self._source: TelemetrySource = source or get_ci_telemetry_source(session)
+
+    async def assess(self) -> list[RegressionObservation]:
+        """Read telemetry and return observations. Pure — no side effects."""
+        observations: list[RegressionObservation] = []
+        for sample in await self._source.fetch():
+            if not sample.is_breach:
+                continue
+            observations.append(
+                RegressionObservation(
+                    fingerprint=_fingerprint(sample.signal_name),
+                    signal_name=sample.signal_name,
+                    repo_hint=sample.repo_hint,
+                    summary=f"Regression detected on {sample.repo_hint}.",
+                    detail=sample.detail
+                    or f"{sample.signal_name} breached its threshold.",
+                    raw_ref=sample.raw_ref,
+                )
+            )
+        return observations
+
+    async def run_cycle(self) -> list[RegressionObservation]:
+        """Assess and notify the CEO. No-op unless self-healing is enabled.
+
+        Detect + notify only; never starts, merges, or deploys anything.
+        """
+        if not settings.self_heal_enabled:
+            return []
+        observations = await self.assess()
+        if not observations:
+            return []
+        notifier = NotificationService()
+        for obs in observations:
+            body = f"[self-heal] {obs.summary}\n\n{obs.detail}"
+            if obs.raw_ref:
+                body += f"\n\nEvidence: {obs.raw_ref}"
+            await notifier.send_ack_notification(
+                from_agent="system", to_agent="ceo", body=body
+            )
+        return observations
+
+
+def get_self_heal_engine(
+    session: AsyncSession, source: TelemetrySource | None = None
+) -> SelfHealEngine:
+    """Construct a SelfHealEngine bound to ``session`` (optionally a test source)."""
+    return SelfHealEngine(session, source=source)

tests/unit/services/test_self_heal_engine.py

@@ -0,0 +1,98 @@
+"""Self-heal regression engine — pure assess + gated, notify-only run_cycle.
+
+``assess()`` turns breaching telemetry samples into observations with no side
+effects; ``run_cycle()`` is a no-op unless ``self_heal_enabled`` and otherwise
+only notifies the CEO (this slice never originates, starts, merges, or deploys).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from roboco.config import settings as cfg
+from roboco.services.notification import NotificationService
+from roboco.services.self_heal_engine import SelfHealEngine
+from roboco.services.telemetry import TelemetrySample
+
+
+class _FakeSource:
+    """A TelemetrySource stand-in returning canned samples."""
+
+    def __init__(self, samples: list[TelemetrySample]) -> None:
+        self._samples = samples
+
+    async def fetch(self) -> list[TelemetrySample]:
+        return list(self._samples)
+
+
+def _sample(value: float) -> TelemetrySample:
+    return TelemetrySample(
+        signal_name="ci_conclusion:roboco",
+        value=value,
+        threshold=1.0,
+        window="latest_completed_run",
+        repo_hint="roboco",
+        observed_at="2026-06-17T00:00:00Z",
+        raw_ref="https://github.com/x/roboco/actions/runs/1",
+        detail="CI on roboco@master concluded 'failure'",
+    )
+
+
+def _engine(samples: list[TelemetrySample]) -> SelfHealEngine:
+    return SelfHealEngine(MagicMock(), source=_FakeSource(samples))
+
+
+@pytest.mark.asyncio
+async def test_assess_breach_yields_observation() -> None:
+    obs = await _engine([_sample(1.0)]).assess()
+    assert len(obs) == 1
+    assert obs[0].repo_hint == "roboco"
+    assert obs[0].signal_name == "ci_conclusion:roboco"
+    assert obs[0].fingerprint  # non-empty stable hash
+
+
+@pytest.mark.asyncio
+async def test_assess_no_breach_yields_nothing() -> None:
+    assert await _engine([_sample(0.0)]).assess() == []
+
+
+@pytest.mark.asyncio
+async def test_assess_is_pure_no_notification(monkeypatch: pytest.MonkeyPatch) -> None:
+    send = AsyncMock()
+    monkeypatch.setattr(NotificationService, "send_ack_notification", send)
+    await _engine([_sample(1.0)]).assess()
+    send.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_run_cycle_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(cfg, "self_heal_enabled", False)
+    send = AsyncMock()
+    monkeypatch.setattr(NotificationService, "send_ack_notification", send)
+    assert await _engine([_sample(1.0)]).run_cycle() == []
+    send.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_run_cycle_notifies_ceo_when_enabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(cfg, "self_heal_enabled", True)
+    send = AsyncMock()
+    monkeypatch.setattr(NotificationService, "send_ack_notification", send)
+    obs = await _engine([_sample(1.0)]).run_cycle()
+    assert len(obs) == 1
+    send.assert_awaited_once()
+    call = send.await_args
+    assert call is not None
+    assert call.kwargs["from_agent"] == "system"
+    assert call.kwargs["to_agent"] == "ceo"
+    assert "[self-heal]" in call.kwargs["body"]
+
+
+@pytest.mark.asyncio
+async def test_fingerprint_is_stable() -> None:
+    a = (await _engine([_sample(1.0)]).assess())[0].fingerprint
+    b = (await _engine([_sample(1.0)]).assess())[0].fingerprint
+    assert a == b