[F115] sample monorepo per (repo,workflow)/(repo,command) not per repo

The CI-watch and dep-update loaders collapsed a monorepo's cell-projects
to one canonical entry per repo (slug-sorted-first), so a repo whose cells
each carry their OWN ci_watch_workflow / dep_update_command had only the
canonical cell's workflow/command sampled — a red on another cell's
workflow or drift on another cell's lockfile was missed (under-count).

Refactor the shared one-per-repo collapse into _projects_one_per_key, keyed
by repo identity for external-PR discovery (unchanged: one review per PR per
repo), by (repo, effective workflow) for CI-watch, and by (repo, command)
for dep-update. Each distinct workflow/command is now sampled once; the
engines' per-git_url fix-task dedup still prevents duplicate fix tasks for
the same repo. _projects_one_per_repo now delegates to _projects_one_per_key.

key_fn uses a string annotation (Callable lives under TYPE_CHECKING, like
the existing Coroutine/Iterable annotations at lines 4193/5279).

roboco/runtime/orchestrator.py

@@ -6235,10 +6235,18 @@ Start by:
             await db.commit()
 
     async def _load_ci_watch_set(self, db: Any) -> list[Any]:
-        """Opted-in projects (``ci_watch_enabled`` + a git_url), one per repo.
+        """Opted-in projects (``ci_watch_enabled`` + a git_url), one per
+        (repo, workflow).
 
-        Collapsing to one canonical project per repo means a monorepo's several
-        cell-projects are watched as a single repo, not N times.
+        A monorepo's several cell-projects can each carry their OWN
+        ``ci_watch_workflow`` (e.g. a backend CI workflow distinct from the
+        frontend's). Collapsing to one canonical project per REPO would watch
+        only the canonical cell's workflow and miss a red on the others (the
+        under-count). Collapse to one canonical project per (repo, effective
+        workflow) instead — every distinct workflow is sampled once, and the
+        engine's per-``git_url`` fix-task dedup still prevents a duplicate fix
+        task for the same repo. The effective workflow is the project override
+        or ``ci_watch_default_workflow`` (matching ``MultiProjectCITelemetrySource``).
         """
         from roboco.services.project import get_project_service
 
@@ -6248,7 +6256,28 @@ Start by:
             for p in projects
             if getattr(p, "ci_watch_enabled", False) and getattr(p, "git_url", None)
         ]
-        return self._projects_one_per_repo(watched)
+        return self._projects_one_per_key(
+            watched,
+            key_fn=lambda p: (
+                self._repo_key(str(getattr(p, "git_url", "") or "")),
+                self._effective_ci_watch_workflow(p),
+            ),
+        )
+
+    @staticmethod
+    def _effective_ci_watch_workflow(project: Any) -> str | None:
+        """The workflow that will actually be polled for ``project``.
+
+        Mirrors ``MultiProjectCITelemetrySource._sample_for``: the project's
+        ``ci_watch_workflow`` override, falling back to the global
+        ``ci_watch_default_workflow``. Used as the per-(repo, workflow) collapse
+        key so two cells sharing a workflow still collapse to one sample.
+        """
+        workflow = str(
+            getattr(project, "ci_watch_workflow", None)
+            or settings.ci_watch_default_workflow
+        ).strip()
+        return workflow or None
 
     async def _dep_update_loop(self) -> None:
         """Dependency-update bot: probe opted-in projects, open update tasks.
@@ -6323,7 +6352,18 @@ Start by:
             await db.commit()
 
     async def _load_dep_update_set(self, db: Any) -> list[Any]:
-        """Projects with a ``dep_update_command`` + a git_url, one per repo."""
+        """Projects with a ``dep_update_command`` + a git_url, one per
+        (repo, command).
+
+        A monorepo's several cell-projects can each carry their OWN
+        ``dep_update_command`` (different ecosystems → different lockfiles,
+        e.g. ``uv lock --upgrade`` vs ``pnpm update -L``). Collapsing to one
+        canonical project per REPO would probe only the canonical cell's
+        lockfile and miss the others' drift (the under-count). Collapse to one
+        canonical project per (repo, command) instead — every distinct command
+        is probed once, and the engine's per-``git_url`` open-task dedup still
+        prevents a duplicate update task for the same repo.
+        """
         from roboco.services.project import get_project_service
 
         projects = await get_project_service(db).list_all(active_only=True)
@@ -6333,7 +6373,13 @@ Start by:
             if str(getattr(p, "dep_update_command", None) or "").strip()
             and getattr(p, "git_url", None)
         ]
-        return self._projects_one_per_repo(eligible)
+        return self._projects_one_per_key(
+            eligible,
+            key_fn=lambda p: (
+                self._repo_key(str(getattr(p, "git_url", "") or "")),
+                str(getattr(p, "dep_update_command", None) or "").strip(),
+            ),
+        )
 
     @staticmethod
     def _repo_key(git_url: str) -> str:
@@ -6351,14 +6397,41 @@ Start by:
         projects). Collapse to one canonical project per repo (deterministic by
         slug so the pick is stable across polls); genuinely separate repos
         (multi-repo) each keep their own. Projects without a git_url are skipped.
