From bd1fb841982ecf3b827d9d2f1a809291c19a148f Mon Sep 17 00:00:00 2001 From: Renn F Date: Wed, 17 Jun 2026 21:00:45 +0200 Subject: [PATCH] feat(self-heal): open a PENDING fix task on regression, then stop Behind the second opt-in (self_heal_originate_enabled), a detected regression also opens a fix task into RoboCo's own delivery lifecycle and STOPS: PENDING, unassigned, confirmed_by_human=False, team=main_pm, source=self_heal, with synthesized acceptance criteria and a self_heal_fp= dedupe marker. It rides the normal dispatchers only once the CEO Approve-&-Starts it; the loop itself never calls start / approve / merge / deploy. - TaskService: SELF_HEAL_SOURCE, extract_self_heal_fingerprint, and list_open_self_heal_tasks (the dedupe + open-cap basis) - SelfHealEngine._originate: per-signal fingerprint dedupe, per-cycle and rolling open-task caps, repo resolved to RoboCo's own project (notify-only when it can't be resolved) - 7 DB-backed tests including the never-start / never-approve invariant --- roboco/services/self_heal_engine.py | 103 ++++++++- roboco/services/task.py | 35 +++ .../services/test_self_heal_originate_db.py | 210 ++++++++++++++++++ 3 files changed, 345 insertions(+), 3 deletions(-) create mode 100644 tests/unit/services/test_self_heal_originate_db.py diff --git a/roboco/services/self_heal_engine.py b/roboco/services/self_heal_engine.py index e34916ff..475e9966 100644 --- a/roboco/services/self_heal_engine.py +++ b/roboco/services/self_heal_engine.py @@ -25,14 +25,25 @@ from __future__ import annotations import hashlib from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast from roboco.config import settings +from roboco.foundation import identity as _foundation +from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team from roboco.services.base import BaseService from roboco.services.notification import NotificationService +from roboco.services.project import get_project_service +from roboco.services.task import ( + SELF_HEAL_SOURCE, + TaskCreateRequest, + extract_self_heal_fingerprint, + get_task_service, +) from roboco.services.telemetry import get_ci_telemetry_source if TYPE_CHECKING: + from uuid import UUID + from sqlalchemy.ext.asyncio import AsyncSession from roboco.services.telemetry import TelemetrySource @@ -86,9 +97,13 @@ class SelfHealEngine(BaseService): return observations async def run_cycle(self) -> list[RegressionObservation]: - """Assess and notify the CEO. No-op unless self-healing is enabled. + """Assess, notify the CEO, and (if originate is on) open fix tasks. - Detect + notify only; never starts, merges, or deploys anything. + No-op unless ``self_heal_enabled``. It always NOTIFIES on a regression; + when ``self_heal_originate_enabled`` it also opens a PENDING fix task per + new regression and STOPS. It never starts, approves, merges, or deploys. + Writes (any opened task) are flushed here; the caller (the orchestrator + loop) owns the commit. """ if not settings.self_heal_enabled: return [] @@ -103,8 +118,90 @@ class SelfHealEngine(BaseService): await notifier.send_ack_notification( from_agent="system", to_agent="ceo", body=body ) + if settings.self_heal_originate_enabled: + await self._originate(observations) return observations + async def _originate(self, observations: list[RegressionObservation]) -> int: + """Open a PENDING fix task per NEW regression, then STOP. Returns count. + + Bounded + deduped: skips a regression that already has an open self-heal + task (by fingerprint), honors the per-cycle and rolling open-task caps, + and resolves the repo to RoboCo's own project. Each task is created + PENDING + UNASSIGNED + ``confirmed_by_human=False`` so it sits inert until + the CEO Approve-&-Starts it — origination is the loop's last act. It NEVER + calls start / approve / merge / deploy. Flushes; the caller commits. + """ + task_svc = get_task_service(self.session) + project_svc = get_project_service(self.session) + open_tasks = await task_svc.list_open_self_heal_tasks() + open_fps: set[str] = set() + for existing in open_tasks: + fp = extract_self_heal_fingerprint(existing.quick_context) + if fp: + open_fps.add(fp) + open_count = len(open_tasks) + created = 0 + for obs in observations: + if created >= settings.self_heal_max_per_cycle: + break + if open_count >= settings.self_heal_max_open_tasks: + self.log.info( + "self-heal open-task cap reached; not originating", + cap=settings.self_heal_max_open_tasks, + ) + break + if obs.fingerprint in open_fps: + continue + project = await project_svc.get_by_slug(obs.repo_hint) + if project is None or project.id is None: + self.log.warning( + "self-heal could not resolve project; notify-only", + repo=obs.repo_hint, + ) + continue + task = await task_svc.create( + TaskCreateRequest( + title=f"Self-heal: fix the CI regression on {obs.repo_hint}", + description=( + f"RoboCo's own CI regressed.\n\n{obs.detail}\n\n" + f"Evidence: {obs.raw_ref}\n\n" + "Investigate and fix the regression at its root so CI " + "returns to green. This task was opened automatically by " + "the self-heal loop and is PENDING your Approve-&-Start; " + "nothing runs until you approve it." + ), + acceptance_criteria=[ + f"CI on {obs.repo_hint}'s default branch is green again", + "The cause of the failing run is fixed at its root, not " + "masked or skipped", + ], + team=Team.MAIN_PM, + created_by=_foundation.AGENTS["system"].uuid, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + project_id=cast("UUID", project.id), + status=TaskStatus.PENDING, + source=SELF_HEAL_SOURCE, + confirmed_by_human=False, + ) + ) + # Carry the fingerprint so a later cycle sees this regression already + # has an open fix task (parsed by extract_self_heal_fingerprint). + task.quick_context = f"self_heal_fp={obs.fingerprint}" + await self.session.flush() + open_fps.add(obs.fingerprint) + open_count += 1 + created += 1 + self.log.info( + "self-heal fix task opened (PENDING; awaiting CEO)", + task_id=str(task.id), + repo=obs.repo_hint, + fingerprint=obs.fingerprint, + ) + return created + def get_self_heal_engine( session: AsyncSession, source: TelemetrySource | None = None diff --git a/roboco/services/task.py b/roboco/services/task.py index 05f34cdf..c0505d95 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -372,6 +372,26 @@ def extract_required_cells(quick_context: str | None) -> list[str]: PR_REVIEW_SOURCES = ("external_pr", "internal_pr") +# Source tag for a self-healing fix task: a PENDING task the self-heal loop opens +# when RoboCo's own CI regresses. It rides the normal lifecycle once the CEO +# Approve-&-Starts it; the loop itself never starts/approves/merges it. +SELF_HEAL_SOURCE = "self_heal" + +_SELF_HEAL_FP_PREFIX = "self_heal_fp=" + + +def extract_self_heal_fingerprint(quick_context: str | None) -> str | None: + """The ``self_heal_fp=`` marker from quick_context, or None. + + The per-signal dedupe key carried on a self-heal task, so the loop can tell + a regression already has an open fix task without a schema change. + """ + for token in (quick_context or "").split(): + if token.startswith(_SELF_HEAL_FP_PREFIX): + return token[len(_SELF_HEAL_FP_PREFIX) :] or None + return None + + _SUPERSEDE_MARKER_PREFIX = "external_pr_supersede" @@ -789,6 +809,21 @@ class TaskService(BaseService): ) return result.first() is not None + async def list_open_self_heal_tasks(self) -> list[TaskTable]: + """Non-terminal self-heal fix tasks — the dedupe + open-cap basis. + + A self-heal task is "open" until it reaches a terminal state. While one + exists for a regression's fingerprint the loop must not originate a + second, and the rolling open-task cap counts these. + """ + result = await self.session.execute( + select(TaskTable).where( + TaskTable.source == SELF_HEAL_SOURCE, + TaskTable.status.notin_([TaskStatus.COMPLETED, TaskStatus.CANCELLED]), + ) + ) + return list(result.scalars().all()) + async def list_external_pr_reviews_awaiting_decision(self) -> list[TaskTable]: """Completed external-PR reviews still awaiting the CEO's decision. diff --git a/tests/unit/services/test_self_heal_originate_db.py b/tests/unit/services/test_self_heal_originate_db.py new file mode 100644 index 00000000..225ec869 --- /dev/null +++ b/tests/unit/services/test_self_heal_originate_db.py @@ -0,0 +1,210 @@ +"""Self-heal task origination against a real Postgres DB. + +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 + UNASSIGNED + ``confirmed_by_human=False`` so +it sits inert until the CEO Approve-&-Starts it. Crucially it NEVER calls +start / approve / merge / deploy — asserted here. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock + +import pytest +from roboco.config import settings as cfg +from roboco.db.tables import AgentTable, ProjectTable +from roboco.foundation import identity as _foundation +from roboco.models.base import AgentRole, AgentStatus, TaskStatus, Team +from roboco.services.notification import NotificationService +from roboco.services.self_heal_engine import SelfHealEngine +from roboco.services.task import TaskService, get_task_service +from roboco.services.telemetry import TelemetrySample + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession + +SYSTEM_UUID = _foundation.AGENTS["system"].uuid +SLUG = "roboco" +ONE = 1 + + +class _FakeSource: + def __init__(self, samples: list[TelemetrySample]) -> None: + self._samples = samples + + async def fetch(self) -> list[TelemetrySample]: + return list(self._samples) + + +def _breach(signal: str) -> TelemetrySample: + return TelemetrySample( + signal_name=signal, + value=1.0, + threshold=1.0, + window="latest_completed_run", + repo_hint=SLUG, + observed_at="2026-06-17T00:00:00Z", + raw_ref="https://github.com/x/roboco/actions/runs/1", + detail=f"{signal} concluded 'failure'", + ) + + +async def _seed_project(session: AsyncSession, slug: str = SLUG) -> None: + """Seed the system agent (FK target for created_by) + RoboCo's own project.""" + session.add( + AgentTable( + id=SYSTEM_UUID, + name="System", + slug=f"system-{slug}", + role=AgentRole.SYSTEM, + team=None, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="system", + capabilities=[], + permissions={}, + metrics={}, + ) + ) + await session.flush() + session.add( + ProjectTable( + name="RoboCo", + slug=slug, + git_url="https://github.com/x/roboco.git", + default_branch="master", + protected_branches=["master"], + assigned_cell=Team.BACKEND, + created_by=SYSTEM_UUID, + is_active=True, + ) + ) + await session.flush() + + +def _enable(monkeypatch: pytest.MonkeyPatch, **overrides: object) -> None: + monkeypatch.setattr(cfg, "self_heal_enabled", True) + monkeypatch.setattr(cfg, "self_heal_originate_enabled", True) + monkeypatch.setattr(cfg, "self_heal_max_open_tasks", 5) + monkeypatch.setattr(cfg, "self_heal_max_per_cycle", 5) + for key, value in overrides.items(): + monkeypatch.setattr(cfg, key, value) + # Keep notification a no-op (its own session/IO is out of scope here). + monkeypatch.setattr(NotificationService, "send_ack_notification", AsyncMock()) + + +@pytest.mark.asyncio +async def test_disabled_originate_creates_no_task( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed_project(db_session) + _enable(monkeypatch, self_heal_originate_enabled=False) + engine = SelfHealEngine(db_session, source=_FakeSource([_breach("ci:roboco")])) + obs = await engine.run_cycle() + assert len(obs) == ONE # detected + notified + assert await get_task_service(db_session).list_open_self_heal_tasks() == [] + + +@pytest.mark.asyncio +async def test_originate_creates_pending_unassigned_task( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed_project(db_session) + _enable(monkeypatch) + engine = SelfHealEngine(db_session, source=_FakeSource([_breach("ci:roboco")])) + await engine.run_cycle() + + open_tasks = await get_task_service(db_session).list_open_self_heal_tasks() + assert len(open_tasks) == ONE + task = open_tasks[0] + assert task.status == TaskStatus.PENDING + assert task.assigned_to is None # inert until the CEO Approve-&-Starts it + 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) + assert "self_heal_fp=" in (task.quick_context or "") + + +@pytest.mark.asyncio +async def test_dedupe_no_second_task_same_fingerprint( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed_project(db_session) + _enable(monkeypatch) + src = _FakeSource([_breach("ci:roboco")]) + await SelfHealEngine(db_session, source=src).run_cycle() + await SelfHealEngine(db_session, source=src).run_cycle() # same fingerprint + open_tasks = await get_task_service(db_session).list_open_self_heal_tasks() + assert len(open_tasks) == ONE + + +@pytest.mark.asyncio +async def test_per_cycle_cap_limits_origination( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed_project(db_session) + _enable(monkeypatch, self_heal_max_per_cycle=1) + # Two distinct regressions in one cycle, cap = 1 → only one task opens. + src = _FakeSource([_breach("ci:roboco:a"), _breach("ci:roboco:b")]) + await SelfHealEngine(db_session, source=src).run_cycle() + assert len(await get_task_service(db_session).list_open_self_heal_tasks()) == ONE + + +@pytest.mark.asyncio +async def test_open_task_cap_blocks_further_origination( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed_project(db_session) + _enable(monkeypatch, self_heal_max_open_tasks=1) + await SelfHealEngine( + db_session, source=_FakeSource([_breach("ci:roboco:a")]) + ).run_cycle() + # A different regression next cycle, but one task is already open → blocked. + await SelfHealEngine( + db_session, source=_FakeSource([_breach("ci:roboco:b")]) + ).run_cycle() + assert len(await get_task_service(db_session).list_open_self_heal_tasks()) == ONE + + +@pytest.mark.asyncio +async def test_unresolved_project_is_notify_only( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed_project(db_session) + _enable(monkeypatch) + ghost = TelemetrySample( + signal_name="ci:ghost", + value=1.0, + threshold=1.0, + window="latest_completed_run", + repo_hint="not-a-registered-project", + observed_at="", + raw_ref="", + ) + # No crash, no task — the regression can't be tied to a repo. + await SelfHealEngine(db_session, source=_FakeSource([ghost])).run_cycle() + assert await get_task_service(db_session).list_open_self_heal_tasks() == [] + + +@pytest.mark.asyncio +async def test_loop_never_starts_or_approves( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed_project(db_session) + _enable(monkeypatch) + approve = AsyncMock() + ceo_approve = AsyncMock() + monkeypatch.setattr(TaskService, "approve_and_start", approve) + monkeypatch.setattr(TaskService, "ceo_approve", ceo_approve) + await SelfHealEngine( + db_session, source=_FakeSource([_breach("ci:roboco")]) + ).run_cycle() + + approve.assert_not_awaited() + ceo_approve.assert_not_awaited() + 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