"""CiWatchEngine — originate a fix task per red opted-in repo, bounded + deduped. Mirrors the self-heal engine: opens a PENDING ci_watch task per red project, never merges/approves; dedupes per repo (git_url) so a still-red repo with an open task gets none; honours per-cycle + rolling caps; a None-signal project (no sample) yields no task. """ from __future__ import annotations from typing import TYPE_CHECKING from uuid import uuid4 import pytest from roboco.config import settings 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.ci_watch_engine import get_ci_watch_engine from roboco.services.task import CI_WATCH_SOURCE, get_task_service from roboco.services.telemetry.source import TelemetrySample if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession SYSTEM_UUID = _foundation.AGENTS["system"].uuid MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid class _FakeSource: def __init__(self, samples: list[TelemetrySample]) -> None: self._samples = samples async def fetch(self, _projects: list[object]) -> list[TelemetrySample]: return list(self._samples) def _breach(slug: str, *, failed: bool = True) -> TelemetrySample: 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="2026-06-25T00:00:00Z", raw_ref=f"https://github.com/x/{slug}/actions/runs/1", detail=f"CI on {slug}@master concluded 'failure'", ) async def _get_or_create_agent( db: AsyncSession, agent_id: object, role: AgentRole, slug: str ) -> None: if await db.get(AgentTable, agent_id) is None: db.add( AgentTable( id=agent_id, name=slug, slug=f"{slug}-{uuid4().hex[:8]}", role=role, team=None, status=AgentStatus.ACTIVE, model_config={}, system_prompt="x", capabilities=[], permissions={}, metrics={}, ) ) await db.flush() async def _seed_project( db: AsyncSession, slug: str, git_url: str, *, workflow: str | None = None ) -> ProjectTable: project = ProjectTable( id=uuid4(), name=slug, slug=slug, git_url=git_url, assigned_cell=Team.BACKEND, created_by=SYSTEM_UUID, ci_watch_enabled=True, ci_watch_workflow=workflow, ) db.add(project) await db.flush() return project @pytest.fixture(autouse=True) async def _enabled(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(settings, "ci_watch_enabled", True) monkeypatch.setattr(settings, "ci_watch_max_per_cycle", 5) monkeypatch.setattr(settings, "ci_watch_max_open_tasks", 5) await _get_or_create_agent(db_session, SYSTEM_UUID, AgentRole.SYSTEM, "system") await _get_or_create_agent(db_session, MAIN_PM_UUID, AgentRole.MAIN_PM, "main-pm") @pytest.mark.asyncio async def test_red_project_opens_one_fix_task(db_session: AsyncSession) -> None: proj = await _seed_project(db_session, "red-a", "https://github.com/x/a.git") engine = get_ci_watch_engine(db_session, source=_FakeSource([_breach("red-a")])) created = await engine.run_cycle([proj]) assert len(created) == 1 task = created[0] assert task.project_id == proj.id assert task.source == CI_WATCH_SOURCE assert task.confirmed_by_human is True assert task.status == TaskStatus.PENDING # opened, never merged/approved @pytest.mark.asyncio async def test_still_red_with_open_task_opens_nothing( db_session: AsyncSession, ) -> None: proj = await _seed_project(db_session, "red-b", "https://github.com/x/b.git") src = _FakeSource([_breach("red-b")]) engine = get_ci_watch_engine(db_session, source=src) first = await engine.run_cycle([proj]) assert len(first) == 1 # Second cycle, same repo still red → deduped (one open task per git_url) second = await engine.run_cycle([proj]) assert second == [] @pytest.mark.asyncio async def test_per_cycle_cap( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(settings, "ci_watch_max_per_cycle", 1) p1 = await _seed_project(db_session, "red-c", "https://github.com/x/c.git") p2 = await _seed_project(db_session, "red-d", "https://github.com/x/d.git") src = _FakeSource([_breach("red-c"), _breach("red-d")]) created = await get_ci_watch_engine(db_session, source=src).run_cycle([p1, p2]) assert len(created) == 1 # capped at one per cycle @pytest.mark.asyncio async def test_green_or_no_signal_opens_nothing(db_session: AsyncSession) -> None: proj = await _seed_project(db_session, "quiet", "https://github.com/x/q.git") # No sample at all (None signal) — engine must not originate. none_engine = get_ci_watch_engine(db_session, source=_FakeSource([])) assert await none_engine.run_cycle([proj]) == [] # A green (non-breaching) sample — also no task. green_engine = get_ci_watch_engine( db_session, source=_FakeSource([_breach("quiet", failed=False)]) ) assert await green_engine.run_cycle([proj]) == [] assert await get_task_service(db_session).list_open_ci_watch_tasks() == [] @pytest.mark.asyncio async def test_disabled_is_noop( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(settings, "ci_watch_enabled", False) proj = await _seed_project(db_session, "red-e", "https://github.com/x/e.git") src = _FakeSource([_breach("red-e")]) assert await get_ci_watch_engine(db_session, source=src).run_cycle([proj]) == [] # --------------------------------------------------------------------------- # #44: dedupe by (git_url, workflow) — a multi-workflow monorepo with two red # workflows gets a fix task per workflow, not one collapsed task per repo # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_two_red_workflows_of_one_monorepo_both_open( db_session: AsyncSession, ) -> None: """#44: same git_url, DIFFERENT workflows, both red → two fix tasks.""" git = "https://github.com/x/mono.git" p_a = await _seed_project(db_session, "mono-wf-a", git, workflow="wf-a.yml") p_b = await _seed_project(db_session, "mono-wf-b", git, workflow="wf-b.yml") src = _FakeSource([_breach("mono-wf-a"), _breach("mono-wf-b")]) projects = [p_a, p_b] created = await get_ci_watch_engine(db_session, source=src).run_cycle(projects) assert len(created) == len(projects) assert {c.project_id for c in created} == {p_a.id, p_b.id} @pytest.mark.asyncio async def test_same_git_url_same_workflow_still_deduped( db_session: AsyncSession, ) -> None: """Regression guard: two cell-projects on one repo, SAME workflow → one task (the (git_url, workflow) key still collapses a same-workflow monorepo).""" git = "https://github.com/x/mono2.git" p1 = await _seed_project(db_session, "mono2-a", git, workflow="wf-a.yml") p2 = await _seed_project(db_session, "mono2-b", git, workflow="wf-a.yml") src = _FakeSource([_breach("mono2-a"), _breach("mono2-b")]) created = await get_ci_watch_engine(db_session, source=src).run_cycle([p1, p2]) assert len(created) == 1 @pytest.mark.asyncio async def test_default_workflow_null_rows_deduped( db_session: AsyncSession, ) -> None: """Two NULL-workflow project rows on one repo both use the default workflow → deduped to one task (coalesce treats NULL as the default workflow).""" git = "https://github.com/x/mono3.git" p1 = await _seed_project(db_session, "mono3-a", git) # ci_watch_workflow=None p2 = await _seed_project(db_session, "mono3-b", git) src = _FakeSource([_breach("mono3-a"), _breach("mono3-b")]) created = await get_ci_watch_engine(db_session, source=src).run_cycle([p1, p2]) assert len(created) == 1 # --------------------------------------------------------------------------- # #148: an empty-string ci_watch_workflow (saved via panel/API, not NULL) must # collapse to the default workflow for dedupe — SQL COALESCE alone treats '' as # a real value, so the DB query diverged from the engine's Python truthiness and # opened a duplicate fix task every red cycle. # #1267: git_url accidentals (case / .git suffix / trailing slash) must collapse # to one repo for dedupe, mirroring the orchestrator's poll-set repo_key. # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_empty_string_workflow_deduped(db_session: AsyncSession) -> None: """#148: a ''-workflow project and a NULL-workflow project on one repo both use the default workflow -> deduped to one task (NULLIF treats '' as NULL).""" git = "https://github.com/x/empty-wf.git" p1 = await _seed_project(db_session, "empty-wf-a", git, workflow="") p2 = await _seed_project(db_session, "empty-wf-b", git) # ci_watch_workflow=None src = _FakeSource([_breach("empty-wf-a"), _breach("empty-wf-b")]) created = await get_ci_watch_engine(db_session, source=src).run_cycle([p1, p2]) assert len(created) == 1 @pytest.mark.asyncio async def test_git_url_accidentals_deduped(db_session: AsyncSession) -> None: """#1267: two projects whose git_url differs only by a ``.git`` suffix (and trailing slash) are the same repo -> one fix task, not two.""" p1 = await _seed_project( db_session, "acc-a", "https://github.com/x/acc.git", workflow="wf.yml" ) p2 = await _seed_project( db_session, "acc-b", "https://github.com/x/acc/", workflow="wf.yml" ) src = _FakeSource([_breach("acc-a"), _breach("acc-b")]) created = await get_ci_watch_engine(db_session, source=src).run_cycle([p1, p2]) assert len(created) == 1