feat(sdk,core,webapp): the transcript storage option, read API and conformance suite

chat.agent takes a storage option, with the platform snapshot as the default and the TranscriptStorage types, defaultStorage, snapshotTranscriptStorage, memoryTranscriptStorage and reduceTranscriptChanges exported from @trigger.dev/sdk/ai. chat.createLoadTranscriptAction(storage) reads a conversation on the server for any storage, and useLoadTranscript renders it in the browser and seeds the transport's resume cursor. runTranscriptStorageTests from @trigger.dev/sdk/ai/test is the contract a storage implementation has to meet.

A new secret-key endpoint, GET /api/v1/sessions/:id/transcript, pages the platform snapshot server-side so rendering the last page of a long conversation does not download all of it; the default storage uses it for paged reads and falls back to the whole blob.

Author: ericallam

apps/webapp/app/routes/api.v1.sessions.$sessionId.transcript.ts

@@ -0,0 +1,75 @@
+import { json } from "@remix-run/server-runtime";
+import { pageTranscriptEntries, parseTranscriptSnapshot } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { chatSnapshotStorageKey } from "~/services/realtime/chatSnapshot.server";
+import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
+import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { downloadPacketFromObjectStore } from "~/v3/objectStore.server";
+
+const ParamsSchema = z.object({
+  sessionId: z.string(),
+});
+
+const SearchParamsSchema = z.object({
+  limit: z.coerce.number().int().min(1).max(1000).optional(),
+  before: z.string().optional(),
+});
+
+function sessionResource(
+  paramId: string,
+  session: { friendlyId: string; externalId: string | null } | null | undefined
+) {
+  const ids = new Set<string>([paramId]);
+  if (session) {
+    ids.add(session.friendlyId);
+    if (session.externalId) ids.add(session.externalId);
+  }
+  return anyResource([...ids].map((id) => ({ type: "sessions" as const, id })));
+}
+
+export const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    searchParams: SearchParamsSchema,
+    corsStrategy: "none",
+    findResource: async (params, auth) =>
+      resolveSessionByIdOrExternalId($replica, auth.environment.id, params.sessionId),
+    authorization: {
+      action: "read",
+      resource: (session, params) => sessionResource(params.sessionId, session),
+    },
+  },
+  async ({ authentication, resource: session, searchParams }) => {
+    if (!session) {
+      return json({ error: "Session not found" }, { status: 404 });
+    }
+
+    let body: unknown;
+    try {
+      const packet = await downloadPacketFromObjectStore(
+        { dataType: "application/store", data: chatSnapshotStorageKey(session) },
+        authentication.environment
+      );
+      body = typeof packet.data === "string" ? JSON.parse(packet.data) : undefined;
+    } catch {
+      body = undefined;
+    }
+
+    const snapshot = parseTranscriptSnapshot(body);
+    if (!snapshot) {
+      return json({ messages: [], state: null });
+    }
+
+    const page = pageTranscriptEntries(snapshot.messages, searchParams);
+    return json({
+      messages: page.entries.map((entry) => entry.message),
+      state: snapshot.state,
+      cursors: {
+        lastOutEventId: snapshot.lastOutEventId,
+        lastInEventId: snapshot.lastInEventId,
+      },
+      nextCursor: page.nextCursor,
+    });
+  }
+);

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

@@ -51,6 +51,7 @@ import {
   CreateSessionStreamWaitpointResponseBody,
   CreateStreamResponseBody,
   CreateUploadPayloadUrlResponseBody,
+  SessionTranscriptResponseBody,
   CreateWaitpointTokenResponseBody,
   CreatedSessionResponseBody,
   DeletedScheduleObject,
@@ -709,6 +710,30 @@ export class ApiClient {
     );
   }
 
