feat(self-heal): CI telemetry source for RoboCo's own repo (dormant)

First slice of the production self-healing loop: a read-only telemetry source
that watches RoboCo's OWN repo CI and normalizes the latest GitHub Actions run
conclusion into breach / no-breach samples for the regression detector. It
targets only the single project named by self_heal_project_slug — RoboCo
healing itself, never other/client repos; the org's repo-agnostic delivery flow
is untouched.

- config: self_heal_enabled / self_heal_project_slug / self_heal_originate_enabled
  plus interval and open-task / per-cycle caps, all default-off
- GitService.get_latest_ci_conclusion: per-project Actions-run lookup (graceful
  None on missing token / no runs / error; never raises into the loop)
- TelemetrySample + TelemetrySource contract + GitHubCITelemetrySource
- 5 unit tests

roboco/config.py

@@ -352,6 +352,61 @@ class Settings(BaseSettings):
         ),
     )
 
+    # ==========================================================================
+    # Production self-healing ("engine 4") — DORMANT by default
+    # ==========================================================================
+    # RoboCo heals ITSELF. A closed loop that watches RoboCo's OWN repo CI (the
+    # single project named by self_heal_project_slug — NOT other/client repos),
+    # detects a regression (a failing CI run on its default branch), notifies the
+    # CEO, and — behind a second opt-in — opens a PENDING fix task into RoboCo's
+    # own delivery lifecycle and STOPS. It never starts, merges, or deploys; every
+    # downstream step stays a human decision. Default OFF: the loop never runs and
+    # no GitHub call is made.
+    self_heal_enabled: bool = Field(
+        default=False,
+        description=(
+            "Master switch for the self-healing loop (detect + notify the CEO). "
+            "OFF by default; when off the background loop does not run at all and "
+            "no CI telemetry is fetched."
+        ),
+    )
+    self_heal_project_slug: str = Field(
+        default="",
+        description=(
+            "The registered project that IS RoboCo itself — the self-heal loop "
+            "watches ONLY this repo's CI and opens fix tasks ONLY into it (RoboCo "
+            "healing itself, not other repos). Empty = no target; the loop no-ops "
+            "even when enabled."
+        ),
+    )
+    self_heal_originate_enabled: bool = Field(
+        default=False,
+        description=(
+            "Second opt-in: when on (and self_heal_enabled), a detected regression "
+            "also opens a PENDING fix task into the regressed project's lifecycle. "
+            "OFF by default — the loop is notify-only. The loop NEVER starts, "
+            "approves, merges, or deploys the task; it stops at PENDING for the CEO."
+        ),
+    )
+    self_heal_interval_seconds: int = Field(
+        default=1800,
+        ge=60,
+        description="Seconds between self-healing telemetry assessment passes.",
+    )
+    self_heal_max_open_tasks: int = Field(
+        default=3,
+        ge=1,
+        description=(
+            "Rolling cap on concurrently-open self-heal tasks across all repos; "
+            "the loop originates nothing more while this many are still open."
+        ),
+    )
+    self_heal_max_per_cycle: int = Field(
+        default=1,
+        ge=1,
+        description="Max self-heal fix tasks the loop may originate in one cycle.",
+    )
+
     # ==========================================================================
     # Workspaces (Multi-Agent Git)
     # ==========================================================================

roboco/services/git.py

@@ -1641,6 +1641,82 @@ class GitService(BaseService):
             "author_association": pr.get("author_association"),
         }
 
