From 5605fee826a3ca23d41f3c1770582e2ef5dddc5b Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 3 Jul 2026 03:53:19 -0700 Subject: [PATCH] [eric] agents: client-pool idle eviction (30min TTL) + fingerprint field-drift diagnostics --- .../apps/agents/manager/run/client_pool.py | 35 ++++++++++++++++++- backend/tests/test_client_pool.py | 31 ++++++++++++++++ 2 files changed, 65 insertions(+), 1 deletion(-) diff --git a/backend/apps/agents/manager/run/client_pool.py b/backend/apps/agents/manager/run/client_pool.py index 88864f79..02796dbe 100644 --- a/backend/apps/agents/manager/run/client_pool.py +++ b/backend/apps/agents/manager/run/client_pool.py @@ -31,6 +31,10 @@ def persistent_client_enabled() -> bool: return os.environ.get("OSW_TTFT_PERSISTENT_CLIENT") == "1" +# Per-session field-level digests from the last fingerprint call; lets a mismatch log WHICH boot field drifted (probe-gated diagnostics only). +p_last_field_digests: Dict[str, Dict[str, str]] = {} + + @typechecked def boot_fingerprint(options_kwargs: Dict, session: AgentSession) -> str: """Hash of every input the CLI subprocess freezes at boot. Includes the full mcp_servers config @@ -40,6 +44,14 @@ def boot_fingerprint(options_kwargs: Dict, session: AgentSession) -> str: frozen = {k: v for k, v in options_kwargs.items() if k not in P_NON_BOOT_KEYS} frozen["p_branch"] = session.active_branch_id frozen["p_compacted_through"] = session.compacted_through_msg_id + if os.environ.get("OSW_TTFT_PROBE") == "1": + digests = {k: hashlib.sha256(json.dumps(v, sort_keys=True, default=str).encode()).hexdigest()[:10] for k, v in frozen.items()} + prev = p_last_field_digests.get(session.id) + if prev is not None: + changed = [k for k in digests if prev.get(k) != digests.get(k)] + [k for k in prev if k not in digests] + if changed: + logger.info(f"[client-pool] {session.id}: fingerprint fields changed: {sorted(set(changed))}") + p_last_field_digests[session.id] = digests blob = json.dumps(frozen, sort_keys=True, default=str) return hashlib.sha256(blob.encode()).hexdigest() @@ -51,9 +63,27 @@ class ClientHandle(BaseModel): client: InstanceOf[object] lock: InstanceOf[asyncio.Lock] connected_at: float + last_used: float turns_served: int = 0 +# A pooled CLI holds ~100MB+ per session; evict clients idle past this so parked chats don't accumulate subprocesses (respawn on the next message is the normal cold path). +P_IDLE_EVICT_SECONDS = float(os.environ.get("OSW_CLIENT_IDLE_EVICT_SECONDS", "1800")) + + +@typechecked +async def evict_idle_clients(pool: Dict[str, "ClientHandle"]) -> None: + """Dispose every handle idle past the TTL, skipping any mid-turn (lock held).""" + now = time.monotonic() + for sid in list(pool.keys()): + handle = pool.get(sid) + if handle is None or handle.lock.locked(): + continue + if now - handle.last_used > P_IDLE_EVICT_SECONDS: + logger.info(f"[client-pool] {sid}: idle-evict after {int(now - handle.last_used)}s") + await dispose_client(pool, sid) + + @typechecked async def acquire_client( pool: Dict[str, ClientHandle], @@ -65,16 +95,19 @@ async def acquire_client( """Return a live client whose boot matches `fingerprint`, connecting fresh when there is none, the fingerprint mismatches, or the caller demands a fresh session (needs_fresh/fork consumed upstream, so the flag must be read BEFORE build_agent_options and passed in).""" + await evict_idle_clients(pool) existing = pool.get(session_id) if existing is not None: if not force_respawn and existing.fingerprint == fingerprint: + existing.last_used = time.monotonic() return existing reason = "force_respawn" if force_respawn else "fingerprint_changed" logger.info(f"[client-pool] {session_id}: respawn ({reason})") await dispose_client(pool, session_id) client = await connect_fn() + now = time.monotonic() handle = ClientHandle( - fingerprint=fingerprint, client=client, lock=asyncio.Lock(), connected_at=time.monotonic(), + fingerprint=fingerprint, client=client, lock=asyncio.Lock(), connected_at=now, last_used=now, ) pool[session_id] = handle logger.info(f"[client-pool] {session_id}: connected fresh client") diff --git a/backend/tests/test_client_pool.py b/backend/tests/test_client_pool.py index 07fd2f07..2e4b3c45 100644 --- a/backend/tests/test_client_pool.py +++ b/backend/tests/test_client_pool.py @@ -129,6 +129,37 @@ def test_reuse_respawn_force_and_teardown(): asyncio.run(run()) +def test_idle_eviction(): + async def run(): + import backend.apps.agents.manager.run.client_pool as cp + pool: Dict[str, ClientHandle] = {} + made: List[FakeClient] = [] + + async def connect(): + return FakeClient(made) + + old_ttl = cp.P_IDLE_EVICT_SECONDS + cp.P_IDLE_EVICT_SECONDS = 0.05 + try: + h = await acquire_client(pool, "s1", "fp", connect) + await acquire_client(pool, "s2", "fp", connect) + await asyncio.sleep(0.1) + # s1 is mid-turn (lock held): the sweep must skip it and evict only the idle s2. + async with h.lock: + await cp.evict_idle_clients(pool) + assert "s1" in pool and "s2" not in pool and made[1].disconnected + await asyncio.sleep(0.1) + await cp.evict_idle_clients(pool) + assert "s1" not in pool and made[0].disconnected + # a fresh acquire after eviction reconnects transparently + h2 = await acquire_client(pool, "s1", "fp", connect) + assert h2.client.alive + finally: + cp.P_IDLE_EVICT_SECONDS = old_ttl + + asyncio.run(run()) + + def test_seeded_simulation_invariants(): """Random op sequences: reuse only on identical fingerprint, dead clients always replaced, pool never re-serves a disposed client, and boots never exceed the one-shot baseline (one per turn)."""