fix(core,sdk): resume session.in via seq cursor on turn-complete header

Replaces the wall-clock dedup cutoff on session.in with a seq-based
resume cursor carried on each turn-complete control record as a
`session-in-event-id` sibling header. On worker boot the cursor is
read off .out's latest turn-complete and seeds .in's SSE subscription
so already-processed user messages aren't replayed from S2.

The timestamp model had a narrow race: the snapshot's lastOutTimestamp
was Date.now() sampled after turn-complete landed on S2, so a user
message arriving between turn-complete and snapshot-write could be
silently filtered as "old" on the next worker boot. Seq numbers are
unique, total-ordered, and stamped by S2, so the same cutoff now
falls out without any clock involved.

The sessionStreams manager grows a second cursor (lastDispatchedSeqNum)
distinct from the existing received cursor. It advances only when a
record is delivered to a once()/wait() consumer or shifted off the
buffer into one. The agent persists it on turn-complete and seeds it
back at the next worker's boot before any .in SSE subscribe opens.
lastOutTimestamp is gone from the snapshot; the rescan is O(1 turn)
since .out is bounded.

Also bundles four small PR-review fixes:
- runStream SSEStreamPart doc had the trigger-control vs
  S2-command-record marker reversed
- agentChat removed a redundant reader.releaseLock() before a
  recursive return that the finally block already covers
- sessions surfaces onControl on SessionSubscribeOptions so consumers
  of the raw .out.read() API see header-form control records
- chatSnapshot drops the now-unused lastOutTimestamp field

Author: ericallam

packages/cli-v3/src/mcp/tools/agentChat.ts

@@ -433,9 +433,8 @@ async function collectAgentResponse(
         // record's seq (set above when the part arrived). The recursive
         // subscribe resumes right after that marker, so we don't replay
         // the entire session.out stream — which would hit a historical
-        // turn-complete and break the loop with empty/old text.
-        reader.releaseLock();
-        // Recurse — subscribe to the new run's stream (same session.out URL)
+        // turn-complete and break the loop with empty/old text. The outer
+        // `finally` block releases the reader before the recursion runs.
         return collectAgentResponse(session, depth + 1);
       }
 

packages/core/src/v3/apiClient/runStream.ts

@@ -179,10 +179,10 @@ export type SSEStreamPart<TChunk = unknown> = {
   /**
    * S2 record headers, when the underlying transport is the v2 batch shape
    * (Session streams). Undefined for v1 streams. Empty array when the record
-   * had no headers. First-header empty-name is a Trigger control protocol
-   * marker (see `trigger-control` records on `session.out`); empty-name
-   * records that S2 itself interprets as command records (trim/fence) are
-   * filtered out before reaching this struct.
+   * had no headers. Trigger control records carry a `trigger-control` named
+   * header (see `trigger-control` records on `session.out`) and may reach
+   * this struct. S2 command records (trim/fence) are identified by an
+   * empty-name first header and are filtered out before enqueue.
    */
   headers?: Array<[string, string]>;
 };

packages/core/src/v3/sessionStreams/chatSnapshot.ts

@@ -13,9 +13,6 @@
  * replay from precisely after the snapshot, and as the trim-chain seed
  * for the agent's next turn.
  *
- * `lastOutTimestamp` is the same record's S2 arrival timestamp (ms since
- * epoch). Used as the dedup cutoff for `session.in` on OOM-retry boot.
- *
  * The `version` field is a forward-compat lever: readers that don't
  * recognise a version silently fall back to no-snapshot behaviour.
  */
