feat(runtime): C3 tunable reaper threshold + heartbeat on every verb dispatch

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.
This commit is contained in:
Renn F
2026-05-12 05:10:32 +02:00
parent eb9cd93e09
commit a47237416e
5 changed files with 287 additions and 13 deletions
@@ -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"
@@ -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)