diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1cf4a1ab..11da5a30 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,6 +4,21 @@ on: push: branches: - master + # Fleet task branches (GitService push/merge, roboco/services/git.py). + # A revision commit that lands on a PR's head via the merge API + # (squash-merging a subtask PR into a parent branch) doesn't reliably + # fire `pull_request`'s synchronize trigger for the PR that already + # has that branch as its head — proven live on PR #406, where two + # revision merges left CI/CodeQL absent (not red) while `push` and + # `pull_request_target` both fired for the same ref update. A real git + # push always fires `pull_request`; this redundant trigger (paired + # with the concurrency group below) closes the gap for merge-API + # revisions without double-running when both events land. + - 'feature/**' + - 'bug/**' + - 'chore/**' + - 'docs/**' + - 'hotfix/**' paths: - 'roboco/**' - 'agents/**' @@ -35,6 +50,17 @@ on: - '.github/workflows/ci.yml' workflow_dispatch: +# A fleet branch that's also an open PR head can get both a `push` and a +# `pull_request` run for the same commit; cancel the older one instead of +# burning two runners on identical work. `head_ref` (set only for +# pull_request) and `ref_name` (the short branch name, valid for push) both +# resolve to the SAME branch name, so the two event shapes share one group — +# plain `github.ref` would NOT (it's `refs/pull//merge` for pull_request +# vs `refs/heads/` for push, so it'd never collapse them). +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true + jobs: quality: name: Python quality gate diff --git a/.github/workflows/code-ql.yml b/.github/workflows/code-ql.yml index 95c79666..6be5e054 100644 --- a/.github/workflows/code-ql.yml +++ b/.github/workflows/code-ql.yml @@ -2,7 +2,12 @@ name: CodeQL on: push: - branches: [master] + # master plus fleet task branches: `pull_request`'s synchronize trigger + # doesn't reliably fire when a revision lands on a PR head via the + # merge API (see ci.yml for the live-proven receipts); `push` does, so + # it's the redundant trigger for a required check that must not go + # ABSENT on a fleet-authored PR revision. + branches: [master, 'feature/**', 'bug/**', 'chore/**', 'docs/**', 'hotfix/**'] paths: - 'roboco/**' - 'agents/**' @@ -25,6 +30,17 @@ on: - cron: '0 0 * * 1' workflow_dispatch: +# A fleet branch that's also an open PR head can get both a `push` and a +# `pull_request` run for the same commit; cancel the older one instead of +# burning two runners on identical work. `head_ref` (set only for +# pull_request) and `ref_name` (the short branch name, valid for push) both +# resolve to the SAME branch name, so the two event shapes share one group — +# plain `github.ref` would NOT (it's `refs/pull//merge` for pull_request +# vs `refs/heads/` for push, so it'd never collapse them). +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true + jobs: analyze: name: Analyze (${{ matrix.language }}) diff --git a/.github/workflows/e2e-smoke.yml b/.github/workflows/e2e-smoke.yml index 7e5258f7..2013f8f7 100644 --- a/.github/workflows/e2e-smoke.yml +++ b/.github/workflows/e2e-smoke.yml @@ -4,6 +4,14 @@ on: push: branches: - master + # Fleet task branches: `pull_request`'s synchronize trigger doesn't + # reliably fire when a revision lands on a PR head via the merge API + # (see ci.yml for the live-proven receipts); `push` does. + - 'feature/**' + - 'bug/**' + - 'chore/**' + - 'docs/**' + - 'hotfix/**' paths: - 'roboco/**' - 'alembic/**' @@ -25,6 +33,17 @@ on: - '.github/workflows/e2e-smoke.yml' workflow_dispatch: +# A fleet branch that's also an open PR head can get both a `push` and a +# `pull_request` run for the same commit; cancel the older one instead of +# burning two runners on identical work. `head_ref` (set only for +# pull_request) and `ref_name` (the short branch name, valid for push) both +# resolve to the SAME branch name, so the two event shapes share one group — +# plain `github.ref` would NOT (it's `refs/pull//merge` for pull_request +# vs `refs/heads/` for push, so it'd never collapse them). +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true + jobs: e2e-smoke: name: e2e lifecycle smoke (scripted agents) diff --git a/.github/workflows/panel-ci.yml b/.github/workflows/panel-ci.yml index 8efe7e13..d5f2255a 100644 --- a/.github/workflows/panel-ci.yml +++ b/.github/workflows/panel-ci.yml @@ -4,6 +4,14 @@ on: push: branches: - master + # Fleet task branches: `pull_request`'s synchronize trigger doesn't + # reliably fire when a revision lands on a PR head via the merge API + # (see ci.yml for the live-proven receipts); `push` does. + - 'feature/**' + - 'bug/**' + - 'chore/**' + - 'docs/**' + - 'hotfix/**' paths: - 'panel/**' - '.github/workflows/panel-ci.yml' @@ -15,6 +23,17 @@ on: - '.github/workflows/panel-ci.yml' workflow_dispatch: +# A fleet branch that's also an open PR head can get both a `push` and a +# `pull_request` run for the same commit; cancel the older one instead of +# burning two runners on identical work. `head_ref` (set only for +# pull_request) and `ref_name` (the short branch name, valid for push) both +# resolve to the SAME branch name, so the two event shapes share one group — +# plain `github.ref` would NOT (it's `refs/pull//merge` for pull_request +# vs `refs/heads/` for push, so it'd never collapse them). +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true + jobs: panel: name: Panel (Next.js) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 2edf6c62..5a6a50c7 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -11583,10 +11583,43 @@ Start now: evidence(task_id="{task_id}") return self._build_main_pm_triage_prompt(task) return self._build_pm_triage_prompt(task) + async def _pending_claim_blocked(self, task_id: str | None) -> bool: + """Dispatch-time probe: is ``task_id`` held by the dependency/sequence + guard right now? + + `_dispatch_pm_work` fetches every PENDING task each tick with no + sequencing filter, so a later-wave MegaTask root-subtask (or any + dependency-blocked task) got a doomed claim attempt every tick — + harmless (the claim chokepoint already refuses it) but pure churn. + Reuses `TaskService.is_pending_claim_blocked` (the exact claim-gate + predicate) so this can't drift from what the claim endpoint enforces. + Fails open (False) on any lookup error — the claim attempt itself is + the safety net and will surface a real error if something's wrong. + """ + if not task_id: + return False + from uuid import UUID + + from roboco.db.base import get_db_context + from roboco.services.task import TaskService + + try: + async with get_db_context() as db: + return await TaskService(db).is_pending_claim_blocked(UUID(task_id)) + except Exception as exc: + logger.warning( + "Claim-block probe failed; falling through to claim attempt", + task_id=task_id, + error=str(exc), + ) + return False + async def _route_unassigned_pm_task( self, client: httpx.AsyncClient, task: dict[str, Any] ) -> None: """Classify and route an unassigned pending task to its target agent.""" + if await self._pending_claim_blocked(task.get("id")): + return routing = self._classify_task_routing(task) agent_id = self._get_routing_target(routing, task) diff --git a/roboco/services/task.py b/roboco/services/task.py index 494eabf2..f766e5b2 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -2750,6 +2750,25 @@ class TaskService(BaseService): return True return await self._claim_blocked_by_sequence(task) is not None + async def is_pending_claim_blocked(self, task_id: UUID) -> bool: + """Public read-only probe: would a claim on ``task_id`` be refused + right now by the dependency/sequence guard? + + Wraps the exact `_claim_blocked_by_sequencing` predicate the claim + chokepoint enforces (no duplicated SQL), so dispatch-time filtering + can't drift from claim-time enforcement. Meant for a dispatcher to + skip a doomed claim attempt on a held later-wave task instead of + round-tripping the claim endpoint every tick — the guard queries + here are the same cost either way; this just avoids the wasted + HTTP claim call and its logging noise. False (not blocked) on a + missing task — nothing to hold back, the caller's claim attempt + will get its own real error. + """ + task = await self.get(task_id) + if task is None: + return False + return await self._claim_blocked_by_sequencing(task) + async def _validate_claim_preconditions( self, task: TaskTable, diff --git a/tests/integration/test_task_service_basics.py b/tests/integration/test_task_service_basics.py index 28e7f889..5120d244 100644 --- a/tests/integration/test_task_service_basics.py +++ b/tests/integration/test_task_service_basics.py @@ -1510,6 +1510,69 @@ async def test_claim_blocked_by_sequence_names_distinct_reason( assert "seq-0 blocker" in reason +# --------------------------------------------------------------------------- +# is_pending_claim_blocked — the public dispatch-time probe the orchestrator +# fetch filter uses to skip a doomed claim attempt (churn reduction). +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_is_pending_claim_blocked_true_for_lower_sequence_sibling( + task_setup: dict, db_session: AsyncSession +) -> None: + svc = task_setup["svc"] + parent = await svc.create(_req(task_setup, title="parent")) + seq0 = await svc.create( + _req(task_setup, title="seq-0 blocker", parent_task_id=parent.id, sequence=0) + ) + seq1 = await svc.create( + _req(task_setup, title="seq-1", parent_task_id=parent.id, sequence=1) + ) + seq0.status = TaskStatus.IN_PROGRESS + await db_session.flush() + + assert await svc.is_pending_claim_blocked(seq1.id) is True + + seq0.status = TaskStatus.COMPLETED + await db_session.flush() + assert await svc.is_pending_claim_blocked(seq1.id) is False + + +@pytest.mark.asyncio +async def test_is_pending_claim_blocked_true_for_unmet_dependency( + task_setup: dict, db_session: AsyncSession +) -> None: + svc = task_setup["svc"] + dep = await svc.create(_req(task_setup, title="dependency")) + task = await svc.create(_req(task_setup, title="dependent")) + await db_session.flush() + await svc.add_dependency(task.id, dep.id) + + assert await svc.is_pending_claim_blocked(task.id) is True + + dep.status = TaskStatus.COMPLETED + await db_session.flush() + assert await svc.is_pending_claim_blocked(task.id) is False + + +@pytest.mark.asyncio +async def test_is_pending_claim_blocked_false_for_clear_task( + task_setup: dict, +) -> None: + """A ready task (no dependency edge, sequence 0, no parent) is never held.""" + svc = task_setup["svc"] + task = await svc.create(_req(task_setup, title="ready")) + assert await svc.is_pending_claim_blocked(task.id) is False + + +@pytest.mark.asyncio +async def test_is_pending_claim_blocked_false_for_missing_task( + task_setup: dict, +) -> None: + svc = task_setup["svc"] + assert await svc.is_pending_claim_blocked(uuid4()) is False + + @pytest.mark.asyncio async def test_claim_batch_wave_blocked_by_all_wave0_siblings_no_edges( task_setup: dict, db_session: AsyncSession diff --git a/tests/unit/runtime/test_pm_dispatch_claim_prefilter.py b/tests/unit/runtime/test_pm_dispatch_claim_prefilter.py new file mode 100644 index 00000000..9ba9ac51 --- /dev/null +++ b/tests/unit/runtime/test_pm_dispatch_claim_prefilter.py @@ -0,0 +1,158 @@ +"""Dispatch-time claim-gate prefilter (churn reduction). + +`_dispatch_pm_work` fetches every PENDING task each tick with no +dependency/sequence filter, so a later-wave / dependency-blocked task got a +doomed claim attempt every tick — harmless (the claim chokepoint already +refuses it via `TaskService._claim_blocked_by_sequencing`) but pure churn, +each probe paying for the guards' sibling queries a second time. +`_route_unassigned_pm_task` now consults `_pending_claim_blocked` (backed by +the public `TaskService.is_pending_claim_blocked`) before ever calling +`_claim_task_for_agent`, so a held task is skipped for the tick instead of +round-tripping the claim endpoint. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, Any, cast +from unittest.mock import AsyncMock, patch +from uuid import uuid4 + +import pytest +from roboco.runtime.orchestrator import AgentOrchestrator + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + import httpx + + +def _orch() -> AgentOrchestrator: + orch = object.__new__(AgentOrchestrator) + orch._instances = {} + return orch + + +def _pending_task(**over: Any) -> dict[str, Any]: + base: dict[str, Any] = { + "id": str(uuid4()), + "status": "pending", + "team": "backend", + "task_type": "code", + "title": "Some code task", + "assigned_to": None, + "created_by": None, + } + base.update(over) + return base + + +# --------------------------------------------------------------------------- +# _route_unassigned_pm_task — the prefilter must gate _claim_task_for_agent +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_held_task_is_not_claim_attempted() -> None: + """A blocked task (unmet dependency / lower-sequence sibling) skips the + claim call entirely — no HTTP round trip, no downstream routing.""" + orch = _orch() + task = _pending_task() + client = cast("httpx.AsyncClient", object()) + with ( + patch.object(orch, "_pending_claim_blocked", new=AsyncMock(return_value=True)), + patch.object(orch, "_classify_task_routing") as classify, + patch.object(orch, "_claim_task_for_agent", new=AsyncMock()) as claim, + patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn, + ): + await orch._route_unassigned_pm_task(client, task) + + claim.assert_not_awaited() + spawn.assert_not_awaited() + classify.assert_not_called() # never even reaches routing classification + + +@pytest.mark.asyncio +async def test_ready_task_still_dispatches() -> None: + """A clear task (prefilter returns False) proceeds through the normal + claim + spawn flow unchanged.""" + orch = _orch() + task = _pending_task() + client = cast("httpx.AsyncClient", object()) + with ( + patch.object(orch, "_pending_claim_blocked", new=AsyncMock(return_value=False)), + patch.object(orch, "_classify_task_routing", return_value="dev"), + patch.object(orch, "_get_routing_target", return_value="be-dev-1"), + patch.object(orch, "_is_agent_active", return_value=False), + patch.object(orch, "_task_git_context", return_value=None), + patch.object( + orch, "_claim_task_for_agent", new=AsyncMock(return_value=True) + ) as claim, + patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn, + ): + await orch._route_unassigned_pm_task(client, task) + + claim.assert_awaited_once_with(client, task["id"], "be-dev-1") + spawn.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# _pending_claim_blocked — the DB-backed probe itself +# --------------------------------------------------------------------------- + + +def _patch_task_service_db(task_svc: AsyncMock) -> tuple[Any, Any]: + @asynccontextmanager + async def _fake_ctx() -> AsyncIterator[AsyncMock]: + yield AsyncMock() + + return ( + patch("roboco.db.base.get_db_context", _fake_ctx), + patch("roboco.services.task.TaskService", return_value=task_svc), + ) + + +@pytest.mark.asyncio +async def test_pending_claim_blocked_delegates_to_task_service() -> None: + orch = _orch() + task_svc = AsyncMock() + task_svc.is_pending_claim_blocked = AsyncMock(return_value=True) + task_id = str(uuid4()) + + db_ctx, task_ctx = _patch_task_service_db(task_svc) + with db_ctx, task_ctx: + assert await orch._pending_claim_blocked(task_id) is True + + task_svc.is_pending_claim_blocked.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_pending_claim_blocked_false_without_task_id() -> None: + """No task id — nothing to probe; never touches the DB.""" + orch = _orch() + with patch("roboco.db.base.get_db_context") as ctx: + assert await orch._pending_claim_blocked(None) is False + ctx.assert_not_called() + + +class _BoomCtx: + """Async context manager that blows up on entry — simulates a DB hiccup.""" + + async def __aenter__(self) -> None: + raise RuntimeError("db unavailable") + + async def __aexit__(self, *exc_info: object) -> bool: + return False + + +@pytest.mark.asyncio +async def test_pending_claim_blocked_fails_open_on_error() -> None: + """A DB/lookup error never blocks dispatch — the claim attempt is the + real safety net and will surface its own error.""" + orch = _orch() + with patch("roboco.db.base.get_db_context", return_value=_BoomCtx()): + assert await orch._pending_claim_blocked(str(uuid4())) is False + + +if __name__ == "__main__": + pytest.main([__file__, "-q"])