diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index ea8743f6..5d2d1725 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -143,6 +143,13 @@ _SHUTDOWN_DRAIN_TIMEOUT_SECONDS = 5.0 # to whatever exit the monitor is now looking at, rather than mis-attributed. _EXPECTED_STOP_FRESH_SECONDS = 120.0 _EXPECTED_STOP_MAX_ENTRIES = 200 +# _route_unassigned_pm_task's creator-skip guard: a PM that just created a task +# is about to assign it one tool-call later, so racing in and claiming it for +# the PM would hijack the delegation. That's only true for a few seconds — +# past this grace the creator's session is long gone and the skip would hold +# the task pending-unassigned forever. Generous enough to cover the PM's next +# tool call; short enough that a genuinely abandoned task recovers fast. +_CREATOR_ROUTE_GRACE_SECONDS = 600 _HTTP_TOO_MANY_REQUESTS = 429 _HTTP_OK = 200 _HTTP_MULTIPLE_CHOICES = 300 # first non-2xx status; 2xx == [_HTTP_OK, this) @@ -14019,6 +14026,42 @@ Start now: evidence(task_id="{task_id}") ) return False + def _creator_route_should_skip( + self, task: dict[str, Any], agent_id: str, routing: str + ) -> bool: + """The creator-skip guard for ``_route_unassigned_pm_task``. + + A PM that just created this task is about to assign it (e.g. be-pm + creating a code subtask to hand to be-dev-1 one tool-call later); + racing in and claiming for the PM would hijack that delegation. True + only while the task is still within ``_CREATOR_ROUTE_GRACE_SECONDS`` + of creation — past that the creator's session is long gone (it + exited without assigning), so this falls through (False) to normal + routing instead of skipping forever. Fails open on an + unparseable/missing ``created_at`` (treated as OLD) — routing is the + safe default, the skip is only an optimization. + """ + created_by = task.get("created_by") + if not created_by or self._resolve_agent_slug(str(created_by)) != agent_id: + return False + age = self._get_task_age(task) + if age is not None and age.total_seconds() < _CREATOR_ROUTE_GRACE_SECONDS: + logger.info( + "Skipping auto-claim: routing target is the creator", + task_id=task.get("id"), + creator=agent_id, + routing=routing, + ) + return True + logger.info( + "Creator-skip grace elapsed; routing task to its creator", + task_id=task.get("id"), + creator=agent_id, + routing=routing, + age_seconds=None if age is None else int(age.total_seconds()), + ) + return False + async def _route_unassigned_pm_task( self, client: httpx.AsyncClient, task: dict[str, Any] ) -> None: @@ -14046,24 +14089,10 @@ Start now: evidence(task_id="{task_id}") await self._handle_board_assigned_task(task, agent_id) return - # Don't auto-claim back to the creator. A PM that just created this - # task is about to assign it (e.g. be-pm creating a code subtask to - # hand to be-dev-1 one tool-call later). Racing in and claiming for - # the PM hijacks the delegation — the PM ends up owning a code task - # it never intended to work on itself. Skip this tick and let the - # next dispatch pick it up once assigned_to is set, OR re-evaluate - # when we have a clearer signal the creator won't route it. - created_by = task.get("created_by") - if created_by: - creator_slug = self._resolve_agent_slug(str(created_by)) - if creator_slug == agent_id: - logger.info( - "Skipping auto-claim: routing target is the creator", - task_id=task.get("id"), - creator=creator_slug, - routing=routing, - ) - return + # Don't auto-claim back to the creator while the task is fresh — see + # _creator_route_should_skip. + if self._creator_route_should_skip(task, agent_id, routing): + return logger.info( "Routing task", diff --git a/tests/unit/runtime/test_pm_dispatch_claim_prefilter.py b/tests/unit/runtime/test_pm_dispatch_claim_prefilter.py index 9ba9ac51..351a3053 100644 --- a/tests/unit/runtime/test_pm_dispatch_claim_prefilter.py +++ b/tests/unit/runtime/test_pm_dispatch_claim_prefilter.py @@ -14,12 +14,13 @@ round-tripping the claim endpoint. from __future__ import annotations from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta 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 +from roboco.runtime.orchestrator import _CREATOR_ROUTE_GRACE_SECONDS, AgentOrchestrator if TYPE_CHECKING: from collections.abc import AsyncIterator @@ -96,6 +97,124 @@ async def test_ready_task_still_dispatches() -> None: spawn.assert_awaited_once() +# --------------------------------------------------------------------------- +# _creator_route_should_skip / _route_unassigned_pm_task — the creator-skip +# guard is age-gated: young means "the creator-PM is about to assign it one +# tool-call later," old means the creator's session is long gone and the +# skip would otherwise wedge the task pending-unassigned forever. +# --------------------------------------------------------------------------- + + +def _iso_age(seconds: float) -> str: + return (datetime.now(UTC) - timedelta(seconds=seconds)).isoformat() + + +async def _route_with_creator_task( + orch: AgentOrchestrator, task: dict[str, Any] +) -> Any: + """Run `_route_unassigned_pm_task` with routing/claim/spawn stubbed so only + the creator-skip guard's own age logic decides the outcome. Returns the + (claim, spawn) mocks for assertion.""" + 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="cell_pm"), + patch.object(orch, "_get_routing_target", return_value="be-pm"), + patch.object(orch, "_resolve_agent_slug", return_value="be-pm"), + 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) + return claim, spawn + + +@pytest.mark.asyncio +async def test_young_creator_task_still_skipped() -> None: + """(a) Fresh task, creator == routing target: skip, no claim/spawn.""" + orch = _orch() + task = _pending_task(created_by=str(uuid4()), created_at=_iso_age(30)) + + claim, spawn = await _route_with_creator_task(orch, task) + + claim.assert_not_awaited() + spawn.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_old_creator_task_falls_through_to_routing() -> None: + """(b) 11+ minutes past creation, creator == routing target: the grace + has elapsed, so normal claim + spawn proceeds (the creator's session + exited without assigning — self-claim is safe for PM-routed work).""" + orch = _orch() + task = _pending_task( + created_by=str(uuid4()), + created_at=_iso_age(_CREATOR_ROUTE_GRACE_SECONDS + 60), + ) + + claim, spawn = await _route_with_creator_task(orch, task) + + claim.assert_awaited_once() + spawn.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_unparseable_created_at_fails_open_to_routing() -> None: + """(c) A garbage created_at can't be aged, so the guard fails open + (treats it as OLD) rather than skip forever.""" + orch = _orch() + task = _pending_task(created_by=str(uuid4()), created_at="not-a-timestamp") + + claim, spawn = await _route_with_creator_task(orch, task) + + claim.assert_awaited_once() + spawn.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_missing_created_at_fails_open_to_routing() -> None: + """(c) No created_at at all — same fail-open treatment as unparseable.""" + orch = _orch() + task = _pending_task(created_by=str(uuid4())) + task.pop("created_at", None) + + claim, spawn = await _route_with_creator_task(orch, task) + + claim.assert_awaited_once() + spawn.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_different_creator_dispatches_regardless_of_age() -> None: + """(d) creator != routing target: the guard never engages, so behavior is + unchanged whether the task is brand new or ancient.""" + orch = _orch() + for age_seconds in (5, _CREATOR_ROUTE_GRACE_SECONDS + 3600): + task = _pending_task(created_by=str(uuid4()), created_at=_iso_age(age_seconds)) + 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="cell_pm"), + patch.object(orch, "_get_routing_target", return_value="be-pm"), + patch.object(orch, "_resolve_agent_slug", return_value="main-pm"), + 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() + spawn.assert_awaited_once() + + # --------------------------------------------------------------------------- # _pending_claim_blocked — the DB-backed probe itself # ---------------------------------------------------------------------------