« Phase 12 README · Warmup · Hitchhiker's · Principal Deep Dive · Core Contributor · Staff Notes

Phase 12 — Deep Dive: Production Agent Services — Async, Streaming, State & Events

The load-bearing idea of this phase is one sentence: every serving property is a data structure plus a scheduling discipline, and all of them share exactly one thread. Streaming is a suspendable coroutine. A concurrency cap is a counter with a wait-queue. An idempotency store is a hash map with an expiry stamp. A drain is an edge-triggered flag. None of it is magic; all of it is mechanism. This document takes the six mechanisms apart at the level of the objects the interpreter actually allocates and the order in which it touches them.

The event loop is a ready-queue plus a selector

asyncio is a while True: over two collaborating structures. The first is a FIFO of ready callbacks (loop._ready, a collections.deque). The second is an OS readiness selector (epoll on Linux, kqueue on macOS) that answers "which file descriptors can I read/write without blocking?" One iteration ("tick") does three things: drain every callback currently in _ready by calling it (each runs until it hits an await on something not yet ready, then returns control); ask the selector, with a timeout equal to the nearest scheduled timer, which fds are now ready; and push the callbacks parked on those fds back onto _ready. Timers live in a fourth structure, a min-heap keyed by deadline (loop._scheduled), so asyncio.sleep(t) is just "insert a timer, yield, get re-enqueued when the heap top is due."

A coroutine is a stack frame the interpreter can freeze. await x compiles to "drive x until it signals not done, then propagate that suspension up through every awaiting frame to the loop." The value that flows up is a Future — a one-slot box with a done flag, a result, and a list of callbacks. When the awaited I/O completes, the loop sets the Future's result, which schedules its callbacks, which resume the parked coroutine at the exact bytecode offset after the await. There is no thread hand-off, no lock. The critical invariant: between two await points, a coroutine has the CPU entirely to itself. Nothing interleaves inside that span. This is why our lab needs no locks around the in_flight counter — the increment and the checks around it never straddle an await, so they are atomic by construction.

That same invariant is the trap. Cooperative scheduling means a coroutine yields only at await. A synchronous time.sleep(5), a blocking requests.get, or a tight CPU loop never yields, so the loop's _ready deque stops draining and every other coroutine — every other request — is frozen for the full duration. The failure is not localized to the offending request; it is global. That is the mechanical reason "never block the loop" is a hard rule and not a style preference.

The async generator: a stream you can suspend mid-flight

A normal generator suspends at yield and resumes on next(). An async generator (async def + yield) suspends at yield and can suspend at await between yields, and it is driven by async for, which desugars to repeated await gen.__anext__(). This is the streaming primitive. stream_agent pulls the injected model with async for token in model(prompt); each __anext__ awaits the next token (in the lab, await asyncio.sleep(0) — a pure yield to the loop), then the generator emits {"event": "token", ...} and re-suspends. The token that just arrived is handed to the consumer before the next one is fetched. Peak memory is O(1) in the answer length: one token in flight, never the whole string. format_sse then serializes each event to the wire: an event: line, one data: line per newline-split payload segment (JSON-encoding non-strings), and the terminating blank line that tells the client "dispatch now." The whole path — model to generator to formatter to socket — is a chain of suspendable frames with no buffer in the middle. Contrast the naive answer = await model.complete(prompt); return answer: it buffers the entire generation, so time-to-first-byte equals time-to-last-token and O(1) memory becomes O(n). The mechanism forces the latency; streaming is the only way to break the coupling.

The semaphore: a counter with a wait-queue

asyncio.Semaphore(n) is an integer _value = n and a FIFO of Futures, _waiters. acquire(): if _value > 0, decrement and return immediately; else append a fresh Future to _waiters and await it — the coroutine parks, the loop runs others. release(): increment _value, then wake the oldest waiter by setting its Future's result, which re-enqueues that coroutine. FIFO ordering makes it fair: no starvation, arrivals served in order. async with sem: is acquire() on enter, release() on exit — including the exit taken by an exception, which is why a crash mid-request does not leak a permit. The invariant the lab asserts: at most max_concurrency coroutines are ever between acquire and release simultaneously, so max_in_flight (the peak ever observed) reaches the cap and never exceeds it.

