From b61148cdc91ad2c0898fddd2b49c7e3b5a4ff705 Mon Sep 17 00:00:00 2001 From: Renn F Date: Thu, 18 Jun 2026 11:37:40 +0200 Subject: [PATCH] feat(grok): reaper watchdog kills wedged opencode containers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The heartbeat reaper deliberately skips a task whose assignee holds a live ACTIVE container, so a Claude agent deep in a long edit/test cycle isn't churned out from under live work. A wedged opencode container breaks that assumption: it stays ACTIVE while firing no gateway verb, so its heartbeat never advances and the live-instance skip would shield its task forever — the exact way the Grok pr_reviewer parked in_progress. Add a longer grok-idle kill threshold (ROBOCO_GROK_IDLE_KILL_SECONDS, default 900s, well past the stream chunk timeout). A GROK instance idle past it is force-removed (its logs dumped to disk first) and evicted from the instance registry, so the same reaper pass then releases the task. Only GROK runtimes are eligible — a quiet Claude agent keeps the heartbeat-skip protection. --- roboco/config.py | 16 +++ roboco/runtime/orchestrator.py | 63 +++++++++- tests/unit/runtime/test_stale_claim_reaper.py | 118 ++++++++++++++++++ 3 files changed, 196 insertions(+), 1 deletion(-) diff --git a/roboco/config.py b/roboco/config.py index 63594c57..bb50ceb0 100644 --- a/roboco/config.py +++ b/roboco/config.py @@ -662,6 +662,22 @@ class Settings(BaseSettings): "override via ROBOCO_STALE_CLAIM_REAP_SECONDS" ), ) + # A GROK (opencode) agent that wedges — an idle model call / stream with no + # gateway verb — is ACTIVE-yet-silent, so the heartbeat reaper's live- + # container skip would shield its task forever (opencode emits no SDK budget + # signal and advances no heartbeat while parked, unlike a Claude agent that + # at least reports). After this longer window the orchestrator kills + evicts + # the container so the reaper releases the task. Longer than + # stale_claim_reap_seconds so only a truly-dead run trips it, never a + # slow-but-working agent. + grok_idle_kill_seconds: int = Field( + default=900, + ge=120, + description=( + "Idle-container kill threshold for GROK agents (seconds); " + "override via ROBOCO_GROK_IDLE_KILL_SECONDS" + ), + ) # A task left CLAIMED/IN_PROGRESS with an assignee but no running container # (e.g. a reassignment that didn't spawn) is invisibly stuck — the heartbeat # reaper can't see it because its heartbeat was seeded fresh at claim time. diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 1f6ee157..1b76f985 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -682,6 +682,10 @@ class AgentOrchestrator: # 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 + # Longer threshold before a wedged (ACTIVE-yet-idle) GROK container is + # killed + evicted so the reaper can release its task; see + # _maybe_kill_wedged_grok. + self._grok_idle_kill_ttl: int = settings.grok_idle_kill_seconds # ========================================================================= # LIFECYCLE @@ -6245,6 +6249,56 @@ Start now: evidence(task_id="{task_id}") instance = instances.get(self._resolve_agent_slug(str(owner))) return instance is not None and instance.state == AgentState.ACTIVE + async def _maybe_kill_wedged_grok( + self, task: Any, last_heartbeat: "datetime | None" + ) -> bool: + """Kill + evict a GROK container that is ACTIVE but idle past the kill TTL. + + ``_assignee_has_active_instance`` shields a live container from the + reaper — correct for a Claude agent that goes quiet during a long + edit/test cycle. A wedged opencode container is the one case that breaks: + it is ACTIVE *and* silent (an idle model call/stream fires no gateway + verb), so its heartbeat never advances and the skip would protect it + forever. Only a GROK instance idle past the longer grok-kill TTL is + eligible here — a Claude agent is never touched. On a kill the container + is removed (its logs dumped to disk first) and dropped from + ``_instances`` so this tick's reaper then releases the task. Returns True + only when a container was actually killed. + """ + from roboco.models.base import ModelProvider + + kill_ttl = getattr(self, "_grok_idle_kill_ttl", 900) + cutoff = datetime.now(UTC) - timedelta(seconds=kill_ttl) + if last_heartbeat is not None and last_heartbeat >= cutoff: + return False + owner = getattr(task, "assigned_to", None) or getattr(task, "claimed_by", None) + if not owner: + return False + slug = self._resolve_agent_slug(str(owner)) + instance = (getattr(self, "_instances", None) or {}).get(slug) + if instance is None or instance.state != AgentState.ACTIVE: + return False + config = instance.config + if config is None or config.provider_type != ModelProvider.GROK.value: + return False + try: + await self._remove_container(f"roboco-agent-{slug}") + except Exception as exc: + logger.error( + "wedged-grok kill failed; will retry next tick", + agent_id=slug, + error=str(exc), + ) + return False + self._instances.pop(slug, None) + logger.warning( + "wedged grok container killed and evicted", + agent_id=slug, + task_id=str(getattr(task, "id", "")), + idle_kill_ttl_s=kill_ttl, + ) + return True + async def _reap_with_service(self, svc: "TaskService") -> None: """Inner reap loop, parameterized by the TaskService to use. @@ -6261,7 +6315,14 @@ Start now: evidence(task_id="{task_id}") for t in candidates: ts = t.last_heartbeat_at if ts is None or ts < cutoff: - if self._assignee_has_active_instance(t): + # A live container normally protects its task. The sole exception + # is a wedged GROK (opencode) container — ACTIVE yet firing no + # verb — which the live-instance skip would shield forever. Kill + + # evict it past the grok-idle TTL (then fall through to release); + # a live non-grok agent, or a grok within the TTL, is skipped. + if self._assignee_has_active_instance( + t + ) and not await self._maybe_kill_wedged_grok(t, ts): continue task_id = require_uuid(t.id) try: diff --git a/tests/unit/runtime/test_stale_claim_reaper.py b/tests/unit/runtime/test_stale_claim_reaper.py index 0248e64c..b29e629d 100644 --- a/tests/unit/runtime/test_stale_claim_reaper.py +++ b/tests/unit/runtime/test_stale_claim_reaper.py @@ -140,3 +140,121 @@ async def test_reap_spares_claims_whose_assignee_container_is_alive() -> None: # The live-assignee task is spared; only the dead one is reaped. svc.unclaim_for_reaper.assert_awaited_once_with(dead_id) + + +def _grok_instance() -> AgentInstance: + cfg = type("C", (), {"provider_type": "grok"})() + return AgentInstance(agent_id="be-dev-1", state=AgentState.ACTIVE, config=cfg) + + +@pytest.mark.asyncio +async def test_reaper_kills_and_releases_wedged_grok_container() -> None: + """A GROK container idle past the kill TTL is killed, evicted, and released. + + Unlike a Claude agent, a wedged opencode container is ACTIVE yet fires no + verb, so the live-instance skip would shield it forever. Past the longer + grok-idle TTL the watchdog removes the container and drops it from + `_instances`, so the same reap pass then unclaims the task. + """ + now = datetime.now(UTC) + task_id = uuid4() + wedged = type( + "T", + (), + { + "id": task_id, + "last_heartbeat_at": now - timedelta(seconds=1200), + "assigned_to": AGENT_UUIDS["be-dev-1"], + "claimed_by": None, + }, + )() + + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._claim_heartbeat_ttl = 300 + orch._grok_idle_kill_ttl = 900 + orch._instances = {"be-dev-1": _grok_instance()} + orch._remove_container = AsyncMock() + svc = AsyncMock() + svc.list_in_progress_or_claimed.return_value = [wedged] + svc.unclaim_for_reaper = AsyncMock() + + await orch._reap_with_service(svc) + + orch._remove_container.assert_awaited_once_with("roboco-agent-be-dev-1") + assert "be-dev-1" not in orch._instances # evicted + svc.unclaim_for_reaper.assert_awaited_once_with(task_id) # released + + +@pytest.mark.asyncio +async def test_reaper_spares_grok_container_within_kill_ttl() -> None: + """A GROK container stale past the claim TTL but within the kill TTL lives. + + Only a truly-dead run (idle past the longer grok-idle TTL) is killed — a + slow-but-working agent is left alone. + """ + now = datetime.now(UTC) + recent = type( + "T", + (), + { + "id": uuid4(), + "last_heartbeat_at": now - timedelta(seconds=600), + "assigned_to": AGENT_UUIDS["be-dev-1"], + "claimed_by": None, + }, + )() + + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._claim_heartbeat_ttl = 300 + orch._grok_idle_kill_ttl = 900 + orch._instances = {"be-dev-1": _grok_instance()} + orch._remove_container = AsyncMock() + svc = AsyncMock() + svc.list_in_progress_or_claimed.return_value = [recent] + svc.unclaim_for_reaper = AsyncMock() + + await orch._reap_with_service(svc) + + orch._remove_container.assert_not_awaited() + assert "be-dev-1" in orch._instances + svc.unclaim_for_reaper.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_reaper_never_kills_non_grok_container() -> None: + """A non-GROK (Claude) container idle past the kill TTL is still spared. + + The watchdog only kills GROK runtimes; a quiet Claude agent keeps the + heartbeat-skip protection regardless of how long it has been silent. + """ + now = datetime.now(UTC) + claude_task = type( + "T", + (), + { + "id": uuid4(), + "last_heartbeat_at": now - timedelta(seconds=1200), + "assigned_to": AGENT_UUIDS["be-dev-1"], + "claimed_by": None, + }, + )() + claude_cfg = type("C", (), {"provider_type": "anthropic"})() + + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._claim_heartbeat_ttl = 300 + orch._grok_idle_kill_ttl = 900 + orch._instances = { + "be-dev-1": AgentInstance( + agent_id="be-dev-1", state=AgentState.ACTIVE, config=claude_cfg + ) + } + orch._remove_container = AsyncMock() + svc = AsyncMock() + svc.list_in_progress_or_claimed.return_value = [claude_task] + svc.unclaim_for_reaper = AsyncMock() + + await orch._reap_with_service(svc) + + orch._remove_container.assert_not_awaited() + assert "be-dev-1" in orch._instances + svc.unclaim_for_reaper.assert_not_awaited()