fix(chat,sdk): hold the resume cursor only behind records that matter

The `session-in-event-id` header has two consumers with opposite needs. A
fresh boot reads it back as the `.in` resume cursor and needs it conservative,
while a client compares it against the append sequence of its own send to
recognise a turn boundary that predates that send, which needs it exact.
Holding the cursor behind every unconsumed record served neither: an
unconsumed control record pushed the header below the sequence of the message
the turn had just answered, so a client discarded its own turn-complete and
stayed streaming.

The cursor is now held only behind records whose loss would matter, which for
chat means messages. Replaying a stop or a handover on the next boot is
benign, and a handover for a turn that never ran is discarded, so control
records no longer need to hold the cursor. The manager takes the rule as a
per-channel predicate and defaults to holding behind everything, so a missing
or throwing predicate can only make the cursor more conservative. This also
ends the case where one never-consumed record pinned the cursor for the rest
of the run.

Two further gaps in the same machinery:

The unclaimed-kind drain and the cursor rule were installed only for
`chat.customAgent`. `chat.agent` builds its task directly and got neither, so
the managed agent, which is the common surface, kept accumulating barriers
mid-turn. Both are now installed for both surfaces, with the drain attached
after each one's resume cursor is seeded so it cannot open the subscribe at
seq 0.

A handover-prepare boot claims the handover kinds so a signal arriving before
`waitForHandover` attaches is not drained, but the claim was released only
inside `waitForHandover`. A loop that never called it held the claim for the
life of the run, leaving a handover record parked at the head of the channel
where it wedged `chat.messages.next()` permanently. The claim is now also
released at the first turn boundary, by which point the handover window has
closed either way.

Author: ericallam

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

@@ -86,6 +86,14 @@ export class SessionStreamsAPI implements SessionStreamManager {
     return manager.peekRecord(sessionId, io);
   }
 
+  public setCursorBarrier(
+    sessionId: string,
+    io: SessionChannelIO,
+    predicate: SessionStreamRecordPredicate | undefined
+  ): void {
+    this.#getManager().setCursorBarrier?.(sessionId, io, predicate);
+  }
+
   public lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
     return this.#getManager().lastSeqNum(sessionId, io);
   }

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

@@ -70,6 +70,14 @@ export class StandardSessionStreamManager implements SessionStreamManager {
   // Kept separately from `buffer` so the committed cursor can be calculated
   // without depending on buffer traversal.
   private unconsumedSeqNums = new Map<string, Set<number>>();
+
+  /**
+   * Per-channel predicate deciding which buffered records hold the persisted
+   * cursor back. Absent means every record does, which is the conservative
+   * default. Consumers that know their record kinds narrow it so the cursor is
+   * only held behind records whose loss would matter.
+   */
+  private cursorBarriers = new Map<string, SessionStreamRecordPredicate>();
   // High-water mark of seq_nums that have 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
@@ -329,6 +337,37 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     }
   }
 
+  setCursorBarrier(
+    sessionId: string,
+    io: SessionChannelIO,
+    predicate: SessionStreamRecordPredicate | undefined
+  ): void {
+    const key = keyFor(sessionId, io);
+    if (predicate) {
+      this.cursorBarriers.set(key, predicate);
+    } else {
+      this.cursorBarriers.delete(key);
+    }
+  }
+
+  /**
+   * Fails safe: an absent or throwing predicate treats the record as a barrier,
+   * so a mistake here can only make the cursor more conservative, never skip a
+   * record.
+   */
+  #isCursorBarrier(key: string, record: SessionStreamRecord): boolean {
+    const predicate = this.cursorBarriers.get(key);
+    if (!predicate) return true;
+    try {
+      return predicate(record);
+    } catch (error) {
+      if (this.debug) {
+        console.error("[SessionStreamManager] Cursor barrier predicate error:", error);
+      }
+      return true;
+    }
+  }
+
   #markUnconsumedRecord(key: string, seqNum: number): void {
     if (!Number.isFinite(seqNum)) return;
 
@@ -423,6 +462,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
     this.seqNums.clear();
     this.lastDispatchedSeqNums.clear();
     this.unconsumedSeqNums.clear();
+    this.cursorBarriers.clear();
     this.minTimestamps.clear();
     this.handlers.clear();
     this.reconnectAttempts.clear();
@@ -602,7 +642,9 @@ export class StandardSessionStreamManager implements SessionStreamManager {
       this.buffer.set(key, buffered);
     }
     buffered.push(record);
