From 606ccc3327636d0649a0d1c7d10776ad292878d0 Mon Sep 17 00:00:00 2001 From: Renn F Date: Wed, 17 Jun 2026 20:52:47 +0200 Subject: [PATCH] =?UTF-8?q?feat(self-heal):=20regression=20engine=20?= =?UTF-8?q?=E2=80=94=20detect=20+=20notify=20the=20CEO=20(dormant)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 113 +++++++++++++++++++ tests/unit/services/test_self_heal_engine.py | 98 ++++++++++++++++ 2 files changed, 211 insertions(+) create mode 100644 roboco/services/self_heal_engine.py create mode 100644 tests/unit/services/test_self_heal_engine.py diff --git a/roboco/services/self_heal_engine.py b/roboco/services/self_heal_engine.py new file mode 100644 index 00000000..e34916ff --- /dev/null +++ b/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) diff --git a/tests/unit/services/test_self_heal_engine.py b/tests/unit/services/test_self_heal_engine.py new file mode 100644 index 00000000..0f4c4441 --- /dev/null +++ b/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