The wait-queue is backpressure. When the LLM is the bottleneck, callers pile up as parked Futures, and overload surfaces as latency rather than as a stampede on the provider. But an unbounded _waiters list is unbounded memory, so the service bounds the queue: handle bumps a queued counter, and if it already exceeds max_queue it raises ServiceUnavailable before awaiting the semaphore — a fast 503. Load-shedding is a deliberate mechanism, not a failure: reject cheaply at the door instead of accepting work you cannot hold.

Worked trace: a burst of 5 under Semaphore(2), max_queue=2

Five handle coroutines are launched via gather at tick 0. The loop runs them in creation order. R1: queued→1, acquires (value 2→1), enters the model section, hits its first await and parks. R2: queued→2, acquires (1→0), enters, parks on its model await. R3: queued→3, that exceeds max_queue=2 → raises ServiceUnavailable, rejected→1, returns. Same for R4 and R5 → rejected→3. Now the loop resumes R1's model coroutine; it finishes, R1 releases (value 0→1, wakes nobody — queue empty), stores its result, decrements in_flight. R2 likewise. Final metrics: completed=2, rejected=3, max_in_flight=2. The cap held; the excess shed fast. The ordering is fully determined because the only yield point in the injected model is sleep(0), so interleaving is reproducible and the test is exact — no wall clock, no flakiness.

The TTLCache: a hash map with expiry math and a MISS sentinel

set(key, val, ttl) stores (val, now() + ttl) — the expiry is computed once, at write, against the injected integer clock. get(key) reads the entry; if now() < expiry it is a hit; if now() >= expiry it is expired, and the entry is dropped on read (lazy eviction — no background sweeper, no timer, cost amortized onto the next access). Two mechanical details matter. First, expiry is a strict boundary: ttl=5 set at tick 0 is live at tick 4 and dead at tick 5, testable to the exact tick because time is a knob the test turns, not a syscall. Second, get returns a distinct MISS sentinel on absence/expiry, not None — because a stored None is a legitimate hit, and collapsing the two would make "cached the value None" indistinguishable from "nothing cached." This is precisely why Redis separates a nil reply from a stored empty value.

The idempotency replay and its race window

handle checks the idempotency store before acquiring the semaphore: key present and live → return the cached Response marked replayed=True, model invoked zero times, no permit taken. Key absent → run once, set the result under the key with a TTL covering the retry window, return. The lab proves it with a call-counting model: two handle calls, same key, model called once. Advance the fake clock past the TTL and the key re-runs. The honest limit is a race window: a check-then-act on a plain map means two truly concurrent identical requests both read MISS before either writes, so both run. Because a coroutine holds the CPU between awaits, the lab's sequential replay has no window; a real distributed store closes the concurrent case by making first-sight atomic — SET key NX EX ttl reserves the key so the second caller sees "in progress." The mechanism to know: dedupe correctness rests on an atomic reserve, not a read.

Per-subscriber FIFO queues: fan-out vs steal

EventBus.subscribe(topic) allocates a private asyncio.Queue per subscriber; publish(topic, event) iterates every subscriber on the topic and puts the event into each queue. Each subscriber drains its own queue via async for. Two invariants: every subscriber receives every event (fan-out), and within one subscriber events arrive in publish order (queue FIFO). The private-queue choice is the whole design — a single shared queue would let the first reader get an event and remove it, so the others never see it. That shared-queue shape is a competing-consumers work queue (one job to one worker), a different tool. Picking a shared queue when you meant fan-out silently drops events to all-but-one subscriber; picking per-subscriber queues when you meant load-balancing silently runs each job N times. The data structure is the semantics.

Why the naive service fails at the mechanism level

Thread-per-request: each blocking request holds an OS thread (~1 MB stack, a scheduler slot) for its full 8-second I/O wait; Little's Law says 800 in flight needs 800 threads, and the box dies on memory and context-switch overhead long before the LLM is the limit. Buffer-then-return: couples first-byte latency to last-token latency and makes memory O(answer). In-process state: a local dict evaporates on the deploy that recycles the instance, and the load balancer routes turn 2 to a different instance that never saw turn 1. Each failure is structural, not a tuning problem — the async loop, the streaming generator, and the external store are the only shapes that dodge them.