+
+        Used by the external-PR discovery path (one review per PR per repo). The
+        CI-watch and dep-update loaders use :meth:`_projects_one_per_key` with a
+        finer (repo, workflow) / (repo, command) key so a monorepo's per-cell
+        workflow / lockfile-command overrides are each sampled once instead of
+        collapsing to the canonical cell's value.
         """
-        seen: set[str] = set()
+        return cls._projects_one_per_key(
+            projects,
+            key_fn=lambda p: (cls._repo_key(str(getattr(p, "git_url", "") or "")),),
+        )
+
+    @classmethod
+    def _projects_one_per_key(
+        cls, projects: list[Any], *, key_fn: "Callable[[Any], tuple[Any, ...]]"
+    ) -> list[Any]:
+        """One canonical project per distinct key (deterministic by slug).
+
+        ``key_fn`` defines what distinguishes a duplicate: repo identity for
+        external-PR discovery (one review per PR per repo); ``(repo, workflow)``
+        for CI-watch and ``(repo, command)`` for dep-update so a monorepo's
+        several cell-projects — each potentially carrying its OWN workflow /
+        lockfile command — are each sampled once instead of collapsing to the
+        canonical cell's value (the under-count fixed by F115). The first
+        project (by slug) per key is the canonical pick; the engine's
+        per-``git_url`` fix-task dedup still prevents duplicate fix tasks for the
+        same repo. Projects without a git_url are skipped.
+        """
+        seen: set[tuple[Any, ...]] = set()
         canonical: list[Any] = []
-        for project in sorted(projects, key=lambda p: str(p.slug)):
+        for project in sorted(projects, key=lambda p: str(getattr(p, "slug", ""))):
             git_url = getattr(project, "git_url", None)
             if not git_url:
                 continue
-            key = cls._repo_key(git_url)
+            key = key_fn(project)
             if key in seen:
                 continue
             seen.add(key)

tests/unit/runtime/test_ci_watch_loop.py

@@ -33,17 +33,59 @@ async def test_loop_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
 @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)
+    # Same repo, SAME effective workflow (both fall back to the default) → one
+    # canonical entry; the opt-out is excluded.
+    on_a = MagicMock(
+        slug="be",
+        git_url="https://x/a.git",
+        ci_watch_enabled=True,
+        ci_watch_workflow=None,
+    )
+    on_a2 = MagicMock(
+        slug="fe",
+        git_url="https://x/a.git",
+        ci_watch_enabled=True,
+        ci_watch_workflow=None,
+    )
     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 len(watch) == 1  # opt-out excluded; same-repo + same-workflow collapsed
     assert watch[0].git_url == "https://x/a.git"
 
 
+@pytest.mark.asyncio
+async def test_load_watch_set_keeps_distinct_workflows_per_repo() -> None:
+    """F115: a monorepo's several cell-projects each carrying their OWN
+    ``ci_watch_workflow`` must ALL be watched — collapsing to the canonical
+    cell's workflow would miss a red on the other cells' workflows (under-count).
+    Same repo, DIFFERENT workflows → one entry per (repo, workflow). The engine's
+    per-git_url fix-task dedup still prevents duplicate fix tasks for the repo."""
+    orch = _orch()
+    be = MagicMock(
+        slug="be",
+        git_url="https://x/a.git",
+        ci_watch_enabled=True,
+        ci_watch_workflow="backend-ci.yml",
+    )
+    fe = MagicMock(
+        slug="fe",
+        git_url="https://x/a.git",
+        ci_watch_enabled=True,
+        ci_watch_workflow="frontend-ci.yml",
+    )
+    svc = MagicMock()
+    svc.list_all = AsyncMock(return_value=[be, fe])
+    with patch("roboco.services.project.get_project_service", return_value=svc):
+        watch = await orch._load_ci_watch_set(MagicMock())
+    # distinct workflows both watched, NOT collapsed to one canonical cell —
+    # the set-equality assertion proves exactly-two (no magic-value literal).
+    workflows = {p.ci_watch_workflow for p in watch}
+    assert workflows == {"backend-ci.yml", "frontend-ci.yml"}
+
+
 def _db_ctx(db: Any) -> Any:
     @asynccontextmanager
     async def _ctx() -> Any:

tests/unit/runtime/test_dep_update_loop.py

@@ -40,10 +40,35 @@ async def test_load_set_filters_command_one_per_repo() -> None:
     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 len(eligible) == 1  # no-command excluded; same-repo + same-command collapsed
     assert eligible[0].git_url == "https://x/a.git"
 
 
+@pytest.mark.asyncio
+async def test_load_set_keeps_distinct_commands_per_repo() -> None:
+    """F115: a monorepo's several cell-projects each carrying their OWN
+    ``dep_update_command`` (different ecosystems → different lockfiles) must
+    ALL be probed — collapsing to the canonical cell's command would miss the
+    other cells' lockfile drift (under-count). Same repo, DIFFERENT commands →
+    one entry per (repo, command). The engine's per-git_url open-task dedup
+    still prevents duplicate update tasks for the repo."""
+    orch = _orch()
+    be = MagicMock(
+        slug="be", git_url="https://x/a.git", dep_update_command="uv lock --upgrade"
+    )
+    fe = MagicMock(
+        slug="fe", git_url="https://x/a.git", dep_update_command="pnpm update -L"
+    )
+    svc = MagicMock()
+    svc.list_all = AsyncMock(return_value=[be, fe])
+    with patch("roboco.services.project.get_project_service", return_value=svc):
+        eligible = await orch._load_dep_update_set(MagicMock())
+    # distinct commands both probed, NOT collapsed to one canonical cell —
+    # the set-equality assertion proves exactly-two (no magic-value literal).
+    commands = {p.dep_update_command for p in eligible}
+    assert commands == {"uv lock --upgrade", "pnpm update -L"}
+
+
 def _db_ctx(db: Any) -> Any:
     @asynccontextmanager
     async def _ctx() -> Any: