fix(sdk,core): harden the transcript read path

Treat a non-positive page limit as no limit instead of an empty page a
caller cannot tell apart from the end of the transcript. Hold a seeded
resume cursor until the session exists, so a transcript load that resolves
before the session is created still opens the live stream past the
persisted history instead of replaying it.

Author: ericallam

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

@@ -150,7 +150,9 @@ export function parseTranscriptSnapshot<TUIMessage extends UIMessage = UIMessage
 
 /**
  * Select one page of transcript entries, newest last. `before` keeps only the
- * entries ordered before that id; `limit` keeps the last that many. The
+ * entries ordered before that id; `limit` keeps the last that many. A
+ * non-positive `limit` is treated as no limit (every entry, no cursor), so a
+ * caller cannot mistake an empty page for the end of the transcript. The
  * returned `nextCursor` is the id to pass as `before` for the previous page,
  * absent when there is no earlier page.
  */
@@ -164,7 +166,7 @@ export function pageTranscriptEntries<TUIMessage extends UIMessage = UIMessage>(
     if (idx !== -1) entries = entries.slice(0, idx);
   }
   let nextCursor: string | undefined;
-  if (opts?.limit !== undefined && opts.limit >= 0 && entries.length > opts.limit) {
+  if (opts?.limit !== undefined && opts.limit > 0 && entries.length > opts.limit) {
     entries = entries.slice(entries.length - opts.limit);
     nextCursor = entries[0]?.id;
   }

packages/core/test/chatSnapshot.test.ts

@@ -36,6 +36,12 @@ describe("pageTranscriptEntries", () => {
   it("ignores an unknown `before` id", () => {
     expect(pageTranscriptEntries(entries, { before: "zz" }).entries).toHaveLength(5);
   });
+
+  it("treats a zero limit as no limit rather than an empty page", () => {
+    const page = pageTranscriptEntries(entries, { limit: 0 });
+    expect(page.entries).toEqual(entries);
+    expect(page.nextCursor).toBeUndefined();
+  });
 });
 
 const user = { id: "u-1", role: "user" as const, parts: [{ type: "text" as const, text: "hi" }] };

packages/trigger-sdk/src/v3/chat-react.ts

@@ -74,20 +74,19 @@ export type UseLoadTranscriptOptions = {
 /**
  * Move the transport's resume cursor for `chatId` to the transcript's
  * `lastOutEventId`, so the live subscription opens just past the persisted
- * history. A no-op when the transcript carries no cursor or the transport
- * does not know the session yet (the cursor cannot be stored without the
- * session's access token). Returns whether the cursor was seeded.
+ * history. Applied to the session now if it exists, otherwise held by the
+ * transport until the session is created, so a load that resolves before the
+ * session exists still moves the cursor. A no-op when the transcript carries
+ * no cursor. Returns whether a cursor was provided.
  */
 export function seedTranscriptCursor(
-  transport: Pick<TriggerChatTransport, "getSession" | "setSession">,
+  transport: Pick<TriggerChatTransport, "seedResumeCursor">,
   chatId: string,
   cursors: { lastOutEventId?: string } | undefined
 ): boolean {
   const lastEventId = cursors?.lastOutEventId;
   if (!lastEventId) return false;
-  const session = transport.getSession(chatId);
-  if (!session) return false;
-  transport.setSession(chatId, { ...session, lastEventId });
+  transport.seedResumeCursor(chatId, lastEventId);
   return true;
 }
 

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

@@ -730,6 +730,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
   private _onEvent: ((event: ChatTransportEvent) => void) | undefined;
 
   private sessions: Map<string, ChatSessionState> = new Map();
+  private pendingResumeCursors: Map<string, string> = new Map();
   private activeStreams: Map<string, AbortController> = new Map();
   private pendingStarts: Map<string, Promise<ChatSessionState>> = new Map();
   // Last turn-producing send per chat — attribution source for the
@@ -1404,6 +1405,31 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     this.notifySessionChange(chatId, this.toPersisted(this.sessions.get(chatId)!));
   }
 
+  /**
+   * Seed the `.out` resume cursor from a loaded transcript. Applied to the
+   * session now if it exists, otherwise held until the session is created so
+   * the first live subscription opens past the persisted history instead of
+   * replaying it.
+   */
+  seedResumeCursor = (chatId: string, lastEventId: string): void => {
+    const existing = this.sessions.get(chatId);
+    if (existing?.publicAccessToken) {
+      existing.lastEventId = lastEventId;
+      this.notifySessionChange(chatId, this.toPersisted(existing));
+      return;
+    }
+    this.pendingResumeCursors.set(chatId, lastEventId);
+  };
+
+  private applyPendingResumeCursor(chatId: string, state: ChatSessionState): ChatSessionState {
+    const pending = this.pendingResumeCursors.get(chatId);
+    if (pending !== undefined && state.lastEventId === undefined) {
+      state.lastEventId = pending;
+    }
+    this.pendingResumeCursors.delete(chatId);
+    return state;
+  }
+
   setOnSessionChange(
     callback: ((chatId: string, session: ChatSessionPersistedState | null) => void) | undefined
   ): void {
@@ -1614,7 +1640,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     // `sessions: { ... }` already, or the very first `accessToken` call
     // returns a PAT for an out-of-band-created session.
     const token = await this.resolveAccessToken({ chatId });
-    const state: ChatSessionState = { publicAccessToken: token };
+    const state = this.applyPendingResumeCursor(chatId, { publicAccessToken: token });
     this.sessions.set(chatId, state);
     this.notifySessionChange(chatId, state);
     return state;
@@ -1633,10 +1659,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       clientData: (this.defaultMetadata ?? {}) as Record<string, unknown>,
     });
 