@@ -29,7 +26,6 @@ export type ChatSnapshotV1<TUIMessage extends UIMessage = UIMessage> = {
   savedAt: number;
   messages: TUIMessage[];
   lastOutEventId?: string;
-  lastOutTimestamp?: number;
 };
 
 /**
@@ -43,7 +39,6 @@ export const ChatSnapshotV1Schema = z.object({
   savedAt: z.number(),
   messages: z.array(z.unknown()),
   lastOutEventId: z.string().optional(),
-  lastOutTimestamp: z.number().optional(),
 });
 
 /**

packages/core/src/v3/sessionStreams/index.ts

@@ -59,6 +59,18 @@ export class SessionStreamsAPI implements SessionStreamManager {
     this.#getManager().setLastSeqNum(sessionId, io, seqNum);
   }
 
+  public lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
+    return this.#getManager().lastDispatchedSeqNum(sessionId, io);
+  }
+
+  public setLastDispatchedSeqNum(
+    sessionId: string,
+    io: SessionChannelIO,
+    seqNum: number
+  ): void {
+    this.#getManager().setLastDispatchedSeqNum(sessionId, io, seqNum);
+  }
+
   public setMinTimestamp(
     sessionId: string,
     io: SessionChannelIO,

packages/core/src/v3/sessionStreams/manager.ts

@@ -44,6 +44,12 @@ export class StandardSessionStreamManager implements SessionStreamManager {
   private handlers = new Map<string, Set<SessionStreamHandler>>();
   private onceWaiters = new Map<string, OnceWaiter[]>();
   private buffer = new Map<string, unknown[]>();
+  // Parallel to `buffer`: the SSE seq_num of each buffered record. Same
+  // length and order as `buffer[key]`. Used so that when `once()` shifts
+  // a buffered record into a waiter, the cursor (`lastDispatchedSeqNums`)
+  // can advance to that record's seq. Kept as a separate map so the
+  // existing `peek()` shape (returns `unknown`) stays unchanged.
+  private bufferSeqNums = new Map<string, number[]>();
   private tails = new Map<string, TailState>();
   // Per-stream lower-bound timestamp filter. When set, records whose
   // SSE timestamp is <= the bound are dropped before dispatch — used by
@@ -59,6 +65,15 @@ export class StandardSessionStreamManager implements SessionStreamManager {
   // that's already being delivered out-of-band via the waitpoint.
   private explicitlyDisconnected = new Set<string>();
   private seqNums = new Map<string, number>();
+  // Highest seq_num that has been *consumed* (delivered to a once()
+  // waiter or shifted off the buffer into a once() caller) on a channel.
+  // Distinct from `seqNums`, which advances whenever any record is
+  // received from SSE — even ones still sitting in the local buffer.
+  // The committed-consume cursor is what gets persisted on the
+  // turn-complete control record's `session-in-event-id` header so the
+  // next worker boot can resume `.in` from this point without
+  // re-delivering already-handled user messages.
+  private lastDispatchedSeqNums = new Map<string, number>();
   // Reconnect attempt counter per key. Drives the exponential backoff
   // applied by `#ensureTailConnected`'s `.finally` so a persistent
   // backend failure (auth rejection, 5xx, DNS, etc.) doesn't reconnect
@@ -123,8 +138,14 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     const buffered = this.buffer.get(key);
     if (buffered && buffered.length > 0) {
       const data = buffered.shift()!;
+      const seqList = this.bufferSeqNums.get(key);
+      const shiftedSeqNum = seqList?.shift();
       if (buffered.length === 0) {
         this.buffer.delete(key);
+        this.bufferSeqNums.delete(key);
+      }
+      if (shiftedSeqNum !== undefined) {
+        this.#advanceLastDispatched(key, shiftedSeqNum);
       }
       return new InputStreamOncePromise((resolve) => {
         resolve({ ok: true, output: data });
@@ -186,6 +207,25 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     }
   }
 
+  lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
+    return this.lastDispatchedSeqNums.get(keyFor(sessionId, io));
+  }
+
+  setLastDispatchedSeqNum(
+    sessionId: string,
+    io: SessionChannelIO,
+    seqNum: number
+  ): void {
+    this.#advanceLastDispatched(keyFor(sessionId, io), seqNum);
+  }
+
+  #advanceLastDispatched(key: string, seqNum: number): void {
+    const current = this.lastDispatchedSeqNums.get(key);
+    if (current === undefined || seqNum > current) {
+      this.lastDispatchedSeqNums.set(key, seqNum);
+    }
+  }
+
   setMinTimestamp(
     sessionId: string,
     io: SessionChannelIO,
@@ -204,7 +244,15 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     const buffered = this.buffer.get(key);
     if (buffered && buffered.length > 0) {
       buffered.shift();
-      if (buffered.length === 0) this.buffer.delete(key);
+      const seqList = this.bufferSeqNums.get(key);
+      const shiftedSeqNum = seqList?.shift();
+      if (buffered.length === 0) {
+        this.buffer.delete(key);
+        this.bufferSeqNums.delete(key);
+      }
+      if (shiftedSeqNum !== undefined) {
+        this.#advanceLastDispatched(key, shiftedSeqNum);
+      }
       return true;
     }
     return false;
@@ -224,6 +272,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
       this.tails.delete(key);
     }
     this.buffer.delete(key);
+    this.bufferSeqNums.delete(key);
     // Reset the backoff counter so a future re-attach starts fresh —
     // an explicit disconnect is a deliberate teardown, not evidence of
     // a broken backend.
@@ -261,6 +310,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
   reset(): void {
     this.disconnect();
     this.seqNums.clear();
+    this.lastDispatchedSeqNums.clear();
     this.minTimestamps.clear();
     this.handlers.clear();
     this.reconnectAttempts.clear();
@@ -276,6 +326,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     }
     this.onceWaiters.clear();
     this.buffer.clear();
+    this.bufferSeqNums.clear();
   }
 
   #ensureTailConnected(sessionId: string, io: SessionChannelIO): void {
@@ -385,7 +436,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
               // keep as string
             }
           }
-          this.#dispatch(key, data);
+          this.#dispatch(key, data, Number.isFinite(seqNum) ? seqNum : undefined);
         },
         onComplete: () => {
           if (this.debug) {
@@ -410,7 +461,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     }
   }
 
-  #dispatch(key: string, data: unknown): void {
+  #dispatch(key: string, data: unknown, seqNum: number | undefined): void {
     // Any record flowing through = healthy connection; reset the backoff
     // counter so the next disconnect starts fresh.
     this.reconnectAttempts.delete(key);
@@ -423,6 +474,12 @@ export class StandardSessionStreamManager implements SessionStreamManager {
       if (waiter.signal && waiter.abortHandler) {
         waiter.signal.removeEventListener("abort", waiter.abortHandler);
       }
+      // Record was consumed directly by a waiter — advance the
+      // committed-consume cursor immediately. Buffered-then-shifted
+      // records advance the cursor in `once()` / `shiftBuffer()`.
+      if (seqNum !== undefined) {
+        this.#advanceLastDispatched(key, seqNum);
+      }
       waiter.resolve({ ok: true, output: data });
       this.#invokeHandlers(key, data);
       return;
@@ -442,6 +499,14 @@ export class StandardSessionStreamManager implements SessionStreamManager {
       this.buffer.set(key, buffered);
     }
     buffered.push(data);
+    if (seqNum !== undefined) {
+      let bufferedSeqs = this.bufferSeqNums.get(key);
+      if (!bufferedSeqs) {
+        bufferedSeqs = [];
+        this.bufferSeqNums.set(key, bufferedSeqs);
+      }
+      bufferedSeqs.push(seqNum);
+    }
   }
 
   #invokeHandlers(key: string, data: unknown): void {

packages/core/src/v3/sessionStreams/noopManager.ts

@@ -31,6 +31,16 @@ export class NoopSessionStreamManager implements SessionStreamManager {
 
   setLastSeqNum(_sessionId: string, _io: SessionChannelIO, _seqNum: number): void {}
 
+  lastDispatchedSeqNum(_sessionId: string, _io: SessionChannelIO): number | undefined {
+    return undefined;
+  }
+
+  setLastDispatchedSeqNum(
+    _sessionId: string,
+    _io: SessionChannelIO,
+    _seqNum: number
+  ): void {}
+
   setMinTimestamp(
     _sessionId: string,
     _io: SessionChannelIO,

packages/core/src/v3/sessionStreams/types.ts

@@ -45,6 +45,28 @@ export interface SessionStreamManager {
   /** Advance the last-seen sequence number (prevents SSE replay after `.wait` resume). */
   setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void;
 
