Feat/autonomous maintenance (#264)

* feat(ci-watch): config flags

Default-off CI-watch config (mirrors self_heal_*): ci_watch_enabled,
ci_watch_default_workflow (ci.yml), ci_watch_interval_seconds (1800),
ci_watch_max_open_tasks (3), ci_watch_max_per_cycle (1). Registers
ci_watch_enabled in the panel FEATURE_FLAGS. 4 tests.

* feat(ci-watch): per-project ci_watch_enabled/workflow (migration 048)

Adds projects.ci_watch_enabled (bool NOT NULL default false) +
projects.ci_watch_workflow (varchar null) — the per-project opt-in for
multi-repo CI-watch. ProjectTable + Pydantic Project fields + migration 048
(off 047_ws_single_active). Real upgrade->downgrade->upgrade chain verified
against a throwaway Postgres; 2 ORM round-trip tests.

* feat(runtime): prune dangling agent images in the background sweeper

Every agent-image rebuild orphans the prior build's layers as an untagged
<none> image; across deploys these pile up (the operator hit ~80). The sweeper
now runs 'docker image prune -f --filter dangling=true' (dangling only — a
tagged image or one backing a running container is never dangling), throttled
to settings.image_prune_interval_seconds (default 6h) and gated by
image_prune_enabled (default on). Best-effort: any failure is logged, never
raised into the sweeper. Mirrors the transcript-retention prune. 4 tests.

* feat(ci-watch): source tag + open-task dedupe query

CI_WATCH_SOURCE='ci_watch' + TaskService.list_open_ci_watch_tasks(git_url=None):
non-terminal ci_watch tasks (the dedupe + open-cap basis), optionally scoped to
one repo by git_url — a monorepo registers several cell-projects on one git_url,
so dedupe keys on the repo, not the slug. 2 real-PG tests.

* feat(ci-watch): multi-project CI telemetry fan-out

MultiProjectCITelemetrySource.fetch(projects) reuses the hardened per-project
get_latest_ci_conclusion for each opted-in project (passing its ci_watch_workflow
or the configured default). Per-project isolation: a GitHub error or absent
signal yields NO sample (unknown, never read as green) and never aborts the
sweep; only a real conclusion yields a sample (fail→breach, pass→non-breach).
self-heal source untouched. 3 tests + self-heal regression green.

* feat(ci-watch): engine — fan-out, originate, dedupe, cap

CiWatchEngine.run_cycle(projects) mirrors SelfHealEngine: assess via
MultiProjectCITelemetrySource, open one PENDING ci_watch fix task per red repo
(team=main_pm, assigned_to=main-pm, confirmed_by_human=True so it dispatches
without an Approve-&-Start — the fe029fe3 lesson), never starts/approves/merges.
Dedupe per git_url (monorepo → one fix task per repo) + per-cycle/rolling caps.
Default-off; disabled → no-op. 5 real-PG tests (red→one task, dedupe, cap,
green/none→nothing, disabled).

* feat(ci-watch): orchestrator loop tick + watch-set loader

_ci_watch_loop (registered in start(), cancelled in stop(), separate from the
untouched self-heal loop): dormant unless ci_watch_enabled; each interval loads
the watch set (ci_watch_enabled projects, collapsed one-per-repo via the
existing _projects_one_per_repo) and runs CiWatchEngine.run_cycle, committing
opened tasks. _run_ci_watch_cycle extracted for testing; loud warning when
enabled-but-empty. confirmed_by_human=True on the originated task means it
dispatches without an Approve-&-Start (no stranding, the fe029fe3 lesson).
5 tests (disabled no-op, watch-set filter+one-per-repo, empty warn, engine run).

* docs(ci-watch): CHANGELOG + CLAUDE.md for multi-repo CI-watch

Document CI-watch (Added) in the CHANGELOG and the Self-Healing & Feature Flags
section of CLAUDE.md — it generalizes self-heal to opted-in projects, reuses the
hardened per-project CI lookup, never auto-merges, default-off. Adds the
ci_watch_enabled flag to the feature-flags enumeration.

* feat(dep-update): config flags

Default-off dep-update config (mirrors self_heal_*/ci_watch_*): dep_update_enabled,
dep_update_interval_seconds (604800 = weekly), dep_update_max_open_tasks (3),
dep_update_max_per_cycle (1). Registers dep_update_enabled in FEATURE_FLAGS. 4 tests.

* feat(dep-update): per-project dep_update_command/paths (migration 049)

Adds projects.dep_update_command (varchar null) + dep_update_paths (varchar[]
null) — the per-project opt-in for the dependency-update bot. ProjectTable +
Pydantic Project fields + migration 049 (off 048_ci_watch_project_cols). Real
upgrade->downgrade->upgrade chain verified on a throwaway Postgres; 2 ORM tests.

* feat(dep-update): source tag + open-task dedupe query

DEP_UPDATE_SOURCE='dep_update' + TaskService.list_open_dep_update_tasks(git_url=None):
non-terminal dep_update tasks (dedupe + open-cap basis), optionally scoped to one
repo by git_url (monorepo → one open dependency-update task per repo). 2 real-PG
tests.

* feat(dep-update): read-only lockfile-diff probe

WorkspaceService.dry_upgrade_changes_lockfile(project): clones the project's
read clone into a throwaway dir (--no-hardlinks, so the read clone is never
mutated), runs project.dep_update_command (no shell, shlex.split), and reports
whether any lockfile path (dep_update_paths or inferred uv.lock/pnpm-lock.yaml)
is dirty. Fail-safe: null/failing command → False (don't originate on a broken
probe), logged; throwaway always removed; never commits/pushes. 5 real-git tests.

* feat(dep-update): engine — detect, originate, dedupe, cap

DepUpdateEngine.run_cycle(projects) mirrors SelfHealEngine/CiWatchEngine: for
each opted-in project (dep_update_command set) with updates available (the
read-only probe), open one PENDING dep_update task (team=main_pm, assigned-to
main-pm, confirmed_by_human=True), never starts/approves/merges. Cheap checks
(command, per-git_url dedupe) before the expensive probe; per-cycle + rolling
caps. Default-off; disabled → no-op. 6 real-PG tests.

* feat(dep-update): weekly orchestrator loop tick

_dep_update_loop (registered in start(), cancelled in stop(), separate from the
self-heal + CI-watch loops): dormant unless dep_update_enabled; each interval
(default weekly) loads projects with a dep_update_command (one-per-repo) and runs
DepUpdateEngine.run_cycle, committing opened tasks. _run_dep_update_cycle
extracted for testing; loud warning when enabled-but-no-commands. Refactored
stop() to cancel background tasks via a shared _cancel_background_task loop
(keeps it under xenon B as the loop count grows). 4 loop tests.

Task 7 (anti-stranding dispatch guard) is satisfied by construction: no
dispatcher skip targets source='dep_update', and the engine sets
confirmed_by_human=True (the fe029fe3 lesson), asserted in the engine tests —
so the originated task dispatches via the assigned-PM path, never stranded.

* docs(dep-update): CHANGELOG + CLAUDE.md for the dependency-update bot

Document the dep-update bot (Added) in the CHANGELOG and the Self-Healing &
Feature Flags section of CLAUDE.md — read-only lockfile-diff probe, never
auto-merges, per-project opt-in via dep_update_command, default-off. Adds the
dep_update_enabled flag to the feature-flags enumeration.

* feat(ci-watch): route fix-task notification to the project's cell PM

On opening a fix task, CiWatchEngine notifies the red project's own cell PM
(resolved from project.assigned_cell via foundation AGENTS — e.g. BACKEND →
be-pm), not the CEO, once per project per cycle. Best-effort: a notification
failure never rolls back the origination. Adds _cell_pm_slug_for +
_notify_cell_pm. 1 real-PG test (asserts to_agent='be-pm', not 'ceo').

* feat(ci-watch,dep-update): expose per-project opt-ins in the project API

Add ci_watch_enabled/ci_watch_workflow + dep_update_command/dep_update_paths to
ProjectUpdate, ProjectUpdateRequest, the PATCH route mapping, ProjectResponse,
and project_to_response — so the panel edit-project dialog can read + set the
per-project autonomy opt-ins (the columns were unreachable through the API
before). Also threads the previously-dropped quality_command through the update
route. 1 real-PG update round-trip test.

* feat(ci-watch,dep-update): panel project-edit fields for the per-project opt-ins

Adds an 'Autonomous Maintenance' section to the edit-project dialog: a CI-watch
enable switch + workflow input, and a dependency-update command + lockfile-paths
input (comma-separated → list). Threads the four fields through the Project /
ProjectUpdate TS types and the mock-mode create fixture. The global on/off
toggles already live in Settings → Feature Flags; these are the per-project
opt-ins. panel tsc --noEmit + eslint green.

* docs(0.12): CI-watch + dep-update bot + image-prune across user docs + RAG

New docs/optional/autonomous-maintenance.md (mirrors self-heal.md) covering both
engines; optional/index rows; panel settings + projects-and-products notes for
the Feature Flags toggles + the edit-project Autonomous Maintenance fields;
resilience note for the dangling-image prune; env-reference + RAG config-reference
tables for all ROBOCO_CI_WATCH_* / ROBOCO_DEP_UPDATE_* / ROBOCO_IMAGE_PRUNE_*
vars; mkdocs nav entry. reflow-check green; prompts unchanged (operator-facing,
not agent-facing).

* chore(release): 0.12.0

Cut [Unreleased] -> [0.12.0] (CI-watch + dep-update bot + image-prune housekeeping
+ the post-0.11.1 run-hardening fixes). Bumps all 8 canonical version refs to
0.12.0 (pyproject / uv.lock roboco pkg / panel package.json / __init__ /
config.app_version + the README / deployment / agent-image-tag examples).

* fix(pr-review): repo-scope external-PR dedupe (no duplicate review on a monorepo)

external_review_task_exists keyed on (project_id, pr, head_sha), but a monorepo
registers several cell-projects on one git_url and the poll already collapses to
one canonical project per repo — so once a review task was re-pointed to a
sibling project, the next poll (checking the canonical project) no longer saw it
and opened a second review of the same PR (observed: PR #131 reviewed once on
guard-core-saas-frontend, once on -backend). Dedupe now spans every project
sharing the PR's repo (git_url); re-review on a new head SHA still works; a
genuinely different repo with the same PR number is independent. 3 real-PG tests.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/integration/services/test_ci_watch_engine.py

@@ -0,0 +1,159 @@
+"""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) -> 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,
+    )
+    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]) == []

tests/integration/services/test_ci_watch_notify.py

@@ -0,0 +1,104 @@
+"""CI-watch routes its fix-task notification to the project's cell PM.
+
+Not the CEO (a delivery/client repo's red CI is a cell concern) and once per
+project per cycle (the engine opens at most one task per repo per cycle).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock, MagicMock, patch
+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, Team
+from roboco.services.ci_watch_engine import get_ci_watch_engine
+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) -> TelemetrySample:
+    return TelemetrySample(
+        signal_name=f"ci_conclusion:{slug}",
+        value=1.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 _agent(db: AsyncSession, agent_id: Any, 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()
+
+
+@pytest.fixture(autouse=True)
+async def _setup(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 _agent(db_session, SYSTEM_UUID, AgentRole.SYSTEM, "system")
+    await _agent(db_session, MAIN_PM_UUID, AgentRole.MAIN_PM, "main-pm")
+
+
+@pytest.mark.asyncio
+async def test_notifies_backend_cell_pm_once(db_session: AsyncSession) -> None:
+    proj = ProjectTable(
+        id=uuid4(),
+        name="red",
+        slug="red",
+        git_url="https://github.com/x/a.git",
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+        ci_watch_enabled=True,
+    )
+    db_session.add(proj)
+    await db_session.flush()
+
+    notifier = MagicMock()
+    notifier.send_ack_notification = AsyncMock()
+    engine = get_ci_watch_engine(db_session, source=_FakeSource([_breach("red")]))
+    with patch(
+        "roboco.services.ci_watch_engine.NotificationService", return_value=notifier
+    ):
+        created = await engine.run_cycle([proj])
+
+    assert len(created) == 1
+    notifier.send_ack_notification.assert_awaited_once()
+    kwargs = notifier.send_ack_notification.await_args.kwargs
+    assert kwargs["to_agent"] == "be-pm"  # the BACKEND cell PM, not "ceo"
+    assert "red" in kwargs["body"]

tests/integration/services/test_ci_watch_source.py

@@ -0,0 +1,136 @@
+"""CI_WATCH_SOURCE + list_open_ci_watch_tasks — the dedupe / open-cap basis.
+
+Open ci_watch tasks count toward the cap and block a duplicate; terminal ones
+and other-source tasks do not. The git_url scoping keys dedupe on the repo (a
+monorepo registers several cell-projects on one git_url), so a watched repo
+gets at most one open fix task even across its cell-projects.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.foundation import identity as _foundation
+from roboco.models.base import (
+    AgentRole,
+    AgentStatus,
+    Complexity,
+    TaskNature,
+    TaskStatus,
+    TaskType,
+    Team,
+)
+from roboco.models.task import TaskCreateRequest
+from roboco.services.task import CI_WATCH_SOURCE, get_task_service
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+_TWO = 2
+
+
+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, git_url: str) -> ProjectTable:
+    project = ProjectTable(
+        id=uuid4(),
+        name="P",
+        slug=f"p-{uuid4().hex[:8]}",
+        git_url=git_url,
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+    )
+    db.add(project)
+    await db.flush()
+    return project
+
+
+async def _make_ci_watch_task(
+    db: AsyncSession,
+    project: ProjectTable,
+    *,
+    source: str = CI_WATCH_SOURCE,
+    terminal: bool = False,
+) -> None:
+    svc = get_task_service(db)
+    task = await svc.create(
+        TaskCreateRequest(
+            title="CI-watch fix",
+            description="Fix the CI regression on this project's default branch.",
+            acceptance_criteria=["CI is green again"],
+            team=Team.MAIN_PM,
+            assigned_to=MAIN_PM_UUID,
+            created_by=SYSTEM_UUID,
+            task_type=TaskType.CODE,
+            nature=TaskNature.TECHNICAL,
+            estimated_complexity=Complexity.MEDIUM,
+            project_id=project.id,
+            status=TaskStatus.PENDING,
+            source=source,
+            confirmed_by_human=True,
+        )
+    )
+    if terminal:
+        task.status = TaskStatus.COMPLETED
+        await db.flush()
+
+
+@pytest.fixture(autouse=True)
+async def _agents(db_session: AsyncSession) -> None:
+    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_lists_only_open_ci_watch_tasks(db_session: AsyncSession) -> None:
+    proj = await _seed_project(db_session, "https://github.com/x/a.git")
+    await _make_ci_watch_task(db_session, proj)  # open ci_watch
+    await _make_ci_watch_task(db_session, proj, terminal=True)  # terminal ci_watch
+    await _make_ci_watch_task(db_session, proj, source="manual")  # other source
+
+    open_tasks = await get_task_service(db_session).list_open_ci_watch_tasks()
+
+    assert len(open_tasks) == 1
+    assert open_tasks[0].source == CI_WATCH_SOURCE
+    assert open_tasks[0].status != TaskStatus.COMPLETED
+
+
+@pytest.mark.asyncio
+async def test_git_url_scoping_returns_only_that_repo(
+    db_session: AsyncSession,
+) -> None:
+    proj_a = await _seed_project(db_session, "https://github.com/x/a.git")
+    proj_b = await _seed_project(db_session, "https://github.com/x/b.git")
+    await _make_ci_watch_task(db_session, proj_a)
+    await _make_ci_watch_task(db_session, proj_b)
+
+    svc = get_task_service(db_session)
+    assert len(await svc.list_open_ci_watch_tasks()) == _TWO
+    scoped = await svc.list_open_ci_watch_tasks(git_url="https://github.com/x/a.git")
+    assert len(scoped) == 1
+    assert scoped[0].project_id == proj_a.id

tests/integration/services/test_dep_update_engine.py

@@ -0,0 +1,145 @@
+"""DepUpdateEngine — open an update-deps task per opted-in project with updates.
+
+Opens a PENDING dep_update task (never merges/approves) when the probe reports
+updates; skips projects with no command, no updates, or an already-open task for
+the same repo (git_url dedupe); honours per-cycle + rolling caps; dormant when
+disabled.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+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.dep_update_engine import get_dep_update_engine
+from roboco.services.task import DEP_UPDATE_SOURCE
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+
+
+class _FakeWorkspace:
+    def __init__(self, updates: bool = True) -> None:
+        self._updates = updates
+
+    async def dry_upgrade_changes_lockfile(self, _project: Any) -> bool:
+        return self._updates
+
+
+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, *, command: str | None = "uv lock -U"
+) -> ProjectTable:
+    project = ProjectTable(
+        id=uuid4(),
+        name=slug,
+        slug=slug,
+        git_url=git_url,
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+        dep_update_command=command,
+    )
+    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, "dep_update_enabled", True)
+    monkeypatch.setattr(settings, "dep_update_max_per_cycle", 5)
+    monkeypatch.setattr(settings, "dep_update_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_updates_available_opens_one_task(db_session: AsyncSession) -> None:
+    proj = await _seed_project(db_session, "dep-a", "https://github.com/x/a.git")
+    engine = get_dep_update_engine(db_session, workspace=_FakeWorkspace(updates=True))
+
+    created = await engine.run_cycle([proj])
+
+    assert len(created) == 1
+    task = created[0]
+    assert task.project_id == proj.id
+    assert task.source == DEP_UPDATE_SOURCE
+    assert task.confirmed_by_human is True
+    assert task.status == TaskStatus.PENDING
+
+
+@pytest.mark.asyncio
+async def test_no_command_skipped(db_session: AsyncSession) -> None:
+    proj = await _seed_project(
+        db_session, "dep-b", "https://github.com/x/b.git", command=None
+    )
+    engine = get_dep_update_engine(db_session, workspace=_FakeWorkspace(updates=True))
+    assert await engine.run_cycle([proj]) == []
+
+
+@pytest.mark.asyncio
+async def test_no_updates_skipped(db_session: AsyncSession) -> None:
+    proj = await _seed_project(db_session, "dep-c", "https://github.com/x/c.git")
+    engine = get_dep_update_engine(db_session, workspace=_FakeWorkspace(updates=False))
+    assert await engine.run_cycle([proj]) == []
+
+
+@pytest.mark.asyncio
+async def test_dedupe_same_repo(db_session: AsyncSession) -> None:
+    proj = await _seed_project(db_session, "dep-d", "https://github.com/x/d.git")
+    engine = get_dep_update_engine(db_session, workspace=_FakeWorkspace(updates=True))
+    first = await engine.run_cycle([proj])
+    assert len(first) == 1
+    second = await engine.run_cycle([proj])  # still updatable, but already open
+    assert second == []
+
+
+@pytest.mark.asyncio
+async def test_per_cycle_cap(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(settings, "dep_update_max_per_cycle", 1)
+    p1 = await _seed_project(db_session, "dep-e", "https://github.com/x/e.git")
+    p2 = await _seed_project(db_session, "dep-f", "https://github.com/x/f.git")
+    engine = get_dep_update_engine(db_session, workspace=_FakeWorkspace(updates=True))
+    created = await engine.run_cycle([p1, p2])
+    assert len(created) == 1
+
+
+@pytest.mark.asyncio
+async def test_disabled_is_noop(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(settings, "dep_update_enabled", False)
+    proj = await _seed_project(db_session, "dep-g", "https://github.com/x/g.git")
+    engine = get_dep_update_engine(db_session, workspace=_FakeWorkspace(updates=True))
+    assert await engine.run_cycle([proj]) == []

tests/integration/services/test_dep_update_probe.py

@@ -0,0 +1,96 @@
+"""dry_upgrade_changes_lockfile — the read-only lockfile-diff probe.
+
+Runs the project's dep_update_command in an isolated clone of the read clone and
+reports whether a lockfile path got dirty — without ever mutating the read clone
+or committing/pushing. Fail-safe: a null/failing command returns False.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from roboco.services.workspace import WorkspaceService
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _git(cwd: Path, *args: str) -> None:
+    subprocess.run(
+        ["git", *args], cwd=str(cwd), capture_output=True, text=True, check=True
+    )
+
+
+def _make_read_clone(tmp_path: Path) -> Path:
+    repo = tmp_path / "readclone"
+    repo.mkdir()
+    _git(repo, "init", "-q")
+    _git(repo, "config", "user.email", "t@example.com")
+    _git(repo, "config", "user.name", "t")
+    (repo / "uv.lock").write_text("version = 1\n")
+    _git(repo, "add", "-A")
+    _git(repo, "commit", "-q", "-m", "init")
+    return repo
+
+
+def _svc(read_clone: Path) -> WorkspaceService:
+    svc = WorkspaceService.__new__(WorkspaceService)
+    svc.ensure_read_clone = AsyncMock(return_value=read_clone)  # type: ignore[method-assign]
+    return svc
+
+
+def _project(command: str | None, paths: list[str] | None = None) -> MagicMock:
+    return MagicMock(slug="p", dep_update_command=command, dep_update_paths=paths)
+
+
+@pytest.mark.asyncio
+async def test_dirtying_a_lockfile_returns_true(tmp_path: Path) -> None:
+    read_clone = _make_read_clone(tmp_path)
+    svc = _svc(read_clone)
+    cmd = "python3 -c \"open('uv.lock','a').write('x')\""
+
+    assert await svc.dry_upgrade_changes_lockfile(_project(cmd)) is True
+
+    # The read clone itself is never mutated by the probe.
+    status = subprocess.run(
+        ["git", "status", "--porcelain"],
+        cwd=str(read_clone),
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    assert status.stdout.strip() == ""
+
+
+@pytest.mark.asyncio
+async def test_noop_command_returns_false(tmp_path: Path) -> None:
+    svc = _svc(_make_read_clone(tmp_path))
+    assert (
+        await svc.dry_upgrade_changes_lockfile(_project('python3 -c "pass"')) is False
+    )
+
+
+@pytest.mark.asyncio
+async def test_null_command_returns_false(tmp_path: Path) -> None:
+    svc = _svc(_make_read_clone(tmp_path))
+    assert await svc.dry_upgrade_changes_lockfile(_project(None)) is False
+
+
+@pytest.mark.asyncio
+async def test_failing_command_returns_false(tmp_path: Path) -> None:
+    svc = _svc(_make_read_clone(tmp_path))
+    cmd = 'python3 -c "import sys; sys.exit(1)"'
+    assert await svc.dry_upgrade_changes_lockfile(_project(cmd)) is False
+
+
+@pytest.mark.asyncio
+async def test_explicit_dep_update_paths_scope(tmp_path: Path) -> None:
+    read_clone = _make_read_clone(tmp_path)
+    svc = _svc(read_clone)
+    # Command dirties uv.lock, but we only watch a different lockfile → False.
+    cmd = "python3 -c \"open('uv.lock','a').write('x')\""
+    project = _project(cmd, paths=["pnpm-lock.yaml"])
+    assert await svc.dry_upgrade_changes_lockfile(project) is False

tests/integration/services/test_dep_update_source.py

@@ -0,0 +1,132 @@
+"""DEP_UPDATE_SOURCE + list_open_dep_update_tasks — the dedupe / open-cap basis.
+
+Open dep_update tasks count toward the cap and block a duplicate; terminal ones
+and other-source tasks do not. The git_url scoping keys dedupe on the repo so a
+monorepo gets at most one open dependency-update task across its cell-projects.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.foundation import identity as _foundation
+from roboco.models.base import (
+    AgentRole,
+    AgentStatus,
+    Complexity,
+    TaskNature,
+    TaskStatus,
+    TaskType,
+    Team,
+)
+from roboco.models.task import TaskCreateRequest
+from roboco.services.task import DEP_UPDATE_SOURCE, get_task_service
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+_TWO = 2
+
+
+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, git_url: str) -> ProjectTable:
+    project = ProjectTable(
+        id=uuid4(),
+        name="P",
+        slug=f"p-{uuid4().hex[:8]}",
+        git_url=git_url,
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+    )
+    db.add(project)
+    await db.flush()
+    return project
+
+
+async def _make_task(
+    db: AsyncSession,
+    project: ProjectTable,
+    *,
+    source: str = DEP_UPDATE_SOURCE,
+    terminal: bool = False,
+) -> None:
+    task = await get_task_service(db).create(
+        TaskCreateRequest(
+            title="Update dependencies",
+            description="Upgrade dependencies to latest compatible; gate must pass.",
+            acceptance_criteria=["lockfiles refreshed", "gate green"],
+            team=Team.MAIN_PM,
+            assigned_to=MAIN_PM_UUID,
+            created_by=SYSTEM_UUID,
+            task_type=TaskType.CODE,
+            nature=TaskNature.TECHNICAL,
+            estimated_complexity=Complexity.MEDIUM,
+            project_id=project.id,
+            status=TaskStatus.PENDING,
+            source=source,
+            confirmed_by_human=True,
+        )
+    )
+    if terminal:
+        task.status = TaskStatus.COMPLETED
+        await db.flush()
+
+
+@pytest.fixture(autouse=True)
+async def _agents(db_session: AsyncSession) -> None:
+    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_lists_only_open_dep_update_tasks(db_session: AsyncSession) -> None:
+    proj = await _seed_project(db_session, "https://github.com/x/a.git")
+    await _make_task(db_session, proj)
+    await _make_task(db_session, proj, terminal=True)
+    await _make_task(db_session, proj, source="manual")
+
+    open_tasks = await get_task_service(db_session).list_open_dep_update_tasks()
+
+    assert len(open_tasks) == 1
+    assert open_tasks[0].source == DEP_UPDATE_SOURCE
+    assert open_tasks[0].status != TaskStatus.COMPLETED
+
+
+@pytest.mark.asyncio
+async def test_git_url_scoping(db_session: AsyncSession) -> None:
+    proj_a = await _seed_project(db_session, "https://github.com/x/a.git")
+    proj_b = await _seed_project(db_session, "https://github.com/x/b.git")
+    await _make_task(db_session, proj_a)
+    await _make_task(db_session, proj_b)
+
+    svc = get_task_service(db_session)
+    assert len(await svc.list_open_dep_update_tasks()) == _TWO
+    scoped = await svc.list_open_dep_update_tasks(git_url="https://github.com/x/a.git")
+    assert len(scoped) == 1
+    assert scoped[0].project_id == proj_a.id

tests/integration/services/test_external_pr_repo_dedup.py

@@ -0,0 +1,112 @@
+"""External-PR review dedupe is repo-scoped (git_url), not project-scoped.
+
+A monorepo registers several cell-projects on one repo. Ingesting the same PR
+for a sibling project — or re-pointing an existing review task to a sibling —
+must NOT open a second review (the duplicate the operator hit: PR #131 reviewed
+once per cell-project after a re-point). Re-review on a new head SHA still works.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.foundation import identity as _foundation
+from roboco.models.base import AgentRole, AgentStatus, Team
+from roboco.services.task import get_task_service
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+_REPO = "https://github.com/rennf93/guard-core-app"
+_OTHER_REPO = "https://github.com/rennf93/other-app"
+
+
+def _pr(head_sha: str, number: int = 131) -> dict[str, Any]:
+    return {
+        "number": number,
+        "url": f"{_REPO}/pull/{number}",
+        "title": "build(deps): bump the dependencies",
+        "head_sha": head_sha,
+    }
+
+
+async def _seed(db: AsyncSession, slug: str, git_url: str) -> ProjectTable:
+    if await db.get(AgentTable, SYSTEM_UUID) is None:
+        db.add(
+            AgentTable(
+                id=SYSTEM_UUID,
+                name="System",
+                slug=f"system-{uuid4().hex[:8]}",
+                role=AgentRole.SYSTEM,
+                team=None,
+                status=AgentStatus.ACTIVE,
+                model_config={},
+                system_prompt="x",
+                capabilities=[],
+                permissions={},
+                metrics={},
+            )
+        )
+        await db.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name=slug,
+        slug=slug,
+        git_url=git_url,
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+    )
+    db.add(project)
+    await db.flush()
+    return project
+
+
+@pytest.mark.asyncio
+async def test_sibling_project_same_pr_is_deduped(db_session: AsyncSession) -> None:
+    fe = await _seed(db_session, "gca-frontend", _REPO)
+    be = await _seed(db_session, "gca-backend", _REPO)
+    svc = get_task_service(db_session)
+
+    first = await svc.ingest_external_pr(
+        project_id=fe.id, pr=_pr("abc123"), created_by=SYSTEM_UUID, team=Team.FRONTEND
+    )
+    assert first is not None  # first review opens
+    # Same PR + same head, sibling project on the SAME repo → no second review.
+    dup = await svc.ingest_external_pr(
+        project_id=be.id, pr=_pr("abc123"), created_by=SYSTEM_UUID, team=Team.BACKEND
+    )
+    assert dup is None
+
+
+@pytest.mark.asyncio
+async def test_exists_is_repo_scoped(db_session: AsyncSession) -> None:
+    fe = await _seed(db_session, "gca-frontend", _REPO)
+    be = await _seed(db_session, "gca-backend", _REPO)
+    svc = get_task_service(db_session)
+    await svc.ingest_external_pr(
+        project_id=fe.id, pr=_pr("abc123"), created_by=SYSTEM_UUID, team=Team.FRONTEND
+    )
+
+    # The sibling project sees the existing review (the fix); a new head SHA does not.
+    assert await svc.external_review_task_exists(be.id, 131, "abc123") is True
+    assert await svc.external_review_task_exists(be.id, 131, "newsha") is False
+
+
+@pytest.mark.asyncio
+async def test_different_repo_not_deduped(db_session: AsyncSession) -> None:
+    fe = await _seed(db_session, "gca-frontend", _REPO)
+    other = await _seed(db_session, "other", _OTHER_REPO)
+    svc = get_task_service(db_session)
+    await svc.ingest_external_pr(
+        project_id=fe.id, pr=_pr("abc123"), created_by=SYSTEM_UUID, team=Team.FRONTEND
+    )
+
+    # A genuinely different repo with the same PR number is reviewed independently.
+    created = await svc.ingest_external_pr(
+        project_id=other.id, pr=_pr("abc123"), created_by=SYSTEM_UUID, team=Team.BACKEND
+    )
+    assert created is not None

tests/integration/services/test_project_autonomy_update.py

@@ -0,0 +1,72 @@
+"""Project update accepts the autonomous-maintenance opt-in fields.
+
+The CI-watch + dep-update per-project columns must be settable through the
+normal ProjectUpdate path (what the panel edit-project dialog calls), or the
+operator can't opt a project in.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.models import AgentRole, AgentStatus, Team
+from roboco.models.project import ProjectUpdate
+from roboco.services.project import get_project_service
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_project(db_session: AsyncSession) -> ProjectTable:
+    agent = AgentTable(
+        id=uuid4(),
+        name="Dev",
+        slug=f"be-dev-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(agent)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="P",
+        slug=f"p-{uuid4().hex[:8]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=agent.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    return project
+
+
+@pytest.mark.asyncio
+async def test_update_sets_autonomy_opt_ins(db_session: AsyncSession) -> None:
+    project = await _seed_project(db_session)
+    svc = get_project_service(db_session)
+
+    await svc.update(
+        project.id,
+        ProjectUpdate(
+            ci_watch_enabled=True,
+            ci_watch_workflow="ci.yml",
+            dep_update_command="uv lock --upgrade",
+            dep_update_paths=["uv.lock"],
+        ),
+    )
+
+    reloaded = await svc.get(project.id)
+    assert reloaded is not None
+    assert reloaded.ci_watch_enabled is True
+    assert reloaded.ci_watch_workflow == "ci.yml"
+    assert reloaded.dep_update_command == "uv lock --upgrade"
+    assert reloaded.dep_update_paths == ["uv.lock"]

tests/integration/test_migration_ci_watch.py

@@ -0,0 +1,72 @@
+"""Multi-repo CI-watch per-project opt-in columns (migration 048).
+
+Migration 048 adds ``projects.ci_watch_enabled`` (bool, NOT NULL default false —
+an unopted project is unwatched) and ``projects.ci_watch_workflow`` (varchar
+null — scope the CI signal to one workflow file). The real upgrade/downgrade
+chain is verified separately against a throwaway Postgres; these assertions
+guard the resulting schema shape and a value round-trip.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.models import AgentRole, AgentStatus, Team
+from sqlalchemy import select
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_project(db_session: AsyncSession) -> ProjectTable:
+    agent = AgentTable(
+        id=uuid4(),
+        name="Dev",
+        slug=f"be-dev-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(agent)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="B-Proj",
+        slug=f"b-proj-{uuid4().hex[:8]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=agent.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    return project
+
+
+@pytest.mark.asyncio
+async def test_ci_watch_columns_default_off(db_session: AsyncSession) -> None:
+    project = await _seed_project(db_session)
+    assert project.ci_watch_enabled is False
+    assert project.ci_watch_workflow is None
+
+
+@pytest.mark.asyncio
+async def test_ci_watch_columns_round_trip(db_session: AsyncSession) -> None:
+    project = await _seed_project(db_session)
+    project.ci_watch_enabled = True
+    project.ci_watch_workflow = "ci.yml"
+    await db_session.flush()
+    row = (
+        await db_session.execute(
+            select(ProjectTable).where(ProjectTable.id == project.id)
+        )
+    ).scalar_one()
+    assert row.ci_watch_enabled is True
+    assert row.ci_watch_workflow == "ci.yml"

tests/integration/test_migration_dep_update.py

@@ -0,0 +1,71 @@
+"""Dependency-update bot per-project opt-in columns (migration 049).
+
+Migration 049 adds ``projects.dep_update_command`` (varchar null) and
+``projects.dep_update_paths`` (varchar[] null). The real upgrade/downgrade chain
+is verified separately against a throwaway Postgres; these assertions guard the
+resulting schema shape and a value round-trip.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.models import AgentRole, AgentStatus, Team
+from sqlalchemy import select
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_project(db_session: AsyncSession) -> ProjectTable:
+    agent = AgentTable(
+        id=uuid4(),
+        name="Dev",
+        slug=f"be-dev-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(agent)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="B-Proj",
+        slug=f"b-proj-{uuid4().hex[:8]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=agent.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    return project
+
+
+@pytest.mark.asyncio
+async def test_dep_update_columns_default_null(db_session: AsyncSession) -> None:
+    project = await _seed_project(db_session)
+    assert project.dep_update_command is None
+    assert project.dep_update_paths is None
+
+
+@pytest.mark.asyncio
+async def test_dep_update_columns_round_trip(db_session: AsyncSession) -> None:
+    project = await _seed_project(db_session)
+    project.dep_update_command = "uv lock --upgrade"
+    project.dep_update_paths = ["uv.lock", "pnpm-lock.yaml"]
+    await db_session.flush()
+    row = (
+        await db_session.execute(
+            select(ProjectTable).where(ProjectTable.id == project.id)
+        )
+    ).scalar_one()
+    assert row.dep_update_command == "uv lock --upgrade"
+    assert row.dep_update_paths == ["uv.lock", "pnpm-lock.yaml"]

tests/unit/config/test_ci_watch_flag.py

@@ -0,0 +1,35 @@
+"""Multi-repo CI-watch is gated by default-off config flags (mirrors self-heal)."""
+
+from __future__ import annotations
+
+import os
+from unittest import mock
+
+from roboco.config import Settings
+from roboco.services.settings import FEATURE_FLAGS, validate_setting
+
+_DEFAULT_INTERVAL = 1800
+_DEFAULT_MAX_OPEN = 3
+_DEFAULT_MAX_PER_CYCLE = 1
+
+
+def test_ci_watch_disabled_by_default() -> None:
+    s = Settings()
+    assert s.ci_watch_enabled is False
+    assert s.ci_watch_interval_seconds == _DEFAULT_INTERVAL
+    assert s.ci_watch_max_open_tasks == _DEFAULT_MAX_OPEN
+    assert s.ci_watch_max_per_cycle == _DEFAULT_MAX_PER_CYCLE
+    assert s.ci_watch_default_workflow == "ci.yml"
+
+
+def test_ci_watch_reads_env_var() -> None:
+    with mock.patch.dict(os.environ, {"ROBOCO_CI_WATCH_ENABLED": "true"}):
+        assert Settings().ci_watch_enabled is True
+
+
+def test_ci_watch_flag_registered_in_feature_flags() -> None:
+    assert "ci_watch_enabled" in [key for key, _ in FEATURE_FLAGS]
+
+
+def test_ci_watch_flag_validates_as_bool() -> None:
+    validate_setting("ci_watch_enabled", "true")

tests/unit/config/test_dep_update_flag.py

@@ -0,0 +1,34 @@
+"""The dependency-update bot is gated by default-off config flags."""
+
+from __future__ import annotations
+
+import os
+from unittest import mock
+
+from roboco.config import Settings
+from roboco.services.settings import FEATURE_FLAGS, validate_setting
+
+_DEFAULT_INTERVAL = 604800
+_DEFAULT_MAX_OPEN = 3
+_DEFAULT_MAX_PER_CYCLE = 1
+
+
+def test_dep_update_disabled_by_default() -> None:
+    s = Settings()
+    assert s.dep_update_enabled is False
+    assert s.dep_update_interval_seconds == _DEFAULT_INTERVAL
+    assert s.dep_update_max_open_tasks == _DEFAULT_MAX_OPEN
+    assert s.dep_update_max_per_cycle == _DEFAULT_MAX_PER_CYCLE
+
+
+def test_dep_update_reads_env_var() -> None:
+    with mock.patch.dict(os.environ, {"ROBOCO_DEP_UPDATE_ENABLED": "true"}):
+        assert Settings().dep_update_enabled is True
+
+
+def test_dep_update_flag_registered_in_feature_flags() -> None:
+    assert "dep_update_enabled" in [key for key, _ in FEATURE_FLAGS]
+
+
+def test_dep_update_flag_validates_as_bool() -> None:
+    validate_setting("dep_update_enabled", "true")

tests/unit/runtime/test_ci_watch_loop.py

@@ -0,0 +1,87 @@
+"""The orchestrator CI-watch loop: dormant when off, runs the engine when on.
+
+Dormant unless ``ci_watch_enabled``; loads the watch set (opted-in projects, one
+per repo), warns when enabled-but-empty, and runs CiWatchEngine.run_cycle each
+interval. Separate from the single-repo self-heal loop.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from roboco.config import settings
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+def _orch() -> AgentOrchestrator:
+    return AgentOrchestrator.__new__(AgentOrchestrator)
+
+
+@pytest.mark.asyncio
+async def test_loop_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(settings, "ci_watch_enabled", False)
+    orch = _orch()
+    cycle = AsyncMock()
+    orch._run_ci_watch_cycle = cycle  # type: ignore[method-assign]
+    await orch._ci_watch_loop()  # must return immediately, no infinite loop
+    cycle.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_load_watch_set_filters_enabled_one_per_repo() -> None:
+    orch = _orch()
+    on_a = MagicMock(slug="be", git_url="https://x/a.git", ci_watch_enabled=True)
+    on_a2 = MagicMock(slug="fe", git_url="https://x/a.git", ci_watch_enabled=True)
+    off = MagicMock(slug="c", git_url="https://x/c.git", ci_watch_enabled=False)
+    svc = MagicMock()
+    svc.list_all = AsyncMock(return_value=[on_a, on_a2, off])
+    with patch("roboco.services.project.get_project_service", return_value=svc):
+        watch = await orch._load_ci_watch_set(MagicMock())
+    assert len(watch) == 1  # opt-out excluded; same-repo cell-projects collapsed
+    assert watch[0].git_url == "https://x/a.git"
+
+
+def _db_ctx(db: Any):
+    @asynccontextmanager
+    async def _ctx() -> Any:
+        yield db
+
+    return _ctx
+
+
+@pytest.mark.asyncio
+async def test_cycle_warns_and_skips_engine_when_empty() -> None:
+    orch = _orch()
+    orch._load_ci_watch_set = AsyncMock(return_value=[])  # type: ignore[method-assign]
+    get_eng = MagicMock()
+    with (
+        patch("roboco.db.get_db_context", _db_ctx(MagicMock())),
+        patch("roboco.services.ci_watch_engine.get_ci_watch_engine", get_eng),
+    ):
+        await orch._run_ci_watch_cycle()
+    get_eng.assert_not_called()  # empty watch set → no engine run
+
+
+@pytest.mark.asyncio
+async def test_cycle_runs_engine_when_watch_set_present() -> None:
+    orch = _orch()
+    watch = [MagicMock()]
+    orch._load_ci_watch_set = AsyncMock(return_value=watch)  # type: ignore[method-assign]
+    db = MagicMock()
+    db.commit = AsyncMock()
+    engine = MagicMock()
+    engine.run_cycle = AsyncMock(return_value=[])
+    with (
+        patch("roboco.db.get_db_context", _db_ctx(db)),
+        patch(
+            "roboco.services.ci_watch_engine.get_ci_watch_engine",
+            return_value=engine,
+        ) as get_eng,
+    ):
+        await orch._run_ci_watch_cycle()
+    get_eng.assert_called_once()
+    engine.run_cycle.assert_awaited_once_with(watch)
+    db.commit.assert_awaited_once()

tests/unit/runtime/test_dep_update_loop.py

@@ -0,0 +1,87 @@
+"""The orchestrator dep-update loop: dormant when off, runs the engine when on.
+
+Dormant unless ``dep_update_enabled``; loads the eligible set (projects with a
+dep_update_command, one per repo), warns when enabled-but-empty, and runs
+DepUpdateEngine.run_cycle each interval. Separate from self-heal/CI-watch loops.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from roboco.config import settings
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+def _orch() -> AgentOrchestrator:
+    return AgentOrchestrator.__new__(AgentOrchestrator)
+
+
+@pytest.mark.asyncio
+async def test_loop_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(settings, "dep_update_enabled", False)
+    orch = _orch()
+    cycle = AsyncMock()
+    orch._run_dep_update_cycle = cycle  # type: ignore[method-assign]
+    await orch._dep_update_loop()
+    cycle.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_load_set_filters_command_one_per_repo() -> None:
+    orch = _orch()
+    on_a = MagicMock(slug="be", git_url="https://x/a.git", dep_update_command="uv -U")
+    on_a2 = MagicMock(slug="fe", git_url="https://x/a.git", dep_update_command="uv -U")
+    off = MagicMock(slug="c", git_url="https://x/c.git", dep_update_command=None)
+    svc = MagicMock()
+    svc.list_all = AsyncMock(return_value=[on_a, on_a2, off])
+    with patch("roboco.services.project.get_project_service", return_value=svc):
+        eligible = await orch._load_dep_update_set(MagicMock())
+    assert len(eligible) == 1  # no-command excluded; same-repo collapsed
+    assert eligible[0].git_url == "https://x/a.git"
+
+
+def _db_ctx(db: Any):
+    @asynccontextmanager
+    async def _ctx() -> Any:
+        yield db
+
+    return _ctx
+
+
+@pytest.mark.asyncio
+async def test_cycle_warns_and_skips_engine_when_empty() -> None:
+    orch = _orch()
+    orch._load_dep_update_set = AsyncMock(return_value=[])  # type: ignore[method-assign]
+    get_eng = MagicMock()
+    with (
+        patch("roboco.db.get_db_context", _db_ctx(MagicMock())),
+        patch("roboco.services.dep_update_engine.get_dep_update_engine", get_eng),
+    ):
+        await orch._run_dep_update_cycle()
+    get_eng.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_cycle_runs_engine_when_eligible_present() -> None:
+    orch = _orch()
+    eligible = [MagicMock()]
+    orch._load_dep_update_set = AsyncMock(return_value=eligible)  # type: ignore[method-assign]
+    db = MagicMock()
+    db.commit = AsyncMock()
+    engine = MagicMock()
+    engine.run_cycle = AsyncMock(return_value=[])
+    with (
+        patch("roboco.db.get_db_context", _db_ctx(db)),
+        patch(
+            "roboco.services.dep_update_engine.get_dep_update_engine",
+            return_value=engine,
+        ) as get_eng,
+    ):
+        await orch._run_dep_update_cycle()
+    get_eng.assert_called_once()
+    engine.run_cycle.assert_awaited_once_with(eligible)
+    db.commit.assert_awaited_once()

tests/unit/runtime/test_image_prune_sweep.py

@@ -0,0 +1,85 @@
+"""The orchestrator sweeper prunes dangling Docker images (gated + throttled).
+
+Each agent-image rebuild orphans the prior build's layers as an untagged
+``<none>`` image; over many deploys these pile up. The sweeper reclaims them
+with ``docker image prune -f --filter dangling=true`` (dangling only — never a
+tagged image or one backing a running container), gated by
+``image_prune_enabled`` and throttled to ``image_prune_interval_seconds``.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from roboco.config import settings
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+def _orch() -> AgentOrchestrator:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)  # bypass __init__
+    orch._last_image_prune = None
+    return orch
+
+
+def _fake_proc(returncode: int = 0) -> MagicMock:
+    proc = MagicMock()
+    proc.returncode = returncode
+    proc.communicate = AsyncMock(return_value=(b"Total reclaimed space: 1.2GB", b""))
+    return proc
+
+
+@pytest.mark.asyncio
+async def test_prunes_dangling_when_enabled_and_due(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "image_prune_enabled", True)
+    orch = _orch()
+    spawn = AsyncMock(return_value=_fake_proc())
+    with patch("roboco.runtime.orchestrator.asyncio.create_subprocess_exec", spawn):
+        await orch._sweep_dangling_images()
+    spawn.assert_awaited_once()
+    args: tuple[Any, ...] = spawn.await_args.args
+    assert args[:6] == (
+        "docker",
+        "image",
+        "prune",
+        "-f",
+        "--filter",
+        "dangling=true",
+    )
+    assert orch._last_image_prune is not None
+
+
+@pytest.mark.asyncio
+async def test_no_prune_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(settings, "image_prune_enabled", False)
+    orch = _orch()
+    spawn = AsyncMock()
+    with patch("roboco.runtime.orchestrator.asyncio.create_subprocess_exec", spawn):
+        await orch._sweep_dangling_images()
+    spawn.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_throttled_when_recently_pruned(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "image_prune_enabled", True)
+    orch = _orch()
+    orch._last_image_prune = datetime.now(UTC)  # pruned moments ago
+    spawn = AsyncMock()
+    with patch("roboco.runtime.orchestrator.asyncio.create_subprocess_exec", spawn):
+        await orch._sweep_dangling_images()
+    spawn.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_best_effort_swallows_errors(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(settings, "image_prune_enabled", True)
+    orch = _orch()
+    spawn = AsyncMock(side_effect=RuntimeError("no docker"))
+    with patch("roboco.runtime.orchestrator.asyncio.create_subprocess_exec", spawn):
+        await orch._sweep_dangling_images()  # must not raise

tests/unit/services/test_multi_ci_telemetry.py

@@ -0,0 +1,86 @@
+"""MultiProjectCITelemetrySource fans out the hardened per-project CI lookup.
+
+One red project yields a breaching sample; a green one a non-breaching sample; a
+None signal or a per-project error yields NO sample (unknown, never "green") and
+never aborts the sweep. Each project's ci_watch_workflow (or the configured
+default) is passed through to the reused lookup.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from roboco.config import settings
+from roboco.services.telemetry.source import MultiProjectCITelemetrySource
+
+
+def _project(slug: str, workflow: str | None = None) -> MagicMock:
+    return MagicMock(slug=slug, ci_watch_workflow=workflow)
+
+
+def _ci(conclusion: str) -> dict[str, Any]:
+    return {
+        "conclusion": conclusion,
+        "branch": "master",
+        "run_url": f"https://github.com/x/{conclusion}/actions/runs/1",
+        "completed_at": "2026-06-25T00:00:00Z",
+        "run_name": "CI",
+    }
+
+
+@pytest.mark.asyncio
+async def test_fanout_red_green_and_none() -> None:
+    projects = [_project("red"), _project("green"), _project("nosig")]
+
+    async def conclusion(slug: str, **_kwargs: Any) -> Any:
+        return {"red": _ci("failure"), "green": _ci("success"), "nosig": None}[slug]
+
+    git = MagicMock()
+    git.get_latest_ci_conclusion = AsyncMock(side_effect=conclusion)
+    with patch("roboco.services.telemetry.source.GitService", return_value=git):
+        samples = await MultiProjectCITelemetrySource(MagicMock()).fetch(projects)
+
+    by_repo = {s.repo_hint: s for s in samples}
+    assert by_repo["red"].is_breach is True
+    assert by_repo["green"].is_breach is False
+    assert "nosig" not in by_repo  # None signal → no sample (unknown, not green)
+
+
+@pytest.mark.asyncio
+async def test_per_project_error_isolated() -> None:
+    projects = [_project("boom"), _project("ok")]
+
+    async def conclusion(slug: str, **_kwargs: Any) -> Any:
+        if slug == "boom":
+            raise RuntimeError("github down")
+        return _ci("failure")
+
+    git = MagicMock()
+    git.get_latest_ci_conclusion = AsyncMock(side_effect=conclusion)
+    with patch("roboco.services.telemetry.source.GitService", return_value=git):
+        samples = await MultiProjectCITelemetrySource(MagicMock()).fetch(projects)
+
+    by_repo = {s.repo_hint: s for s in samples}
+    assert "boom" not in by_repo  # error → no sample, never aborts the sweep
+    assert by_repo["ok"].is_breach is True  # others still returned
+
+
+@pytest.mark.asyncio
+async def test_per_project_workflow_passthrough(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "ci_watch_default_workflow", "ci.yml")
+    projects = [_project("custom", workflow="release.yml"), _project("default")]
+    git = MagicMock()
+    git.get_latest_ci_conclusion = AsyncMock(return_value=_ci("success"))
+    with patch("roboco.services.telemetry.source.GitService", return_value=git):
+        await MultiProjectCITelemetrySource(MagicMock()).fetch(projects)
+
+    workflows = {
+        c.args[0]: c.kwargs["workflow"]
+        for c in git.get_latest_ci_conclusion.await_args_list
+    }
+    assert workflows["custom"] == "release.yml"
+    assert workflows["default"] == "ci.yml"