-    this.#markUnconsumedRecord(key, record.seqNum);
+    if (this.#isCursorBarrier(key, record)) {
+      this.#markUnconsumedRecord(key, record.seqNum);
+    }
     this.#drainOnceWaitersFromBuffer(key);
   }
 

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

@@ -55,6 +55,12 @@ export class NoopSessionStreamManager implements SessionStreamManager {
     return undefined;
   }
 
+  setCursorBarrier(
+    _sessionId: string,
+    _io: SessionChannelIO,
+    _predicate: SessionStreamRecordPredicate | undefined
+  ): void {}
+
   lastSeqNum(_sessionId: string, _io: SessionChannelIO): number | undefined {
     return undefined;
   }

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

@@ -82,6 +82,16 @@ export interface SessionStreamManager {
   /** Non-blocking peek at the head record, including its durable metadata. */
   peekRecord?(sessionId: string, io: SessionChannelIO): SessionStreamRecord | undefined;
 
+  /**
+   * Narrow which buffered records hold the persisted cursor back. Absent means
+   * every record does.
+   */
+  setCursorBarrier?(
+    sessionId: string,
+    io: SessionChannelIO,
+    predicate: SessionStreamRecordPredicate | undefined
+  ): void;
+
   /** Last S2 sequence number seen on the given channel. */
   lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined;
 

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

@@ -41,6 +41,7 @@ export class TestSessionStreamManager implements SessionStreamManager {
   private seqNums = new Map<string, number>();
   private dispatchedSeqNums = new Map<string, number>();
   private unconsumedSeqNums = new Map<string, Set<number>>();
+  private cursorBarriers = new Map<string, SessionStreamRecordPredicate>();
 
   on(sessionId: string, io: SessionChannelIO, handler: Handler): { off: () => void } {
     const key = keyFor(sessionId, io);
@@ -197,6 +198,16 @@ export class TestSessionStreamManager implements SessionStreamManager {
     return this.peekRecord(sessionId, io)?.data;
   }
 
+  setCursorBarrier(
+    sessionId: string,
+    io: SessionChannelIO,
+    predicate: SessionStreamRecordPredicate | undefined
+  ): void {
+    const key = keyFor(sessionId, io);
+    if (predicate) this.cursorBarriers.set(key, predicate);
+    else this.cursorBarriers.delete(key);
+  }
+
   peekRecord(sessionId: string, io: SessionChannelIO): SessionStreamRecord | undefined {
     return this.buffer.get(keyFor(sessionId, io))?.[0];
   }
@@ -257,6 +268,16 @@ export class TestSessionStreamManager implements SessionStreamManager {
     }
   }
 
+  #isCursorBarrier(key: string, record: SessionStreamRecord): boolean {
+    const predicate = this.cursorBarriers.get(key);
+    if (!predicate) return true;
+    try {
+      return predicate(record);
+    } catch {
+      return true;
+    }
+  }
+
   #markUnconsumedRecord(key: string, seqNum: number): void {
     if (!Number.isFinite(seqNum)) return;
 
@@ -407,7 +428,9 @@ export class TestSessionStreamManager implements SessionStreamManager {
       this.buffer.set(key, buffered);
     }
     buffered.push(record);
-    this.#markUnconsumedRecord(key, record.seqNum);
+    if (this.#isCursorBarrier(key, record)) {
+      this.#markUnconsumedRecord(key, record.seqNum);
+    }
     this.#drainOnceWaitersFromBuffer(key);
   }
 

packages/trigger-sdk/src/v3/ai.ts

@@ -1563,6 +1563,23 @@ export type ChatMessages = RealtimeDefinedInputStream<ChatTaskWirePayload> & {
   next(options?: { timeoutInSeconds?: number }): Promise<ChatMessageRecord | undefined>;
 };
 
+/**
+ * Only message records hold the persisted `.in` cursor back.
+ *
+ * The `session-in-event-id` header serves two consumers with opposite needs:
+ * `findLatestSessionInCursor` reads it as a resume cursor and wants it
+ * conservative, while a client reads it to correlate its own send's
+ * turn-complete and wants it exact. Holding the cursor behind an unconsumed
+ * control record satisfies neither: resume safety does not need it (replaying a
+ * stop or a handover is benign, and a handover for a turn that never ran is
+ * discarded), while a client comparing the header against its own append
+ * sequence sees a value below its send and discards its own turn boundary.
+ * @internal
+ */
+function isChatCursorBarrier(record: { data: unknown }): boolean {
+  return (record.data as ChatInputChunk | undefined)?.kind === "message";
+}
+
 function isChatMessageRecord(record: { data: unknown }): boolean {
   return (record.data as ChatInputChunk | undefined)?.kind === "message";
 }