+  /**
+   * One page of a `chat.agent` session's persisted transcript, most recent
+   * messages first when `limit` is set. Secret key only.
+   */
+  getSessionTranscript(
+    sessionId: string,
+    options?: { limit?: number; before?: string },
+    requestOptions?: ZodFetchOptions
+  ) {
+    const query = new URLSearchParams();
+    if (options?.limit !== undefined) query.set("limit", String(options.limit));
+    if (options?.before !== undefined) query.set("before", options.before);
+    const suffix = query.size > 0 ? `?${query.toString()}` : "";
+    return zodfetch(
+      SessionTranscriptResponseBody,
+      `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionId)}/transcript${suffix}`,
+      {
+        method: "GET",
+        headers: this.#getHeaders(false),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
   retrieveRun(runId: string, requestOptions?: ZodFetchOptions) {
     return zodfetch(
       RetrieveRunResponse,

packages/core/src/v3/schemas/api.ts

@@ -1005,6 +1005,21 @@ export const CreateUploadPayloadUrlResponseBody = z.object({
   storagePath: z.string().optional(),
 });
 
+/** One page of a `chat.agent` session transcript, from `GET /api/v1/sessions/{id}/transcript`. */
+export const SessionTranscriptResponseBody = z.object({
+  messages: z.array(z.unknown()),
+  state: z.unknown().nullable(),
+  cursors: z
+    .object({
+      lastOutEventId: z.string().optional(),
+      lastInEventId: z.string().optional(),
+    })
+    .optional(),
+  nextCursor: z.string().optional(),
+});
+
+export type SessionTranscriptResponseBody = z.infer<typeof SessionTranscriptResponseBody>;
+
 export const WorkersListResponseBody = z
   .object({
     type: z.string(),

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

@@ -142,6 +142,29 @@ export function parseTranscriptSnapshot<TUIMessage extends UIMessage = UIMessage
   return undefined;
 }
 
+/**
+ * Select one page of transcript entries, newest last. `before` keeps only the
+ * entries ordered before that id; `limit` keeps the last that many. The
+ * returned `nextCursor` is the id to pass as `before` for the previous page,
+ * absent when there is no earlier page.
+ */
+export function pageTranscriptEntries<TUIMessage extends UIMessage = UIMessage>(
+  all: TranscriptSnapshotEntry<TUIMessage>[],
+  opts: { limit?: number; before?: string } | undefined
+): { entries: TranscriptSnapshotEntry<TUIMessage>[]; nextCursor: string | undefined } {
+  let entries = all;
+  if (opts?.before !== undefined) {
+    const idx = entries.findIndex((e) => e.id === opts.before);
+    if (idx !== -1) entries = entries.slice(0, idx);
+  }
+  let nextCursor: string | undefined;
+  if (opts?.limit !== undefined && opts.limit >= 0 && entries.length > opts.limit) {
+    entries = entries.slice(entries.length - opts.limit);
+    nextCursor = entries[0]?.id;
+  }
+  return { entries, nextCursor };
+}
+
 /**
  * S3 key suffix for a session's snapshot blob. The webapp's presigned
  * URL routes prefix this with `packets/{projectRef}/{envSlug}/`.

packages/core/test/chatSnapshot.test.ts

@@ -1,10 +1,43 @@
 import { describe, expect, it } from "vitest";
 import {
+  pageTranscriptEntries,
   parseTranscriptSnapshot,
   type ChatSnapshotV1,
+  type TranscriptSnapshotEntry,
   type TranscriptSnapshotV2,
 } from "../src/v3/sessionStreams/chatSnapshot.js";
 
+describe("pageTranscriptEntries", () => {
+  const entries: TranscriptSnapshotEntry[] = ["a", "b", "c", "d", "e"].map((id) => ({
+    id,
+    final: true,
+    message: { id, role: "user", parts: [] },
+  }));
+
+  it("returns everything with no options and no cursor", () => {
+    expect(pageTranscriptEntries(entries, undefined)).toEqual({ entries, nextCursor: undefined });
+  });
+
+  it("returns the newest `limit` entries with the cursor for the page before", () => {
+    const page = pageTranscriptEntries(entries, { limit: 2 });
+    expect(page.entries.map((e) => e.id)).toEqual(["d", "e"]);
+    expect(page.nextCursor).toBe("d");
+  });
+
+  it("pages backwards with `before` until the cursor runs out", () => {
+    const page = pageTranscriptEntries(entries, { limit: 2, before: "d" });
+    expect(page.entries.map((e) => e.id)).toEqual(["b", "c"]);
+    expect(page.nextCursor).toBe("b");
+    const last = pageTranscriptEntries(entries, { limit: 2, before: "b" });
+    expect(last.entries.map((e) => e.id)).toEqual(["a"]);
+    expect(last.nextCursor).toBeUndefined();
+  });
+
+  it("ignores an unknown `before` id", () => {
+    expect(pageTranscriptEntries(entries, { before: "zz" }).entries).toHaveLength(5);
+  });
+});
+
 const user = { id: "u-1", role: "user" as const, parts: [{ type: "text" as const, text: "hi" }] };
 const assistant = {
   id: "a-1",

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

@@ -81,6 +81,7 @@ import {
   restoreModelLane,
   type TranscriptChange,
   type TranscriptChangeReason,
+  type TranscriptLoadResult,
   type TranscriptRuntimeState,
   type TranscriptShadow,
   type TranscriptStorage,
@@ -277,6 +278,25 @@ export {
   __writeChatSnapshotProductionPathForTests,
 } from "./chatSnapshotIo.js";
 
+export {
+  defaultStorage,
+  memoryTranscriptStorage,
+  reduceTranscriptChanges,
+  snapshotTranscriptStorage,
+  type LoadContextEvent,
+  type MemoryTranscriptStorage,
+  type TranscriptChange,
+  type TranscriptChangeReason,
+  type TranscriptChangeset,
+  type TranscriptCursors,
+  type TranscriptLoadOptions,
+  type TranscriptLoadResult,
+  type TranscriptScope,
+  type TranscriptState,
+  type TranscriptStorage,
+  type TranscriptStorageContext,
+} from "./transcriptStorage.js";
+
 /**
  * Merge two `UIMessage[]` lists by `id`, with the second list winning on
  * collision. Used at run boot to combine the snapshot's persisted history
@@ -6154,6 +6174,33 @@ export type ChatAgentOptions<
     event: HydrateMessagesEvent<inferSchemaOut<TClientDataSchema>, TUIMessage>
   ) => TUIMessage[] | Promise<TUIMessage[]>;
 
+  /**
+   * Where the conversation is persisted. The runtime calls `save` after
+   * every turn, failed turn and history-changing action with the changes
+   * since the last save, and `load` once when a new run boots to continue
+   * the conversation.
+   *
+   * Defaults to `defaultStorage`, the platform's snapshot in object storage
+   * that the Sessions dashboard renders. Bring your own to write each change
+   * to your database; `memoryTranscriptStorage()` is the reference
+   * implementation and `runTranscriptStorageTests` from
+   * `@trigger.dev/sdk/ai/test` checks yours against the contract.
+   *
+   * A storage with `loadContext` also owns the model's context on every
+   * turn, which is what `hydrateMessages` did. The two cannot be combined.
+   *
+   * @example
+   * ```ts
+   * chat.agent({
+   *   id: "my-chat",
+   *   storage: myPostgresTranscriptStorage,
+   *   run: async ({ messages, signal, streamText }) =>
+   *     streamText({ model, messages, abortSignal: signal }),
+   * });
+   * ```
+   */
+  storage?: TranscriptStorage<inferSchemaOut<TClientDataSchema>>;
+
   /**
    * Called at the start of every turn, after message accumulation and `onChatStart` (turn 0),
    * but before the `run` function executes.
@@ -6740,6 +6787,7 @@ function chatAgent<
     onChatStart,
     onValidateMessages,
     hydrateMessages,
+    storage,
     actionSchema,
     onAction,
     onTurnStart,
@@ -6769,12 +6817,17 @@ function chatAgent<
   } = options;
 
   if (hydrateMessages) {
-    const storageAtDefinition = transcriptStorageOverride ?? defaultStorage;
-    if (typeof storageAtDefinition.loadContext === "function") {
+    if (storage) {
+      throw new Error(
+        `chat.agent: "${options.id}" sets both \`hydrateMessages\` and \`storage\`. ` +
+          "`hydrateMessages` is deprecated and replaced by the storage: `save` receives every " +
+          "change and `loadContext` on the storage owns the model's context. Remove `hydrateMessages`."
+      );
+    }
+    if (typeof (transcriptStorageOverride ?? defaultStorage).loadContext === "function") {
       throw new Error(
         `chat.agent: "${options.id}" sets \`hydrateMessages\` and uses a transcript storage with ` +
-          "`loadContext`. Both would own the model's context; keep one. `hydrateMessages` is " +
-          "deprecated, so prefer `loadContext` on the storage."
+          "`loadContext`. Both would own the model's context; keep one."
       );
     }
     warnHydrateMessagesDeprecatedOnce(options.id);
@@ -6946,7 +6999,10 @@ function chatAgent<
       // collectively cost ~600ms on every first-message TTFC. Both reads
       // swallow errors internally; the agent stays available either way.
       const sessionIdForSnapshot = payload.sessionId ?? payload.chatId;
-      const transcriptStorage = transcriptStorageOverride ?? defaultStorage;
+      const transcriptStorage: TranscriptStorage<unknown> =
+        (storage as TranscriptStorage<unknown> | undefined) ??
+        transcriptStorageOverride ??
+        defaultStorage;
       const storageLoadContext = transcriptStorage.loadContext?.bind(transcriptStorage);
       /**
        * Who supplies the model's context each turn: the deprecated
@@ -12659,6 +12715,68 @@ async function mintPublicTokenWithOverride(args: {
   });
 }
 
+export type CreateChatLoadTranscriptActionOptions = {
+  /**
+   * Scope the action to a specific API client configuration (secret key,
+   * base URL) instead of the process-wide one. The default storage reads
+   * through this client.
+   */
+  apiClient?: ApiClientConfiguration;
+  /** Page size when the caller passes none. */
+  limit?: number;
+};
+
+export type ChatLoadTranscriptParams<TClientData = unknown> = {
+  chatId: string;
+  clientData?: TClientData;
+  limit?: number;
+  before?: string;
+};
+
+/**
+ * Creates a server-side helper that reads a conversation from a transcript
+ * storage, for rendering history before the chat connects. Works the same
+ * for every storage, the platform default included, so the browser never
+ * reads a store directly and the secret key stays on the server.
+ *
+ * Wrap it in a Next.js server action (or any server-side handler), scope it
+ * to the authenticated user through `clientData`, and pass the result to
+ * `useLoadTranscript` in the browser.
+ *
+ * @example
+ * ```ts
+ * // actions.ts
+ * "use server";
+ * import { chat, defaultStorage } from "@trigger.dev/sdk/ai";
+ *
+ * export const loadTranscript = chat.createLoadTranscriptAction(defaultStorage, { limit: 50 });
+ * ```
+ */
+function createChatLoadTranscriptAction<TClientData = unknown>(
+  storage: TranscriptStorage<TClientData>,
+  options?: CreateChatLoadTranscriptActionOptions
+): (params: ChatLoadTranscriptParams<TClientData>) => Promise<TranscriptLoadResult> {
+  return async (params) => {
+    if (!params.chatId) {
+      throw new Error("chat.createLoadTranscriptAction: params.chatId is required.");
+    }
+    if (options?.apiClient) {
+      const { apiClient, ...rest } = options;
+      return apiClientManager.runWithConfig(apiClient, () =>
+        createChatLoadTranscriptAction(storage, rest)(params)
+      );
+    }
+    const limit = params.limit ?? options?.limit;
+    return storage.load(
+      { chatId: params.chatId, clientData: params.clientData as TClientData },
+      {
+        ...(limit !== undefined ? { limit } : {}),
+        ...(params.before !== undefined ? { before: params.before } : {}),
+      }
+    );
+  };
+}
+
 export const chat = {
   /** Create a chat agent. See {@link chatAgent}. */
   agent: chatAgent,
@@ -12670,6 +12788,8 @@ export const chat = {
   withClientData,
   /** Create a server-side helper for starting (or resuming) a Session for a chatId. See {@link createChatStartSessionAction}. */
   createStartSessionAction: createChatStartSessionAction,
+  /** Returns a server-side helper that reads a conversation from a transcript storage. */
+  createLoadTranscriptAction: createChatLoadTranscriptAction,
   /** Pipe a stream to the chat transport. See {@link pipeChat}. */
   pipe: pipeChat,
   /** Return from `onAction` to run a turn on the edited history. See {@link chatTurn}. */