Agent SDK 서브프로세스 풀은 독립 인프라야

import asyncio, time
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

@dataclass
class Entry:
    client: ClaudeSDKClient
    last_used: float = field(default_factory=time.monotonic)
    turn_lock: asyncio.Lock = field(default_factory=asyncio.Lock)

class SessionManager:
    def __init__(self, max_size: int = 50, ttl_seconds: int = 1800):
        self._entries: dict[str, Entry] = {}
        self._pool_lock = asyncio.Lock()
        self.max_size = max_size
        self.ttl = ttl_seconds

    @asynccontextmanager
    async def turn(self, conv_id: str, options: ClaudeAgentOptions):
        """Lend one client for one whole turn."""
        async with self._pool_lock:
            await self.sweep()
            entry = self._entries.get(conv_id)
            if entry is None:
                if len(self._entries) >= self.max_size:
                    raise RuntimeError("session pool full")
                client = ClaudeSDKClient(options=options)
                await client.connect()
                entry = self._entries[conv_id] = Entry(client)
            entry.last_used = time.monotonic()  # checked out, so not idle
        async with entry.turn_lock:  # held until the turn is over
            try:
                yield entry.client
            finally:
                entry.last_used = time.monotonic()  # the idle clock starts here

    async def sweep(self, now: float | None = None):
        """One pass. A client whose turn holds its lock is never idle."""
        now = time.monotonic() if now is None else now
        for conv_id, entry in list(self._entries.items()):
            if entry.turn_lock.locked():
                continue
            if now - entry.last_used > self.ttl:
                del self._entries[conv_id]
                await entry.client.disconnect()

    async def reaper(self, every: float = 60):
        """Background loop; it runs the same pass the tests call."""
        while True:
            await asyncio.sleep(every)
            async with self._pool_lock:
                await self.sweep()


# One turn, start to finish, inside the lock:
#
#     async with pool.turn(conv_id, options) as client:
#         await client.query(prompt)
#         async for message in client.receive_response():
#             ...

import asyncio, time
import pytest
import session_pool

class FakeClient:
    def __init__(self, options=None):
        self.connected = False
    async def connect(self):
        self.connected = True
    async def disconnect(self):
        self.connected = False

@pytest.fixture
def pool(monkeypatch):
    monkeypatch.setattr(session_pool, "ClaudeSDKClient", FakeClient)
    return session_pool.SessionManager(ttl_seconds=300)

def test_a_turn_in_progress_is_never_swept(pool):
    async def scenario():
        async with pool.turn("conv-a", options=None) as client:
            # Twenty minutes into one long turn, by the sweep's clock.
            await pool.sweep(now=time.monotonic() + 1200)
            return client.connected
    assert asyncio.run(scenario())  # delete the lock check: this goes red

def test_a_client_idle_past_the_ttl_is_swept(pool):
    async def scenario():
        async with pool.turn("conv-a", options=None) as client:
            pass
        await pool.sweep(now=time.monotonic() + 301)
        return client.connected
    assert not asyncio.run(scenario())