From a47237416e9c868a23bb4c6d966ded6475ac9236 Mon Sep 17 00:00:00 2001 From: Renn F Date: Tue, 12 May 2026 05:10:32 +0200 Subject: [PATCH] feat(runtime): C3 tunable reaper threshold + heartbeat on every verb dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Smoke run 3 showed agents reaped at the 3-min stale-claim window while they were actively retrying rejected verbs. Two causes: 1. The reaper threshold was hardcoded at 180s via claim_stale_seconds. LLM inference + retry loops routinely take longer than that between verb-successes. Added settings.stale_claim_reap_seconds (default 600s); override via ROBOCO_STALE_CLAIM_REAP_SECONDS env var. claim_stale_seconds (spawn-filter cutoff) is unchanged at 180s. 2. last_heartbeat_at only refreshed on verb SUCCESS. A verb stuck in a rejection loop (e.g. tracing_gap missing journal:decision) showed no heartbeat updates even though the agent was alive. Added a best-effort heartbeat refresh inside _emit_rejection so EVERY verb dispatch — success or rejection — counts as activity. Heartbeat approach: option (b) — touch inside _emit_rejection (single centralized rejection path). Requires no middleware layer, no HTTP body parsing, and no new files. The _touch guard for task_id=None means agent-level rejections (no task context) are a safe no-op. Net effect: agents stop being reaped mid-retry. Genuinely-stuck containers (no verb dispatch at all) still reap normally at 600s. Spec ref: Wave C Task C3. --- roboco/config.py | 16 ++ roboco/runtime/orchestrator.py | 18 +- .../services/gateway/choreographer/_impl.py | 17 +- .../gateway/test_heartbeat_on_rejection.py | 157 ++++++++++++++++++ .../runtime/test_heartbeat_reaper_tunable.py | 92 ++++++++++ 5 files changed, 287 insertions(+), 13 deletions(-) create mode 100644 tests/unit/gateway/test_heartbeat_on_rejection.py create mode 100644 tests/unit/runtime/test_heartbeat_reaper_tunable.py diff --git a/roboco/config.py b/roboco/config.py index 6684e4f5..103e12c9 100644 --- a/roboco/config.py +++ b/roboco/config.py @@ -326,6 +326,22 @@ class Settings(BaseSettings): ge=60, description="Claim heartbeat staleness threshold (seconds)", ) + # Wave C3 (2026-05-12). Reaper window for stale-claim detection. + # Smoke run 3 reaped agents at ~180s while they were actively + # retrying — LLM inference + retry loops routinely exceed 3 min + # between verb successes. 600s is large enough to accommodate that + # without letting a genuinely-stuck container linger. + # Distinct from claim_stale_seconds (which drives trigger_filter + # spawn queueing); keeping them separate avoids a window where a + # higher reap threshold would also delay spawn-queue decisions. + stale_claim_reap_seconds: int = Field( + default=600, + ge=60, + description=( + "Reaper-only stale claim threshold (seconds); " + "override via ROBOCO_STALE_CLAIM_REAP_SECONDS" + ), + ) spawn_cooldown_seconds: int = Field( default=60, ge=1, diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 63c2e86e..dfbe29ac 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -499,14 +499,16 @@ class AgentOrchestrator: # is in a loop — without this gate the orchestrator re-spawns every # tick forever (seen in production on 2026-04-22). self._pm_respawn_tracker: dict[tuple[str, str], dict[str, Any]] = {} - # Stale-claim reaper config. Sourced from the same setting that - # `trigger_filter` consumes so both layers agree on the staleness - # cutoff on the same dispatch tick — the reaper runs first, frees - # the row, and any spawn `trigger_filter` queues lands on an - # unclaimed task. Tests bypass `__init__` via `__new__` and bind - # 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 + # Stale-claim reaper config. Wave C3 (2026-05-12): sourced from + # stale_claim_reap_seconds (default 600) rather than + # claim_stale_seconds (default 180). The two settings are now + # distinct: claim_stale_seconds drives trigger_filter (spawn + # queueing); stale_claim_reap_seconds drives the reaper. + # Smoke run 3 showed agents reaped at 180s while actively retrying + # rejected verbs — LLM inference routinely exceeds that window. + # Tests bypass `__init__` via `__new__` and set _claim_heartbeat_ttl + # directly; production never uses _task_svc from __init__. + self._claim_heartbeat_ttl: int = settings.stale_claim_reap_seconds # ========================================================================= # LIFECYCLE diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 3e550345..22d17c84 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -465,6 +465,17 @@ class Choreographer: """ if env.error is None: return env + # Wave C3 (2026-05-12): refresh heartbeat on every rejection so an + # agent stuck in a verb-rejection loop (e.g., tracing_gap while + # retrying) does not look idle to the reaper. Best-effort: a + # heartbeat failure must never alter the envelope returned to the + # agent. _touch already guards task_id=None. + try: + await self._touch(task_id) + except Exception as exc: + logger.warning( + "heartbeat touch failed on rejection", error=str(exc), verb=verb + ) from uuid import uuid4 as _uuid4 details: dict[str, Any] = { @@ -2291,11 +2302,7 @@ class Choreographer: context_briefing={}, ) # Rule 2: same-assignee same-type fallback (non-spine types) - if ( - sib_assignee - and sib_assignee == new_assignee - and sib_type == new_type - ): + if sib_assignee and sib_assignee == new_assignee and sib_type == new_type: return Envelope.invalid_state( message=( f"sibling subtask already assigned to " diff --git a/tests/unit/gateway/test_heartbeat_on_rejection.py b/tests/unit/gateway/test_heartbeat_on_rejection.py new file mode 100644 index 00000000..28c151a5 --- /dev/null +++ b/tests/unit/gateway/test_heartbeat_on_rejection.py @@ -0,0 +1,157 @@ +"""Wave C3 (2026-05-12): _emit_rejection must touch last_heartbeat_at. + +Pre-fix: heartbeat only refreshed on verb SUCCESS. Smoke run 3 showed +agents being reaped while actively retrying rejected verbs — the +heartbeat was stale even though the agent was alive and calling verbs. + +Fix: _emit_rejection calls self._touch(task_id) for every rejection so +the agent's heartbeat stays current even in a pure-rejection loop. + +Constraints (from spec): + - The touch must NOT fire on success envelopes (_emit_rejection already + short-circuits on env.error is None — this is unchanged). + - The touch is best-effort: a task.heartbeat() failure must not change + the envelope returned to the agent. + - task_id=None is safe: _touch already guards that case. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps + + +def _make_deps(**overrides: Any) -> ChoreographerDeps: + base: dict[str, Any] = { + "task": AsyncMock(), + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + base.update(overrides) + repo = base["evidence_repo"] + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(repo, method).return_value = [] + task = base["task"] + task.session = MagicMock() + task.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + return ChoreographerDeps(**base) + + +# --------------------------------------------------------------------------- +# Core acceptance: rejection on a role-mismatch fires heartbeat +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_heartbeat_fires_on_rejection_not_authorized() -> None: + """A not_authorized rejection (PM trying code task) must still touch heartbeat.""" + aid = uuid4() + tid = uuid4() + code_task = MagicMock( + id=tid, + status="pending", + assigned_to=None, + task_type="code", + priority=1, + parent_task_id=None, + sequence=0, + team="backend", + ) + task_svc = AsyncMock() + task_svc.get.return_value = code_task + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.get_subtasks.return_value = [] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_work_on(aid, tid, plan="x") + + assert env.error == "not_authorized" + task_svc.heartbeat.assert_awaited_with(tid) + + +# --------------------------------------------------------------------------- +# not_found rejection also touches heartbeat (task_id still known from arg) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_heartbeat_fires_on_not_found_rejection() -> None: + """not_found rejection passes task_id to _emit_rejection; heartbeat still fires.""" + aid = uuid4() + tid = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = None # task not found + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_am_done(aid, tid, notes="something") + + assert env.error == "not_found" + task_svc.heartbeat.assert_awaited_with(tid) + + +# --------------------------------------------------------------------------- +# Success path must NOT get an extra heartbeat from _emit_rejection +# (the existing _touch calls on the success path already cover it) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_heartbeat_not_double_fired_on_success() -> None: + """give_me_work idle-path succeeds without touching heartbeat at all.""" + aid = uuid4() + task_svc = AsyncMock() + task_svc.list_assigned_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.give_me_work(aid) + + assert env.error is None + task_svc.heartbeat.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# Heartbeat failure on rejection must not swallow or alter the envelope +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_heartbeat_failure_on_rejection_does_not_propagate() -> None: + """If task.heartbeat() raises during a rejection, the envelope is still returned.""" + aid = uuid4() + tid = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = None + task_svc.heartbeat.side_effect = RuntimeError("DB down") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_am_done(aid, tid, notes="x") + + assert env.error == "not_found" diff --git a/tests/unit/runtime/test_heartbeat_reaper_tunable.py b/tests/unit/runtime/test_heartbeat_reaper_tunable.py new file mode 100644 index 00000000..bbf370ab --- /dev/null +++ b/tests/unit/runtime/test_heartbeat_reaper_tunable.py @@ -0,0 +1,92 @@ +"""Wave C3 (2026-05-12): heartbeat reaper threshold honors config setting. + +The reaper previously used a hardcoded 180s (= claim_stale_seconds) which +was also the spawn-filter cutoff. Smoke run 3 showed agents being reaped +at ~3 minutes while actively retrying rejected verbs; LLM inference alone +can exceed that window. + +Two fixes ship together: + - stale_claim_reap_seconds (default 600) drives the reaper; the + spawn-filter keeps claim_stale_seconds unchanged. + - _emit_rejection now touches last_heartbeat_at so every verb attempt + (success or rejection) counts as agent activity. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from roboco.config import settings +from roboco.runtime.orchestrator import AgentOrchestrator + +_EXPECTED_DEFAULT_REAP_SECONDS = 600 + + +@pytest.mark.asyncio +async def test_reaper_default_threshold_is_600() -> None: + """settings.stale_claim_reap_seconds defaults to 600, not the old 180.""" + assert settings.stale_claim_reap_seconds == _EXPECTED_DEFAULT_REAP_SECONDS + + +@pytest.mark.asyncio +async def test_orchestrator_init_uses_stale_claim_reap_seconds() -> None: + """_claim_heartbeat_ttl is sourced from stale_claim_reap_seconds, + not claim_stale_seconds.""" + # Build an orchestrator via __new__ to avoid touching the DB, then check + # the field that _reap_with_service reads. + orch = AgentOrchestrator.__new__(AgentOrchestrator) + # Manually run the subset of __init__ that sets _claim_heartbeat_ttl. + orch._claim_heartbeat_ttl = settings.stale_claim_reap_seconds + assert orch._claim_heartbeat_ttl == _EXPECTED_DEFAULT_REAP_SECONDS + + +@pytest.mark.asyncio +async def test_reaper_does_not_reap_under_custom_threshold() -> None: + """Task whose heartbeat is 800s old is NOT reaped when threshold=900.""" + now = datetime.now(UTC) + task_safe = type( + "T", + (), + { + "id": uuid4(), + "last_heartbeat_at": now - timedelta(seconds=800), + }, + )() + + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._claim_heartbeat_ttl = 900 # custom threshold + svc = AsyncMock() + svc.list_in_progress_or_claimed.return_value = [task_safe] + svc.unclaim_for_reaper = AsyncMock() + + await orch._reap_with_service(svc) + + svc.unclaim_for_reaper.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_reaper_does_reap_over_custom_threshold() -> None: + """Task whose heartbeat is 1000s old IS reaped when threshold=900.""" + now = datetime.now(UTC) + stale_id = uuid4() + task_stale = type( + "T", + (), + { + "id": stale_id, + "last_heartbeat_at": now - timedelta(seconds=1000), + }, + )() + + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._claim_heartbeat_ttl = 900 + svc = AsyncMock() + svc.list_in_progress_or_claimed.return_value = [task_stale] + svc.unclaim_for_reaper = AsyncMock() + + await orch._reap_with_service(svc) + + svc.unclaim_for_reaper.assert_awaited_once_with(stale_id)