feat(self-heal): scope CI signal to a workflow + warn on missing target

Two hardening fixes from the gap review:
- Optional self_heal_ci_workflow scopes the CI signal to one workflow file
  (the workflow-scoped Actions endpoint). Without it, "latest completed run
  across all workflows" could miss a red CI run masked by a later passing
  workflow, or false-trigger on a non-CI workflow — unreliable on a
  multi-workflow repo.
- The loop logs a warning when self-heal is armed but self_heal_project_slug
  is unset, so a misconfiguration isn't mistaken for "all green".

Tests cover the workflow-scoped endpoint.
This commit is contained in:
Renn F
2026-06-17 21:58:48 +02:00
parent 7ef7d8414e
commit 33fa21d00a
5 changed files with 61 additions and 10 deletions
+10
View File
@@ -379,6 +379,16 @@ class Settings(BaseSettings):
"even when enabled."
),
)
self_heal_ci_workflow: str = Field(
default="",
description=(
"Optional GitHub Actions workflow file name (e.g. 'ci.yml') to scope "
"the CI signal to. Empty = use the latest completed run across ALL "
"workflows on the default branch, which is imprecise when the repo "
"has several workflows; set this to the real CI workflow for a "
"reliable signal."
),
)
self_heal_originate_enabled: bool = Field(
default=False,
description=(
+9
View File
@@ -4682,6 +4682,15 @@ Start by:
from roboco.db import get_db_context
from roboco.services.self_heal_engine import get_self_heal_engine
# Operability: self-heal is armed but has no target → it will silently
# no-op every cycle. Say so once at startup so a misconfiguration (unset
# or wrong ROBOCO_SELF_HEAL_PROJECT_SLUG) isn't mistaken for "all green".
if not settings.self_heal_project_slug.strip():
logger.warning(
"self-heal enabled but self_heal_project_slug is unset — the loop "
"will not detect anything until the target project is configured"
)
interval = settings.self_heal_interval_seconds
while self._running:
try:
+23 -9
View File
@@ -1642,7 +1642,7 @@ class GitService(BaseService):
}
async def get_latest_ci_conclusion(
self, project_slug: str
self, project_slug: str, *, workflow: str | None = None
) -> dict[str, Any] | None:
"""Latest completed CI (GitHub Actions) run on a project's default branch.
@@ -1650,10 +1650,12 @@ class GitService(BaseService):
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
Resolves owner/repo and the git token PER PROJECT. ``workflow`` (a
workflow file name like ``ci.yml``) scopes the signal to one workflow —
without it the latest run across ALL workflows is used, which is
imprecise on a multi-workflow repo. Returns ``None`` on a missing token,
unparseable remote, GitHub error, or a repo with no matching 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)
@@ -1668,7 +1670,7 @@ class GitService(BaseService):
return None
branch = project.default_branch or "main"
run = await self._fetch_latest_ci_run(
project_slug, owner, repo, branch, git_token
project_slug, (owner, repo), branch, git_token, workflow
)
if run is None:
return None
@@ -1682,14 +1684,26 @@ class GitService(BaseService):
}
async def _fetch_latest_ci_run(
self, project_slug: str, owner: str, repo: str, branch: str, git_token: str
self,
project_slug: str,
owner_repo: tuple[str, str],
branch: str,
git_token: str,
workflow: str | None = None,
) -> dict[str, Any] | None:
"""GET the most recent completed Actions run on ``branch``; None on error."""
"""GET the most recent completed Actions run on ``branch``; None on error.
Scopes to ``workflow`` (a workflow file name) when given — the precise
signal — otherwise looks across all workflows.
"""
owner, repo = owner_repo
api_base = settings.github_api_base_url.rstrip("/")
base = f"{api_base}/repos/{owner}/{repo}/actions"
url = f"{base}/workflows/{workflow}/runs" if workflow else f"{base}/runs"
try:
async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
resp = await client.get(
f"{api_base}/repos/{owner}/{repo}/actions/runs",
url,
headers={
"Authorization": f"Bearer {git_token}",
"Accept": "application/vnd.github+json",
+4 -1
View File
@@ -78,7 +78,10 @@ class GitHubCITelemetrySource:
slug = settings.self_heal_project_slug.strip()
if not slug:
return []
ci = await GitService(self.session).get_latest_ci_conclusion(slug)
workflow = settings.self_heal_ci_workflow.strip() or None
ci = await GitService(self.session).get_latest_ci_conclusion(
slug, workflow=workflow
)
if ci is None:
return []
conclusion = (ci.get("conclusion") or "").lower()
@@ -241,3 +241,18 @@ async def test_get_latest_ci_conclusion_none_when_no_runs() -> None:
patch("roboco.services.git.httpx.AsyncClient", return_value=client),
):
assert await svc.get_latest_ci_conclusion("roboco") is None
@pytest.mark.asyncio
async def test_get_latest_ci_conclusion_scopes_to_workflow() -> None:
# With a workflow file given, hit the workflow-scoped endpoint (precise signal).
svc = _service()
client = _client(_resp(200, json_payload={"workflow_runs": [_run("success")]}))
with (
_patch_project_ci(),
patch("roboco.services.git.httpx.AsyncClient", return_value=client),
):
out = await svc.get_latest_ci_conclusion("roboco", workflow="ci.yml")
assert out is not None
assert out["conclusion"] == "success"
assert client.get.await_args.args[0].endswith("/actions/workflows/ci.yml/runs")