mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
fix(reaper): don't reap a live container the instance registry forgot
The stale-claim reaper skips a task whose assignee holds an ACTIVE instance, but that check reads the in-memory _instances registry — lost on an orchestrator restart while the agent's container keeps running. The reaper then released a task out from under a live agent it had merely forgotten (registry amnesia). On a registry MISS, fall back to asking Docker directly (_inspect_container_state); a still-running container is spared. A known instance (active or stopped) stays authoritative, and an uninitialised registry (unit-test harness) keeps the prior behaviour.
This commit is contained in:
@@ -6749,6 +6749,35 @@ Start now: evidence(task_id="{task_id}")
|
|||||||
instance = instances.get(self._resolve_agent_slug(str(owner)))
|
instance = instances.get(self._resolve_agent_slug(str(owner)))
|
||||||
return instance is not None and instance.state == AgentState.ACTIVE
|
return instance is not None and instance.state == AgentState.ACTIVE
|
||||||
|
|
||||||
|
async def _assignee_container_running(self, task: Any) -> bool:
|
||||||
|
"""Docker-liveness fallback for the reaper on an instance-registry MISS.
|
||||||
|
|
||||||
|
``_assignee_has_active_instance`` reads the in-memory ``_instances``
|
||||||
|
registry, which is lost on an orchestrator restart while the agent's
|
||||||
|
container keeps running. Without a fallback the heartbeat-stale reaper
|
||||||
|
then releases a task out from under a live agent the orchestrator has
|
||||||
|
merely forgotten — registry amnesia, the over-reap that hit be-dev-1.
|
||||||
|
This asks Docker directly, but ONLY on a true registry miss: a known
|
||||||
|
instance (ACTIVE or stopped) is authoritative and not second-guessed,
|
||||||
|
and an uninitialised registry (``None`` — e.g. a unit-test harness) is
|
||||||
|
left to the existing behaviour. Any error (no docker binary, inspect
|
||||||
|
fails) yields False, so non-Docker test/dev contexts are unaffected.
|
||||||
|
"""
|
||||||
|
instances = getattr(self, "_instances", None)
|
||||||
|
if instances is None:
|
||||||
|
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))
|
||||||
|
if slug in instances:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
is_running, _ = await self._inspect_container_state(f"roboco-agent-{slug}")
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
return is_running
|
||||||
|
|
||||||
def _wedged_grok_slug(
|
def _wedged_grok_slug(
|
||||||
self, task: Any, last_heartbeat: "datetime | None"
|
self, task: Any, last_heartbeat: "datetime | None"
|
||||||
) -> str | None:
|
) -> str | None:
|
||||||
@@ -6828,14 +6857,19 @@ Start now: evidence(task_id="{task_id}")
|
|||||||
for t in candidates:
|
for t in candidates:
|
||||||
ts = t.last_heartbeat_at
|
ts = t.last_heartbeat_at
|
||||||
if ts is None or ts < cutoff:
|
if ts is None or ts < cutoff:
|
||||||
# A live container normally protects its task. The sole exception
|
# A live container normally protects its task. Prefer the
|
||||||
# is a wedged GROK container — ACTIVE yet firing no verb — which
|
# in-memory registry; on a registry MISS (e.g. the orchestrator
|
||||||
# the live-instance skip would shield forever. Kill +
|
# restarted and forgot a still-running container) fall back to
|
||||||
# evict it past the grok-idle TTL (then fall through to release);
|
# asking Docker, so we don't reap a task out from under a live
|
||||||
# a live non-grok agent, or a grok within the TTL, is skipped.
|
# agent. The sole exception is a wedged GROK container — ACTIVE
|
||||||
if self._assignee_has_active_instance(
|
# yet firing no verb — which the live 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.
|
||||||
|
live = self._assignee_has_active_instance(
|
||||||
t
|
t
|
||||||
) and not await self._maybe_kill_wedged_grok(t, ts):
|
) or await self._assignee_container_running(t)
|
||||||
|
if live and not await self._maybe_kill_wedged_grok(t, ts):
|
||||||
continue
|
continue
|
||||||
task_id = require_uuid(t.id)
|
task_id = require_uuid(t.id)
|
||||||
try:
|
try:
|
||||||
|
|||||||
@@ -267,3 +267,106 @@ async def test_reaper_never_kills_non_grok_container(
|
|||||||
remove_mock.assert_not_awaited()
|
remove_mock.assert_not_awaited()
|
||||||
assert "be-dev-1" in orch._instances
|
assert "be-dev-1" in orch._instances
|
||||||
svc.unclaim_for_reaper.assert_not_awaited()
|
svc.unclaim_for_reaper.assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_reap_spares_live_container_on_registry_miss(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Registry amnesia: the orchestrator forgot the instance (empty `_instances`,
|
||||||
|
e.g. after a restart) but the container is still running per Docker — the
|
||||||
|
task must NOT be reaped. This closes the be-dev-1 over-reap: a live agent the
|
||||||
|
orchestrator merely lost track of is no longer churned out from under work.
|
||||||
|
"""
|
||||||
|
now = datetime.now(UTC)
|
||||||
|
task_id = uuid4()
|
||||||
|
task = type(
|
||||||
|
"T",
|
||||||
|
(),
|
||||||
|
{
|
||||||
|
"id": task_id,
|
||||||
|
"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 = {} # registry lost; container still up
|
||||||
|
monkeypatch.setattr(
|
||||||
|
orch, "_inspect_container_state", AsyncMock(return_value=(True, None))
|
||||||
|
)
|
||||||
|
svc = AsyncMock()
|
||||||
|
svc.list_in_progress_or_claimed.return_value = [task]
|
||||||
|
svc.unclaim_for_reaper = AsyncMock()
|
||||||
|
|
||||||
|
await orch._reap_with_service(svc)
|
||||||
|
|
||||||
|
svc.unclaim_for_reaper.assert_not_awaited() # spared — Docker says it's alive
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_reap_releases_on_registry_miss_when_container_gone(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Registry miss AND Docker says the container is gone → genuinely dead → reaped."""
|
||||||
|
now = datetime.now(UTC)
|
||||||
|
task_id = uuid4()
|
||||||
|
task = type(
|
||||||
|
"T",
|
||||||
|
(),
|
||||||
|
{
|
||||||
|
"id": task_id,
|
||||||
|
"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 = {}
|
||||||
|
monkeypatch.setattr(
|
||||||
|
orch, "_inspect_container_state", AsyncMock(return_value=(False, 0))
|
||||||
|
)
|
||||||
|
svc = AsyncMock()
|
||||||
|
svc.list_in_progress_or_claimed.return_value = [task]
|
||||||
|
svc.unclaim_for_reaper = AsyncMock()
|
||||||
|
|
||||||
|
await orch._reap_with_service(svc)
|
||||||
|
|
||||||
|
svc.unclaim_for_reaper.assert_awaited_once_with(task_id)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_registry_uninitialised_skips_docker_fallback() -> None:
|
||||||
|
"""With `_instances` never initialised (None — the __new__ unit harness), the
|
||||||
|
Docker fallback is skipped and the stale task reaps as before; no accidental
|
||||||
|
Docker probing where there's no registry to be amnesiac about.
|
||||||
|
"""
|
||||||
|
now = datetime.now(UTC)
|
||||||
|
task_id = uuid4()
|
||||||
|
task = type(
|
||||||
|
"T",
|
||||||
|
(),
|
||||||
|
{
|
||||||
|
"id": task_id,
|
||||||
|
"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
|
||||||
|
# _instances intentionally NOT set -> getattr yields None -> no fallback.
|
||||||
|
svc = AsyncMock()
|
||||||
|
svc.list_in_progress_or_claimed.return_value = [task]
|
||||||
|
svc.unclaim_for_reaper = AsyncMock()
|
||||||
|
|
||||||
|
await orch._reap_with_service(svc)
|
||||||
|
|
||||||
|
svc.unclaim_for_reaper.assert_awaited_once_with(task_id)
|
||||||
|
|||||||
Reference in New Issue
Block a user