+    async def get_latest_ci_conclusion(
+        self, project_slug: str
+    ) -> dict[str, Any] | None:
+        """Latest completed CI (GitHub Actions) run on a project's default branch.
+
+        The inbound telemetry signal for self-healing: the most recent COMPLETED
+        workflow run on the project's default branch, normalized to
+        ``conclusion`` (``success`` / ``failure`` / ``timed_out`` / ...),
+        ``head_sha``, ``run_url``, ``run_name``, ``branch`` and ``completed_at``.
+        Repo-agnostic — resolves owner/repo and the git token PER PROJECT, so it
+        works on any registered repo, never a hardcoded one. Returns ``None`` on a
+        missing token, unparseable remote, GitHub error, or a repo with no Actions
+        runs (a repo that doesn't use GitHub Actions yields no signal, not a false
+        one). It never raises into the poll loop.
+        """
+        project = await get_project_service(self.session).get_by_slug(project_slug)
+        if project is None or not project.git_url:
+            return None
+        try:
+            owner, repo = self._parse_git_url(project.git_url)
+        except GitError:
+            return None
+        git_token = await self._token_for_project(project_slug)
+        if not git_token:
+            return None
+        branch = project.default_branch or "main"
+        run = await self._fetch_latest_ci_run(
+            project_slug, owner, repo, branch, git_token
+        )
+        if run is None:
+            return None
+        return {
+            "conclusion": run.get("conclusion"),
+            "head_sha": run.get("head_sha"),
+            "run_url": run.get("html_url") or "",
+            "run_name": run.get("name") or "",
+            "branch": branch,
+            "completed_at": run.get("updated_at"),
+        }
+
+    async def _fetch_latest_ci_run(
+        self, project_slug: str, owner: str, repo: str, branch: str, git_token: str
+    ) -> dict[str, Any] | None:
+        """GET the most recent completed Actions run on ``branch``; None on error."""
+        api_base = settings.github_api_base_url.rstrip("/")
+        try:
+            async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
+                resp = await client.get(
+                    f"{api_base}/repos/{owner}/{repo}/actions/runs",
+                    headers={
+                        "Authorization": f"Bearer {git_token}",
+                        "Accept": "application/vnd.github+json",
+                        "X-GitHub-Api-Version": "2022-11-28",
+                    },
+                    params={"branch": branch, "status": "completed", "per_page": 1},
+                )
+        except httpx.HTTPError as e:
+            self.log.warning(
+                "get_latest_ci_conclusion request failed",
+                project=project_slug,
+                error=str(e),
+            )
+            return None
+        if not resp.is_success:
+            self.log.warning(
+                "get_latest_ci_conclusion non-2xx",
+                project=project_slug,
+                status=resp.status_code,
+            )
+            return None
+        data = resp.json()
+        runs = data.get("workflow_runs") if isinstance(data, dict) else None
+        if not runs:
+            return None
+        return cast("dict[str, Any]", runs[0])
+
     async def _post_pr(
         self,
         owner: str,

roboco/services/telemetry/__init__.py

@@ -0,0 +1,17 @@
+"""Telemetry ingestion for production self-healing ("engine 4")."""
+
+from roboco.services.telemetry.source import (
+    FAILURE_CONCLUSIONS,
+    GitHubCITelemetrySource,
+    TelemetrySample,
+    TelemetrySource,
+    get_ci_telemetry_source,
+)
+
+__all__ = [
+    "FAILURE_CONCLUSIONS",
+    "GitHubCITelemetrySource",
+    "TelemetrySample",
+    "TelemetrySource",
+    "get_ci_telemetry_source",
+]

roboco/services/telemetry/source.py

@@ -0,0 +1,106 @@
+"""Telemetry ingestion for production self-healing ("engine 4").
+
+RoboCo heals ITSELF: this reads a health signal for RoboCo's OWN repo — the
+single project named by ``settings.self_heal_project_slug`` — and normalizes it
+into ``TelemetrySample``s the regression detector can assess. The sample
+contract is the only thing the detector depends on, so the source is swappable:
+a GitHub Actions CI source today, another CI/APM source later, with no change to
+the engine.
+
+Read-only and repo-singular by design — it watches RoboCo's own project, never
+other/client repos (the agent org's general, repo-agnostic delivery work is a
+separate concern). ``fetch`` returns no samples (and never raises) when self-heal
+has no target or the signal is unavailable.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+from roboco.config import settings
+from roboco.services.git import GitService
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+# GitHub Actions run conclusions that count as a regression signal. ``cancelled``
+# / ``neutral`` / ``skipped`` / ``action_required`` are deliberately excluded —
+# they are not a failing build.
+FAILURE_CONCLUSIONS = frozenset({"failure", "timed_out", "startup_failure"})
+
+
+@dataclass(frozen=True)
+class TelemetrySample:
+    """One normalized health reading for RoboCo's own repo.
+
+    ``value >= threshold`` is a breach (a regression). For CI: ``value`` is 1.0
+    when the latest completed run failed and 0.0 when it passed; ``threshold`` is
+    1.0. The string fields carry enough to describe and link the signal.
+    """
+
+    signal_name: str
+    value: float
+    threshold: float
+    window: str
+    repo_hint: str  # the self-heal project slug (RoboCo's own repo)
+    observed_at: str
+    raw_ref: str  # a link to the underlying evidence (e.g. the CI run URL)
+    detail: str = ""
+
+    @property
+    def is_breach(self) -> bool:
+        """True when the reading breaches its threshold (a regression)."""
+        return self.value >= self.threshold
+
+
+@runtime_checkable
+class TelemetrySource(Protocol):
+    """Pull-based, read-only health source. ``fetch`` never raises into the loop."""
+
+    async def fetch(self) -> list[TelemetrySample]: ...
+
+
+class GitHubCITelemetrySource:
+    """CI health for RoboCo's own repo, from GitHub Actions run conclusions.
+
+    Watches ONLY ``settings.self_heal_project_slug`` (RoboCo healing itself): the
+    latest completed run on that project's default branch. A failing run yields a
+    breaching sample, a passing run a non-breaching one; no target / no run / any
+    error yields no samples.
+    """
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def fetch(self) -> list[TelemetrySample]:
+        slug = settings.self_heal_project_slug.strip()
+        if not slug:
+            return []
+        ci = await GitService(self.session).get_latest_ci_conclusion(slug)
+        if ci is None:
+            return []
+        conclusion = (ci.get("conclusion") or "").lower()
+        failed = conclusion in FAILURE_CONCLUSIONS
+        run_name = ci.get("run_name") or ""
+        detail = f"CI on {slug}@{ci.get('branch')} concluded '{conclusion}'"
+        if run_name:
+            detail += f" ({run_name})"
+        return [
+            TelemetrySample(
+                signal_name=f"ci_conclusion:{slug}",
+                value=1.0 if failed else 0.0,
+                threshold=1.0,
+                window="latest_completed_run",
+                repo_hint=slug,
+                observed_at=str(ci.get("completed_at") or ""),
+                raw_ref=str(ci.get("run_url") or ""),
+                detail=detail,
+            )
+        ]
+
+
+def get_ci_telemetry_source(session: AsyncSession) -> GitHubCITelemetrySource:
+    """Construct the GitHub-CI telemetry source bound to ``session``."""
+    return GitHubCITelemetrySource(session)

tests/unit/services/test_telemetry_source.py

@@ -0,0 +1,81 @@
+"""Self-heal telemetry source — CI conclusion normalized to samples.
+
+The GitHub-CI source watches ONLY RoboCo's own project
+(``settings.self_heal_project_slug``): a failing run is a breaching sample, a
+passing run a non-breaching one, and no target / no run yields nothing. It is
+read-only and never raises.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from roboco.config import settings as cfg
+from roboco.services.git import GitService
+from roboco.services.telemetry import GitHubCITelemetrySource
+
+
+def _ci(conclusion: str) -> dict[str, str]:
+    return {
+        "conclusion": conclusion,
+        "head_sha": "abc123",
+        "run_url": "https://github.com/x/roboco/actions/runs/1",
+        "run_name": "CI",
+        "branch": "master",
+        "completed_at": "2026-06-17T00:00:00Z",
+    }
+
+
+@pytest.mark.asyncio
+async def test_no_target_yields_no_samples(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(cfg, "self_heal_project_slug", "")
+    assert await GitHubCITelemetrySource(MagicMock()).fetch() == []
+
+
+@pytest.mark.asyncio
+async def test_failing_ci_is_a_breach(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(cfg, "self_heal_project_slug", "roboco")
+    monkeypatch.setattr(
+        GitService, "get_latest_ci_conclusion", AsyncMock(return_value=_ci("failure"))
+    )
+    samples = await GitHubCITelemetrySource(MagicMock()).fetch()
+    assert len(samples) == 1
+    sample = samples[0]
+    assert sample.is_breach is True
+    assert sample.repo_hint == "roboco"
+    assert sample.signal_name == "ci_conclusion:roboco"
+    assert "failure" in sample.detail
+    assert sample.raw_ref.endswith("/runs/1")
+
+
+@pytest.mark.asyncio
+async def test_passing_ci_is_not_a_breach(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(cfg, "self_heal_project_slug", "roboco")
+    monkeypatch.setattr(
+        GitService, "get_latest_ci_conclusion", AsyncMock(return_value=_ci("success"))
+    )
+    samples = await GitHubCITelemetrySource(MagicMock()).fetch()
+    assert len(samples) == 1
+    assert samples[0].is_breach is False
+
+
+@pytest.mark.asyncio
+async def test_cancelled_run_is_not_a_breach(monkeypatch: pytest.MonkeyPatch) -> None:
+    # A cancelled run is not a failing build — it must not count as a regression.
+    monkeypatch.setattr(cfg, "self_heal_project_slug", "roboco")
+    monkeypatch.setattr(
+        GitService, "get_latest_ci_conclusion", AsyncMock(return_value=_ci("cancelled"))
+    )
+    samples = await GitHubCITelemetrySource(MagicMock()).fetch()
+    assert len(samples) == 1
+    assert samples[0].is_breach is False
+
+
+@pytest.mark.asyncio
+async def test_no_run_yields_no_samples(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(cfg, "self_heal_project_slug", "roboco")
+    monkeypatch.setattr(
+        GitService, "get_latest_ci_conclusion", AsyncMock(return_value=None)
+    )
+    assert await GitHubCITelemetrySource(MagicMock()).fetch() == []