fix(ci): fleet-branch push triggers + dispatcher claim prefilter (#463)

* fix(ci): fleet-branch push triggers close the absent-check gap; dispatcher claim prefilter

PROVEN with API receipts: when the PM squash-merges a subtask PR into a
branch that is itself another PR's head (GitService.merge_pull_request →
GitHub's Merge API), the pull_request synchronize webhook fires
unreliably (1 of 3 in the live sample) while plain push events fired
100% — so PR heads sat with ABSENT required checks that three review
rounds mistook for green. CI, CodeQL, e2e-smoke, and panel-ci now also
trigger on push to the fleet's branch types, deduped by a concurrency
group keyed on head_ref||ref_name so a branch that is also a PR head
never double-runs.

Dispatcher churn: _route_unassigned_pm_task consults the claim guards'
own predicate (TaskService.is_pending_claim_blocked, a public wrapper —
no duplicated SQL) before routing, so dependency- or sequence-held
tasks skip the tick with zero HTTP claim round-trips; fails open so a
DB hiccup degrades to the old behavior.

* chore(docs): reflow hard-wrapped prose inherited from the six-PR merge train

* chore(foundation): regenerate lifecycle artifacts; reflow inherited prose

---------

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

.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/<n>/merge` for pull_request
+# vs `refs/heads/<branch>` 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

.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/<n>/merge` for pull_request
+# vs `refs/heads/<branch>` 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 }})

.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/<n>/merge` for pull_request
+# vs `refs/heads/<branch>` 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)

.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/<n>/merge` for pull_request
+# vs `refs/heads/<branch>` 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)

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)
 

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,

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

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"])