+  /**
+   * Highest sequence number that has been *consumed* on the channel —
+   * delivered to a `once()` waiter or shifted off the buffer into one.
+   * Distinct from {@link lastSeqNum}, which advances on every received
+   * record regardless of whether anything consumed it. Used by
+   * `chat.agent` to persist the `.in` resume cursor on each
+   * `turn-complete` control record so the next worker boot can resume
+   * the channel from this point without replaying processed messages.
+   */
+  lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined;
+
+  /**
+   * Seed the committed-consume cursor at worker boot — e.g. from the
+   * `session-in-event-id` header on the latest `turn-complete` on
+   * `.out`. Monotonic: only ever advances forward, never backwards.
+   */
+  setLastDispatchedSeqNum(
+    sessionId: string,
+    io: SessionChannelIO,
+    seqNum: number
+  ): void;
+
   /**
    * Set a per-stream lower-bound SSE timestamp. Records whose timestamp
    * is `<= minTimestamp` are dropped before dispatch. Used by chat.agent

packages/core/src/v3/sessionStreams/wireProtocol.ts

@@ -28,6 +28,18 @@ export const PUBLIC_ACCESS_TOKEN_HEADER = "public-access-token" as const;
 /** Header name carrying the agent's last S2 event id on a handover bridge. */
 export const SESSION_STATE_LAST_EVENT_ID_HEADER = "last-event-id" as const;
 
+/**
+ * Header on `turn-complete` records carrying the highest `session.in`
+ * seq_num the agent committed to processing during this turn. Read on
+ * the next worker boot to seed `.in`'s resume cursor — anything past
+ * this seq is new and gets delivered; anything at-or-before was already
+ * processed and is skipped. Decimal-string form of the seq_num.
+ *
+ * Omitted when no `.in` records have been consumed yet (first turn of a
+ * fresh chat triggered via the wire payload).
+ */
+export const SESSION_IN_EVENT_ID_HEADER = "session-in-event-id" as const;
+
 export const TRIGGER_CONTROL_SUBTYPE = {
   TURN_COMPLETE: "turn-complete",
   UPGRADE_REQUIRED: "upgrade-required",

packages/core/src/v3/test/test-session-stream-manager.ts

@@ -145,6 +145,21 @@ export class TestSessionStreamManager implements SessionStreamManager {
     this.seqNums.set(keyFor(sessionId, io), seqNum);
   }
 
+  lastDispatchedSeqNum(_sessionId: string, _io: SessionChannelIO): number | undefined {
+    // The test harness drives records via `__sendFromTest` without seq
+    // numbers, so the committed-consume cursor stays undefined. Tests
+    // that need cursor behaviour exercise it via the real manager.
+    return undefined;
+  }
+
+  setLastDispatchedSeqNum(
+    _sessionId: string,
+    _io: SessionChannelIO,
+    _seqNum: number
+  ): void {
+    // no-op — see comment on `lastDispatchedSeqNum`.
+  }
+
   setMinTimestamp(
     _sessionId: string,
     _io: SessionChannelIO,