---
title: "Dropped Tokens: Fixing Multi-Turn Agent Streams That Die Mid-Flight"
date: "August 18, 2026"
description: "A 25-second tool call went silent, a corporate proxy cut it at 15 seconds, and a reconnect ran the tool twice. Keep-alives and a Last-Event-ID resume fix both."
category: "Systems Architecture"
canonical: "https://ulukaya.dev/posts/the-leaky-abstraction-vol2"
---

# Dropped Tokens: Fixing Multi-Turn Agent Streams That Die Mid-Flight

My agent called a database tool that took 25 seconds. While the tool ran, my Server-Sent Events (SSE) stream sent nothing. A corporate proxy sat between my client and Cloud Run. At 15 seconds it saw no traffic and [severed the idle socket](#lab-chunk-drop). Then my client reconnected, and things got worse. The client had no idempotent checkpoint offset, so the work ran again and caused duplicate side effects.

		

When my agent calls an external tool, it stops generating tokens. The tool might query a database, run a compiler in a sandbox, or call a third-party API. For those seconds, zero bytes flow over the connection. A carrier NAT gateway or a proxy on the way sees a quiet socket and terminates it. To defend my multi-turn agents, I send keep-alive pings while the tool runs. I also rebuild the stream from a log of numbered events, so a replay never repeats one. And I resume the session inside a transaction. I run the gateway on Cloud Run and keep the event log in Firestore.

		

An idempotent step is safe to repeat: running it twice has the same effect as running it once. A tool that writes a database row is not idempotent. So on a reconnect, my client never sends the prompt again. It sends the number of the last event it saw, in a Last-Event-ID header, and the server replays only the events after it.

		

> The multi-turn reality: When a tool call in my agent takes 15 to 30 seconds, standard HTTP/1.1 and SSE connections often drop. Suppose my design keeps the stream state in memory, in short-lived backend containers. Then a reconnecting client either repeats costly tool side effects or meets state drift it cannot recover from. With no idempotency ledger, my recovery protection is $0.00.

	

	
	

## The architectural gap: Tool latency vs. proxy timeouts

	

In a standard request and response, I can put a bound on latency. A multi-turn agent run has two phases that take turns. My model streams tokens fast, and then it goes silent for a long time while a tool runs:

	
		

### 01. Silent proxy connection drops

		

Carrier NAT gateways, CDN edges and corporate proxies close a connection that stays idle too long. The limit is often between 15 and 60 seconds. My Cloud Run gateway has no idle cut like that. Its limit is the request timeout: 300 seconds by default, and up to 3,600. My agent pauses its text to wait on an external tool, such as an asynchronous BigQuery query or a database transaction with several steps. During that wait, zero bytes cross the connection.

		

The proxy then drops the socket without sending a TCP `FIN` or `RST` packet to my client. Those are the packets that normally close or reset a connection, so my client never learns it is gone. My client UI stays stuck in a loading state. Meanwhile my backend keeps running the work in the background.

		

> The keep-alive rule: While a tool runs, my production streaming gateways send an SSE comment, : ping\n\n, more often than every 15 seconds. A line that starts with a colon is a comment, so the client ignores it. The traffic keeps the TCP socket state (IETF RFC 9293) active through intermediate proxies.

	

	
		

### 02. The stateless reconnection trap

		

Sometimes my mobile or web client switches networks, for example from Wi-Fi to cellular. Sometimes it recovers from a silent timeout. Either way, it opens a new connection. In a naive serverless design, I ran into three separate failures:

		
			- **A different container:** My reconnected request landed on a different container instance in Google Cloud Run. That instance did not hold my previous session's stream buffer in memory.

			- **The tool ran twice:** My client blindly sent the original prompt again. My agent ran the non-idempotent tool calls a second time. That created duplicate database rows and duplicate API charges.

			- **The UI drew everything again:** My server replayed the whole conversation from the start over the new stream. My client UI stuttered, re-rendered hundreds of tokens and lost its scroll position.

		
	

	
	

## The old way vs. the new way

	

For my production multi-turn agents, I stopped assuming the stream lives in memory. I moved to a durable, event-sourced session transport. Event-sourced means the server writes every event to an append-only log and rebuilds the session from that log:

	
		
			
				
					Failure mode
					The old way (naive in-memory streaming)
					The new way (transactional session gateway)
				
			
			
				
					Silence while a tool runs **Idle tool latency**
					Sends nothing, gets cut. Zero bytes sent during tool execution; proxy drops socket after 15s.
					Sends a ping every 10 s. Background heartbeat emitter sends periodic SSE comments (`: ping\n\n`) every 10s.
				
				
					The connection drops mid-reply **Mid-stream disconnect**
					The stream is gone. Stream state lost on container recycle; client restart aborts session.
					Every event is saved in order. An event log in Cloud Firestore gives every event a `seq` number that only goes up. Tool boundaries are committed before the tool runs, and tokens in batches every 500 ms, so a crash loses only the tokens since the last commit. Each event carries a 24-hour `expireAt`, and a TTL policy deletes the event after that time passes.
				
				
					The client comes back **Reconnection ingress**
					Sends the prompt again. Client re-submits prompt, risking duplicate non-idempotent tool actions.
					Asks only for what it missed. Client sends the [`Last-Event-ID` header](https://html.spec.whatwg.org/multipage/server-sent-events.html). The gateway replays only the unacknowledged events from Firestore. If another container is still finishing the turn, the gateway keeps following the journal.
				
				
					Two retries at once **Tool concurrency**
					Both may run the tool. Concurrent client retries trigger race conditions in parallel containers.
					Only the lease holder writes. The caller takes a Firestore lease, and every journal commit checks it in the same transaction. A container that lost the lease writes nothing, and its awaited `tool_start` checkpoint fails before the tool runs.
				
			
		
	

	
		
			
			
			
		
		
			- **Idempotent stream consumer.** It tracks the event sequence IDs, which only go up. When the transport drops, it reconnects with exponential backoff. It passes `Last-Event-ID`, so it resumes with no gap.

			- **Stateful multi-turn reassembler.** While a tool runs, it sends a keep-alive ping more often than every 15 s. It streams model tokens to the client and saves them in batches.

			- **Event-sourced session journal.** My append-only log of token chunks, tool call requests and tool results. A resume replays everything up to the last commit, and a tool boundary is committed before the tool runs.

		
	

	

	

I built the disconnect simulator below to show how a crash in mid-flight corrupts an agent run. [Drop the connection](#lab-chunk-drop) during a multi-turn stream. A blind re-send repeats every token before the drop. My idempotent session journal resumes from `Last-Event-ID` and repeats none:

	

*[Interactive Architecture Simulator: chunk-drop - Explore live at https://ulukaya.dev/posts/the-leaky-abstraction-vol2#lab-chunk-drop]*

	
	

## Production-ready TypeScript implementation

	

Below is my production-tested `AgentSessionStreamGateway` implementation. I deploy it on **Cloud Run** or **Firebase App Hosting**. It sends heartbeats during tool calls and lets a client reconnect with `Last-Event-ID`. Before this class runs, the caller takes the session lease from the table above and passes its owner ID in. Every journal commit checks that ID in the same transaction. Once another owner holds the lease, the commit throws `LeaseLostError`. The caller awaits `emit('tool_start')` before it runs a tool, so a container that lost the lease stops there. A reconnect that finds another owner calls `replayFrom(lastEventId, true)`, which follows that owner's journal until done:

	

```
import type { Response } from 'express';
import { Timestamp, type DocumentReference, type Firestore } from '@google-cloud/firestore';

export interface StreamEvent {
  seq: number;
  type: 'token' | 'tool_start' | 'tool_end' | 'done' | 'error';
  data: string; // JSON-encoded payload, stored and replayed byte for byte
  timestamp: number;
}

export class LeaseLostError extends Error {}

export class AgentSessionStreamGateway {
  private heartbeatTimer?: NodeJS.Timeout;
  private flushTimer?: NodeJS.Timeout;
  private currentSeq: number = 0;
  private replayed = false;
  private eventBuffer: { ref: DocumentReference; event: StreamEvent }[] = [];
  private flushChain: Promise = Promise.resolve();
  private readonly REPLAY_PAGE_SIZE = 200;
  private readonly TOKEN_FLUSH_MS = 500;
  private readonly JOURNAL_TTL_MS = 24 * 60 * 60 * 1000;

  constructor(
    private readonly sessionId: string,
    private readonly res: Response,
    private readonly db: Firestore,
    private readonly ownerId: string // the ID the caller wrote to leaseOwner when it took the lease
  ) {}

  /**
   * Initializes SSE response headers and begins periodic keep-alive pings.
   */
  public initHeaders(): void {
    this.res.setHeader('Content-Type', 'text/event-stream');
    this.res.setHeader('Cache-Control', 'no-cache, no-transform');
    this.res.setHeader('Connection', 'keep-alive');
    this.res.setHeader('X-Accel-Buffering', 'no');
    this.res.flushHeaders();

    // Emit an SSE comment ping every 10 seconds to prevent proxy timeouts
    this.heartbeatTimer = setInterval(() => {
      if (this.isOpen()) {
        this.res.write(': ping\n\n');
      }
    }, 10_000);
    // Stop pinging the moment the client disconnects
    this.res.on('close', () => clearInterval(this.heartbeatTimer));

    // Persist buffered tokens every 500 ms, not only at tool boundaries
    this.flushTimer = setInterval(() => this.flushInBackground(), this.TOKEN_FLUSH_MS);
  }

  /**
   * Replays every event after lastEventId, one page at a time, and continues the
   * sequence from there so a new event never reuses an ID the client already has.
   * Call it once before emit(), with 0 for a new session.
   */
  public async replayFrom(lastEventId: number, follow = false): Promise {
    let cursor = Number.isSafeInteger(lastEventId) && lastEventId > 0 ? lastEventId : 0;
    let finished = false;

    while (true) {
      const page = await this.eventsRef()
        .where('seq', '>', cursor)
        .orderBy('seq', 'asc')
        .limit(this.REPLAY_PAGE_SIZE)
        .get();

      for (const doc of page.docs) {
        const event = doc.data() as StreamEvent;
        this.writeSseFrame(event);
        cursor = event.seq;
        if (event.type === 'done' || event.type === 'error') finished = true;
      }
      if (page.size === this.REPLAY_PAGE_SIZE) continue;
      // follow: another instance owns the turn, so keep reading its journal until done
      if (!follow || finished || !this.isOpen()) break;
      await new Promise((resolve) => setTimeout(resolve, this.TOKEN_FLUSH_MS));
    }

    this.currentSeq = cursor;
    this.replayed = true;
    return cursor;
  }

  /**
   * Emits tokens instantly to client socket and buffers state for batched persistence.
   */
  public emit(type: StreamEvent['type'], payload: unknown): Promise {
    if (!this.replayed) throw new Error('Call replayFrom() before emit()');
    this.currentSeq += 1;
    const event: StreamEvent = {
      seq: this.currentSeq,
      type,
      data: JSON.stringify(payload ?? null),
      timestamp: Date.now(),
    };

    // 1. Flush immediately to client socket (zero latency penalty on streaming)
    this.writeSseFrame(event);

    // 2. Buffer in memory for batched commit; the auto-ID is fixed here so a retry rewrites the same doc
    this.eventBuffer.push({ ref: this.eventsRef().doc(), event });

    // 3. Checkpoint tool boundaries now; await emit('tool_start') before the tool runs
    if (type === 'token') return Promise.resolve();
    const checkpoint = this.flushBuffer();
    checkpoint.catch((error) => console.error(`Journal flush failed for session ${this.sessionId}`, error));
    return checkpoint;
  }

  /**
   * Flushes in-flight event buffer to Firestore in one transaction that also checks the lease.
   * Commits run one at a time; a failed commit goes back into the buffer unless the lease was lost.
   */
  public flushBuffer(): Promise {
    const run = this.flushChain.then(() => this.commitBuffered());
    this.flushChain = run.catch(() => {}); // one failure must not block later flushes
    return run;
  }

  private async commitBuffered(): Promise {
    const pending = this.eventBuffer.splice(0);
    if (pending.length === 0) return;

    try {
      // The lease check and the writes commit together, so a container that lost the lease writes nothing
      await this.db.runTransaction(async (tx) => {
        const session = await tx.get(this.db.collection('agent_sessions').doc(this.sessionId));
        if (session.get('leaseOwner') !== this.ownerId) throw new LeaseLostError(`Lost the lease on ${this.sessionId}`);
        for (const { ref, event } of pending) {
          // A TTL policy on expireAt (exempt from indexing) deletes each event after it expires
          tx.set(ref, { ...event, expireAt: Timestamp.fromMillis(event.timestamp + this.JOURNAL_TTL_MS) });
        }
      });
    } catch (error) {
      if (!(error instanceof LeaseLostError)) this.eventBuffer.unshift(...pending); // retried on the next flush
      throw error;
    }
  }

  private flushInBackground(): void {
    this.flushBuffer().catch((error) => {
      console.error(`Journal flush failed for session ${this.sessionId}; will retry`, error);
    });
  }

  private writeSseFrame(event: StreamEvent): void {
    if (!this.isOpen()) return;
    // JSON.stringify never emits a raw newline, so one data: line carries the payload
    this.res.write(`id: ${event.seq}\nevent: ${event.type}\ndata: ${event.data}\n\n`);
  }

  private isOpen(): boolean {
    // After a client disconnect, writableEnded stays false and destroyed turns true
    return !this.res.writableEnded && !this.res.destroyed;
  }

  private eventsRef() {
    return this.db.collection('agent_sessions').doc(this.sessionId).collection('events');
  }

  public async close(): Promise {
    clearInterval(this.heartbeatTimer);
    clearInterval(this.flushTimer);
    try {
      // Queued behind any in-flight commit, and retried: nothing else will store the final events
      for (let attempt = 1; ; attempt++) {
        try {
          await this.flushBuffer();
          break;
        } catch (error) {
          if (attempt >= 5 || error instanceof LeaseLostError) throw error;
          await new Promise((resolve) => setTimeout(resolve, 250 * 2 ** attempt));
        }
      }
    } finally {
      if (!this.res.writableEnded) {
        this.res.end();
      }
    }
  }
}
```

	

	

> Architectural takeaway: I do not rely on short-lived HTTP connections to carry a multi-turn agent stream. While a tool runs, I keep the socket active with keep-alive pings. I save the event stream in Cloud Firestore, with sequence IDs that only go up. After a network drop, my client reconnects with Last-Event-ID and the server recovers the state from there.

	

> Architecture blueprint and spec: Inspect my complete Deterministic Agent Runtime Blueprint → or generate production-ready specification files with my noVibes Agent Spec Generator →

	
		

## Industry validation and benchmarks

		
			- [Belayer: Efficient Fault Tolerance for LLM Agentic RL Training (Jul 2026)](https://arxiv.org/abs/2608.14635v2): Confirms that long agent runs with stateful side effects (DB writes, file changes) need explicit checkpoints and idempotent replay. Without them, a transport disconnect leads to duplicate side effects.

			- [Concordia: JIT-Compiled Persistent-Kernel Checkpointing for Fault-Tolerant LLM Inference (Jun 2026)](https://arxiv.org/abs/2606.23521v1): Confirms that restoring checkpoint state in under a second removes recovery stalls during mid-stream network drops.

			- [WHATWG HTML Standard: Server-Sent Events (SSE) Protocol and Last-Event-ID](https://html.spec.whatwg.org/multipage/server-sent-events.html)

			- [Google Cloud Run: HTTPS Ingress, Timeouts, and Streaming Configuration](https://cloud.google.com/run/docs/triggering/https-request)

			- [Cloud Firestore: Atomic Transactions and Batched Operations](https://firebase.google.com/docs/firestore/manage-data/transactions)

			- [Firebase AI Logic Documentation and Tool Calling Architecture](https://firebase.google.com/docs/ai-logic)

			- [Firebase App Check: Production Attestation for Streaming Backends](https://firebase.google.com/docs/app-check)

			- [IETF RFC 9293: Transmission Control Protocol (TCP) Specification](https://datatracker.ietf.org/doc/html/rfc9293)