-    const state: ChatSessionState = {
+    const state = this.applyPendingResumeCursor(chatId, {
       publicAccessToken,
       isStreaming: false,
-    };
+    });
     this.sessions.set(chatId, state);
     this.notifySessionChange(chatId, state);
     return state;

packages/trigger-sdk/test/use-load-transcript.test.ts

@@ -1,56 +1,59 @@
 import { describe, expect, it } from "vitest";
-import type { ChatSessionPersistedState } from "../src/v3/chat.js";
+import { TriggerChatTransport } from "../src/v3/chat.js";
 import { seedTranscriptCursor } from "../src/v3/chat-react.js";
 
-function fakeTransport(initial?: ChatSessionPersistedState) {
-  const sessions = new Map<string, ChatSessionPersistedState>();
-  if (initial) sessions.set("chat-1", initial);
-  const calls: Array<{ chatId: string; session: ChatSessionPersistedState }> = [];
-  return {
-    calls,
-    sessions,
-    getSession: (chatId: string) => sessions.get(chatId),
-    setSession: (chatId: string, session: ChatSessionPersistedState) => {
-      calls.push({ chatId, session });
-      sessions.set(chatId, session);
-    },
-  };
+function transportWithStart() {
+  return new TriggerChatTransport({
+    task: "my-chat",
+    accessToken: () => "pat",
+    startSession: async () => ({ publicAccessToken: "pat-from-start" }),
+  });
 }
 
 describe("seedTranscriptCursor", () => {
-  it("moves the transport's resume cursor to the transcript's lastOutEventId", () => {
-    const transport = fakeTransport({
-      publicAccessToken: "pat",
-      lastEventId: "3",
-      activeInputSeq: 2,
-      isStreaming: false,
-    });
-
-    expect(seedTranscriptCursor(transport, "chat-1", { lastOutEventId: "42" })).toBe(true);
-    expect(transport.calls).toEqual([
-      {
-        chatId: "chat-1",
-        session: {
-          publicAccessToken: "pat",
-          lastEventId: "42",
-          activeInputSeq: 2,
-          isStreaming: false,
-        },
+  it("forwards the transcript's lastOutEventId to the transport", () => {
+    const calls: Array<{ chatId: string; lastEventId: string }> = [];
+    const transport = {
+      seedResumeCursor: (chatId: string, lastEventId: string) => {
+        calls.push({ chatId, lastEventId });
       },
-    ]);
-  });
+    };
 
-  it("does nothing when the transport does not know the session yet", () => {
-    const transport = fakeTransport();
-    expect(seedTranscriptCursor(transport, "chat-1", { lastOutEventId: "42" })).toBe(false);
-    expect(transport.calls).toEqual([]);
+    expect(seedTranscriptCursor(transport, "chat-1", { lastOutEventId: "42" })).toBe(true);
+    expect(calls).toEqual([{ chatId: "chat-1", lastEventId: "42" }]);
   });
 
   it("does nothing when the transcript carries no cursor", () => {
-    const transport = fakeTransport({ publicAccessToken: "pat", lastEventId: "3" });
+    const calls: string[] = [];
+    const transport = {
+      seedResumeCursor: (chatId: string) => {
+        calls.push(chatId);
+      },
+    };
+
     expect(seedTranscriptCursor(transport, "chat-1", undefined)).toBe(false);
     expect(seedTranscriptCursor(transport, "chat-1", { lastOutEventId: "" })).toBe(false);
-    expect(transport.calls).toEqual([]);
-    expect(transport.getSession("chat-1")?.lastEventId).toBe("3");
+    expect(calls).toEqual([]);
+  });
+});
+
+describe("TriggerChatTransport resume cursor", () => {
+  it("holds a seeded cursor until the session is created, then applies it", async () => {
+    const transport = transportWithStart();
+
+    seedTranscriptCursor(transport, "chat-1", { lastOutEventId: "42" });
+    expect(transport.getSession("chat-1")).toBeUndefined();
+
+    await transport.start("chat-1");
+    expect(transport.getSession("chat-1")?.lastEventId).toBe("42");
+  });
+
+  it("applies a seeded cursor immediately when the session already exists", async () => {
+    const transport = transportWithStart();
+
+    await transport.start("chat-1");
+    seedTranscriptCursor(transport, "chat-1", { lastOutEventId: "99" });
+
+    expect(transport.getSession("chat-1")?.lastEventId).toBe("99");
   });
 });