From 7d9215412b5ecf1bfa7fc4348784f37ddbfcd0f3 Mon Sep 17 00:00:00 2001 From: Renn F Date: Thu, 18 Jun 2026 22:17:52 +0200 Subject: [PATCH] feat(grok): reap abandoned interactive chats (M3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An interactive intake/secretary chat the human abandoned (closed the tab without confirming or stopping) leaked its container until the orchestrator restarted — the wedged-grok reaper is task-driven and these run task_id=None, and an SSE disconnect intentionally does NOT reap (so a page reload can reconnect). Reap by IDLE TIME, not connection state: PrompterLiveRegistry tracks last_activity (bumped on every push/deliver = a turn), and the 60s sweeper retires sessions idle past ROBOCO_INTERACTIVE_IDLE_REAP_SECONDS (default 1800; 0 disables) via reap_intake_session / reap_secretary_session. An active or page-reloaded chat that keeps exchanging turns stays fresh and is never reaped; board-review-parked sessions (task_id set) are exempt. Provider-agnostic — fixes the leak for both Claude and Grok interactive. Tests: idle-only reap (active/parked/closed excluded), activity bump keeps a session alive, threshold 0 disables. Gate green (ruff/mypy/xenon + prompter_live). --- roboco/config.py | 15 +++++++++ roboco/runtime/orchestrator.py | 38 +++++++++++++++++++++++ roboco/services/prompter_live.py | 27 ++++++++++++++++ tests/unit/services/test_prompter_live.py | 30 ++++++++++++++++++ 4 files changed, 110 insertions(+) diff --git a/roboco/config.py b/roboco/config.py index 153bfd68..2eb2d30f 100644 --- a/roboco/config.py +++ b/roboco/config.py @@ -691,6 +691,21 @@ class Settings(BaseSettings): "0 disables. Override via ROBOCO_GROK_MAX_COST_USD" ), ) + # An interactive intake/secretary chat the human abandoned (closed the tab + # without confirming/stopping) otherwise leaks its container until the + # orchestrator restarts. The sweeper reaps a live session whose + # time-since-last-turn (push/deliver) exceeds this; measured on activity, NOT + # connection state, so an active or page-reloaded chat that keeps exchanging + # turns is never reaped (board-review-parked sessions are also exempt). + # Seconds; 0 disables. Provider-agnostic (Claude + Grok interactive). + interactive_idle_reap_seconds: int = Field( + default=1800, + ge=0, + description=( + "Idle-reap threshold for live intake/secretary chats (seconds); " + "0 disables. Override via ROBOCO_INTERACTIVE_IDLE_REAP_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 96715846..847290c7 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -3273,6 +3273,40 @@ class AgentOrchestrator: await self.stop_agent(SECRETARY_AGENT_ID, graceful=True) logger.info("Secretary session reaped", session_id=session_id) + async def _reap_idle_interactive_sessions(self) -> None: + """Retire live intake/secretary chats idle past the configured threshold. + + An abandoned chat (the human closed the tab without confirming or + stopping) otherwise leaks its container until the orchestrator restarts. + Idle is measured by time-since-last-turn (push/deliver), NOT connection + state, so an active or page-reloaded chat that keeps exchanging turns is + never reaped; board-review-parked sessions are exempt. Provider-agnostic + (Claude + Grok interactive). Disabled when the threshold is 0. + """ + from roboco.services.prompter_live import get_live_registry + + threshold = float(settings.interactive_idle_reap_seconds) + for session_id, agent_id in get_live_registry().idle_session_ids(threshold): + try: + if agent_id == INTAKE_AGENT_ID: + await self.reap_intake_session(session_id) + elif agent_id == SECRETARY_AGENT_ID: + await self.reap_secretary_session(session_id) + else: + continue + logger.info( + "Reaped idle interactive session", + session_id=session_id, + agent_id=agent_id, + idle_threshold_s=threshold, + ) + except Exception as exc: + logger.warning( + "Idle interactive reap failed", + session_id=session_id, + error=str(exc), + ) + def _resolve_secretary_host_paths(self) -> dict[str, str | None]: """Host paths for the Secretary container's mounts (claude + prompt). @@ -4638,6 +4672,10 @@ Start by: await db.rollback() logger.warning("Notification sweep failed", error=str(e)) + # Retire abandoned live intake/secretary chats (idle past the threshold) + # so a closed-tab session doesn't leak its container until restart. + await self._reap_idle_interactive_sessions() + # Budget kill-switch — runs every sweep. Any agent whose SDK reports # halt=true has breached its per-session tool-call cap; terminate the # container so the next dispatcher tick doesn't waste tokens on the diff --git a/roboco/services/prompter_live.py b/roboco/services/prompter_live.py index b4b00bc9..11633b7e 100644 --- a/roboco/services/prompter_live.py +++ b/roboco/services/prompter_live.py @@ -18,6 +18,7 @@ This module owns no SDK or Claude code — it's pure plumbing, fully unit-tested from __future__ import annotations import asyncio +import time from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any @@ -47,6 +48,11 @@ class LiveIntakeSession: # session stays alive (not reaped) so the board's feedback can be injected # in-context for an in-place re-draft. ``None`` for a normal live chat. task_id: str | None = None + # Monotonic timestamp of the last turn (agent->panel push or panel->agent + # deliver). The idle-reap sweep uses this to retire an ABANDONED chat without + # touching an active or page-reloaded one (reload reconnects + keeps + # exchanging turns, so activity stays fresh). Seeded at open. + last_activity: float = field(default_factory=time.monotonic) class PrompterLiveRegistry: @@ -153,9 +159,29 @@ class PrompterLiveRegistry: session = self._sessions.get(session_id) if session is None or session.closed: return False + session.last_activity = time.monotonic() session.queue.put_nowait(event) return True + def idle_session_ids(self, threshold_seconds: float) -> list[tuple[str, str]]: + """Return ``(session_id, agent_id)`` for live chats idle past the threshold. + + Idle = ``threshold_seconds`` with no turn (push/deliver). Closed and + board-review-parked (``task_id`` set) sessions are excluded — the latter + are intentionally kept alive while the board reviews. ``threshold <= 0`` + yields nothing (the caller treats that as "disabled"). + """ + if threshold_seconds <= 0: + return [] + now = time.monotonic() + return [ + (sid, s.agent_id) + for sid, s in self._sessions.items() + if not s.closed + and s.task_id is None + and now - s.last_activity > threshold_seconds + ] + async def stream(self, session_id: str) -> AsyncIterator[dict[str, Any]]: """Yield queued agent events until the session is closed.""" session = self._sessions.get(session_id) @@ -174,6 +200,7 @@ class PrompterLiveRegistry: session = self._sessions.get(session_id) if session is None or session.closed: return False + session.last_activity = time.monotonic() # human turn = activity url = f"http://roboco-agent-{session.agent_id}:{SDK_PORT}/turn" client = self._client or httpx.AsyncClient(timeout=10.0) try: diff --git a/tests/unit/services/test_prompter_live.py b/tests/unit/services/test_prompter_live.py index 0dcb789a..974acf7a 100644 --- a/tests/unit/services/test_prompter_live.py +++ b/tests/unit/services/test_prompter_live.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import json +import time import httpx import pytest @@ -22,6 +23,35 @@ def test_open_get_close() -> None: assert reg.get("s1") is None +def test_idle_session_ids_reaps_only_abandoned_chats() -> None: + reg = PrompterLiveRegistry() + old = reg.open("idle", "intake-1") + old.last_activity = time.monotonic() - 4000 # silent for >1h + reg.open("fresh", "intake-1") # just opened — active + parked = reg.open("parked", "intake-1") # board-review parked + parked.last_activity = time.monotonic() - 4000 + parked.task_id = "task-9" + reg.open("done", "secretary-1") + reg.close("done") # closed sessions excluded + + idle = dict(reg.idle_session_ids(1800)) + assert "idle" in idle and idle["idle"] == "intake-1" # abandoned -> reaped + assert "fresh" not in idle # active -> kept + assert "parked" not in idle # board-review parked -> exempt + assert "done" not in idle # closed -> excluded + # Disabled when threshold <= 0. + assert reg.idle_session_ids(0) == [] + + +def test_activity_bump_keeps_session_alive() -> None: + reg = PrompterLiveRegistry() + s = reg.open("s1", "intake-1") + s.last_activity = time.monotonic() - 4000 + assert ("s1", "intake-1") in reg.idle_session_ids(1800) + reg.push("s1", {"kind": "text", "text": "hi"}) # an agent turn = activity + assert reg.idle_session_ids(1800) == [] # no longer idle + + def test_close_by_agent_closes_matching_sessions_with_error() -> None: reg = PrompterLiveRegistry() reg.open("s1", "intake-1")