mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
fix(orchestrator): consolidate stale-heartbeat config + drop dead _task_svc slot
I1: claim_heartbeat_ttl_seconds (300s) overlapped semantically with the pre-existing claim_stale_seconds (180s). Between 180-300s of silence, trigger_filter queued duplicate spawns while the reaper hadn't yet released the claim — exactly the dispatcher churn the reaper was supposed to close. Collapse to one field (claim_stale_seconds, 180s); reaper now consumes the same setting trigger_filter uses, so both agree on 'stale' on the same tick and the reaper runs first. I2: _task_svc injection slot on AgentOrchestrator.__init__ was production-dead (always None) and only used by __new__-based test instances. Drop the __init__ slot + the production branch in _reap_stale_claims that read it. Tests still pre-bind on __new__ instances; the attribute exists per-instance, not per-class.
This commit is contained in:
+8
-8
@@ -300,19 +300,19 @@ class Settings(BaseSettings):
|
|||||||
)
|
)
|
||||||
|
|
||||||
# Gateway coordination thresholds
|
# Gateway coordination thresholds
|
||||||
|
# Single source of truth for "claim heartbeat is stale": consumed both by
|
||||||
|
# `trigger_filter` (deciding whether to QUEUE a fresh spawn) and by
|
||||||
|
# `_reap_stale_claims` (deciding whether to RELEASE the claim back to
|
||||||
|
# pending). Keeping them on one field guarantees both layers agree on
|
||||||
|
# the same tick — the reaper runs first, releases the row, and the
|
||||||
|
# queued spawn finds an unclaimed task. Splitting them into two fields
|
||||||
|
# opens a window where trigger_filter queues duplicate spawns against a
|
||||||
|
# claim the reaper hasn't yet released — pure dispatcher churn.
|
||||||
claim_stale_seconds: int = Field(
|
claim_stale_seconds: int = Field(
|
||||||
default=180,
|
default=180,
|
||||||
ge=60,
|
ge=60,
|
||||||
description="Claim heartbeat staleness threshold (seconds)",
|
description="Claim heartbeat staleness threshold (seconds)",
|
||||||
)
|
)
|
||||||
claim_heartbeat_ttl_seconds: int = Field(
|
|
||||||
default=300,
|
|
||||||
ge=60,
|
|
||||||
description=(
|
|
||||||
"If an agent's last_heartbeat_at is older than this, the dispatcher"
|
|
||||||
" considers the claim dead and releases the task back to pending."
|
|
||||||
),
|
|
||||||
)
|
|
||||||
spawn_cooldown_seconds: int = Field(
|
spawn_cooldown_seconds: int = Field(
|
||||||
default=60,
|
default=60,
|
||||||
ge=1,
|
ge=1,
|
||||||
|
|||||||
@@ -451,12 +451,14 @@ class AgentOrchestrator:
|
|||||||
# is in a loop — without this gate the orchestrator re-spawns every
|
# is in a loop — without this gate the orchestrator re-spawns every
|
||||||
# tick forever (seen in production on 2026-04-22).
|
# tick forever (seen in production on 2026-04-22).
|
||||||
self._pm_respawn_tracker: dict[tuple[str, str], dict[str, Any]] = {}
|
self._pm_respawn_tracker: dict[tuple[str, str], dict[str, Any]] = {}
|
||||||
# Stale-claim reaper config + injectable service slot. Production
|
# Stale-claim reaper config. Sourced from the same setting that
|
||||||
# code leaves `_task_svc` None so the reaper opens its own per-tick
|
# `trigger_filter` consumes so both layers agree on the staleness
|
||||||
# session; tests can pre-set a mock on an instance built via
|
# cutoff on the same dispatch tick — the reaper runs first, frees
|
||||||
# `__new__` to bypass this dance.
|
# the row, and any spawn `trigger_filter` queues lands on an
|
||||||
self._claim_heartbeat_ttl: int = settings.claim_heartbeat_ttl_seconds
|
# unclaimed task. Tests bypass `__init__` via `__new__` and bind
|
||||||
self._task_svc: TaskService | None = None
|
# a mock `_task_svc` on the instance directly; production never
|
||||||
|
# uses that attribute, so it's not initialized here.
|
||||||
|
self._claim_heartbeat_ttl: int = settings.claim_stale_seconds
|
||||||
|
|
||||||
# =========================================================================
|
# =========================================================================
|
||||||
# LIFECYCLE
|
# LIFECYCLE
|
||||||
@@ -3539,16 +3541,13 @@ Start now: evidence(task_id="{task_id}")
|
|||||||
here in the orchestrator; the actual UPDATE statements live in
|
here in the orchestrator; the actual UPDATE statements live in
|
||||||
``TaskService.unclaim_for_reaper``.
|
``TaskService.unclaim_for_reaper``.
|
||||||
|
|
||||||
If an instance has a pre-bound ``_task_svc`` (used by tests and
|
Opens a fresh per-tick session — short-lived because the reaper
|
||||||
conceivable future caching), use it directly. Otherwise open a
|
runs on every dispatch cycle and the work is cheap (one SELECT
|
||||||
per-tick session — short-lived because the reaper runs on every
|
plus N UPDATEs for the typically-empty stale set). Tests that
|
||||||
dispatch cycle and the work is cheap (one SELECT plus N UPDATEs
|
need to inject a mock service do so by building an instance via
|
||||||
for the typically-empty stale set).
|
``__new__`` (bypassing this method) and calling
|
||||||
|
``_reap_with_service`` directly.
|
||||||
"""
|
"""
|
||||||
if self._task_svc is not None:
|
|
||||||
await self._reap_with_service(self._task_svc)
|
|
||||||
return
|
|
||||||
|
|
||||||
from roboco.db.base import get_session_factory
|
from roboco.db.base import get_session_factory
|
||||||
from roboco.services.task import TaskService
|
from roboco.services.task import TaskService
|
||||||
|
|
||||||
|
|||||||
@@ -40,13 +40,13 @@ async def test_reap_stale_claims_releases_dead_holders() -> None:
|
|||||||
|
|
||||||
orch = AgentOrchestrator.__new__(AgentOrchestrator) # bypass __init__
|
orch = AgentOrchestrator.__new__(AgentOrchestrator) # bypass __init__
|
||||||
orch._claim_heartbeat_ttl = 300
|
orch._claim_heartbeat_ttl = 300
|
||||||
orch._task_svc = AsyncMock()
|
svc = AsyncMock()
|
||||||
orch._task_svc.list_in_progress_or_claimed.return_value = [stale_task, fresh_task]
|
svc.list_in_progress_or_claimed.return_value = [stale_task, fresh_task]
|
||||||
orch._task_svc.unclaim_for_reaper = AsyncMock()
|
svc.unclaim_for_reaper = AsyncMock()
|
||||||
|
|
||||||
await orch._reap_stale_claims()
|
await orch._reap_with_service(svc)
|
||||||
|
|
||||||
orch._task_svc.unclaim_for_reaper.assert_awaited_once_with(stale_id)
|
svc.unclaim_for_reaper.assert_awaited_once_with(stale_id)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
@@ -57,13 +57,13 @@ async def test_reap_stale_claims_releases_holders_with_null_heartbeat() -> None:
|
|||||||
|
|
||||||
orch = AgentOrchestrator.__new__(AgentOrchestrator)
|
orch = AgentOrchestrator.__new__(AgentOrchestrator)
|
||||||
orch._claim_heartbeat_ttl = 300
|
orch._claim_heartbeat_ttl = 300
|
||||||
orch._task_svc = AsyncMock()
|
svc = AsyncMock()
|
||||||
orch._task_svc.list_in_progress_or_claimed.return_value = [null_task]
|
svc.list_in_progress_or_claimed.return_value = [null_task]
|
||||||
orch._task_svc.unclaim_for_reaper = AsyncMock()
|
svc.unclaim_for_reaper = AsyncMock()
|
||||||
|
|
||||||
await orch._reap_stale_claims()
|
await orch._reap_with_service(svc)
|
||||||
|
|
||||||
orch._task_svc.unclaim_for_reaper.assert_awaited_once_with(null_id)
|
svc.unclaim_for_reaper.assert_awaited_once_with(null_id)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
@@ -81,14 +81,14 @@ async def test_reap_stale_claims_swallows_unclaim_errors() -> None:
|
|||||||
|
|
||||||
orch = AgentOrchestrator.__new__(AgentOrchestrator)
|
orch = AgentOrchestrator.__new__(AgentOrchestrator)
|
||||||
orch._claim_heartbeat_ttl = 300
|
orch._claim_heartbeat_ttl = 300
|
||||||
orch._task_svc = AsyncMock()
|
svc = AsyncMock()
|
||||||
orch._task_svc.list_in_progress_or_claimed.return_value = [task_a, task_b]
|
svc.list_in_progress_or_claimed.return_value = [task_a, task_b]
|
||||||
orch._task_svc.unclaim_for_reaper = AsyncMock(
|
svc.unclaim_for_reaper = AsyncMock(
|
||||||
side_effect=[RuntimeError("transient"), None]
|
side_effect=[RuntimeError("transient"), None]
|
||||||
)
|
)
|
||||||
|
|
||||||
await orch._reap_stale_claims()
|
await orch._reap_with_service(svc)
|
||||||
|
|
||||||
# Both stale tasks attempted; second succeeded despite first raising.
|
# Both stale tasks attempted; second succeeded despite first raising.
|
||||||
expected_attempts = 2
|
expected_attempts = 2
|
||||||
assert orch._task_svc.unclaim_for_reaper.await_count == expected_attempts
|
assert svc.unclaim_for_reaper.await_count == expected_attempts
|
||||||
|
|||||||
Reference in New Issue
Block a user