@@ -2000,6 +2017,31 @@ function releaseChatInputKinds(kinds: readonly string[]): void {
   locals.set(chatInputDrainKey, attachUnclaimedChatInputDrain());
 }
 
+/**
+ * Narrow what holds the persisted `.in` cursor back. Sets no listener, so it is
+ * safe to call before the resume cursor is seeded.
+ * @internal
+ */
+function setChatCursorBarrier(chatId: string): void {
+  sessionStreams.setCursorBarrier(chatId, "in", isChatCursorBarrier);
+}
+
+/**
+ * Claim the kinds this boot has a consumer for and drain the rest.
+ *
+ * Attaches a `.in` listener, so it MUST run after the resume cursor is seeded;
+ * attaching first makes the subscribe open at seq 0 and replay every record the
+ * previous run already answered.
+ * @internal
+ */
+function attachChatInputDrain(payload: { trigger?: string }): void {
+  const claimed = chatClaimedKinds();
+  if (payload.trigger === "handover-prepare") {
+    for (const kind of CHAT_HANDOVER_KINDS) claimed.add(kind);
+  }
+  locals.set(chatInputDrainKey, attachUnclaimedChatInputDrain());
+}
+
 /**
  * Per-turn deferred promises. Registered via `chat.defer()`, awaited
  * before `onTurnComplete` fires. Reset each turn.
@@ -5527,19 +5569,13 @@ function chatCustomAgent<
       locals.set(lastTurnCompleteSeqNumKey, { value: undefined });
       markChatAgentRunForStreamsWarning();
       taskContext.setConversationId(payload.chatId);
+      setChatCursorBarrier(payload.chatId);
       stampConversationIdOnActiveSpan(payload.chatId);
       // Seed the `.in` resume cursor before user code attaches any `.in`
       // listener — otherwise a continuation boot replays already-answered
       // messages into the loop's first wait.
       await seedSessionInResumeCursorForCustomLoop(payload);
-      // Claim the kinds this boot actually has a consumer for, then attach the
-      // drain for everything else. Handover kinds are only claimed on a
-      // handover-prepare boot, which is the only boot that waits for them.
-      const claimed = chatClaimedKinds();
-      if (payload.trigger === "handover-prepare") {
-        for (const kind of CHAT_HANDOVER_KINDS) claimed.add(kind);
-      }
-      locals.set(chatInputDrainKey, attachUnclaimedChatInputDrain());
+      attachChatInputDrain(payload);
       return userRun(payload, runOptions);
     },
   });
@@ -5646,6 +5682,7 @@ function chatAgent<
       locals.set(lastTurnCompleteSeqNumKey, { value: undefined });
       markChatAgentRunForStreamsWarning();
       taskContext.setConversationId(payload.chatId);
+      setChatCursorBarrier(payload.chatId);
 
       // Stamp `gen_ai.conversation.id` on the run-level span. Every
       // nested span inherits the same attribute via
@@ -5932,6 +5969,8 @@ function chatAgent<
         }
       }
 
+      attachChatInputDrain(payload);
+
       // ── Recovery boot + chain reconstruction ────────────────────────
       if (!hydrateMessages) {
         const settledMessages = mergeByIdReplaceWins<TUIMessage>(
@@ -9109,6 +9148,15 @@ function createStopSignal(): {
 async function chatWriteTurnComplete(options?: {
   publicAccessToken?: string;
 }): Promise<{ lastEventId?: string; sessionInEventId?: string }> {
+  // A handover-prepare boot claims the handover kinds so a signal arriving
+  // before `waitForHandover` attaches is not drained. A loop that never calls
+  // `waitForHandover` would otherwise hold that claim for the life of the run,
+  // leaving any handover record parked at the head of the channel: it wedges
+  // `chat.messages.next()` and, before the cursor barrier narrowed, pinned the
+  // persisted cursor forever. By the time a turn completes the handover window
+  // is over either way.
+  releaseChatInputKinds(CHAT_HANDOVER_KINDS);
+
   const result = await writeTurnCompleteChunk(undefined, options?.publicAccessToken);
   // Same cursor written to the `session-in-event-id` header inside
   // `writeTurnCompleteChunk`; surfaced here so the caller can persist it.