mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
feat(grok): reaper watchdog kills wedged opencode containers
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.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user