feat(sdk,core,webapp): transcript storage for chat.agent

.changeset/transcript-storage.md

@@ -0,0 +1,19 @@
+---
+"@trigger.dev/sdk": minor
+"@trigger.dev/core": minor
+---
+
+`chat.agent` persists a conversation through a `TranscriptStorage`: an adapter with `load` and `save` that the runtime drives after every turn, failed turn and history-changing action. The platform snapshot stays the default; bring your own to write the conversation to your database as it happens. Each save carries both the changes since the last one (so a row store writes only what changed, and an undo is one `truncateAfter`) and the whole transcript as it now stands (so a document store writes it as-is with no state of its own).
+
+```ts
+chat.agent({
+  id: "my-chat",
+  storage: myTranscriptStorage,
+  run: async ({ messages, signal, streamText }) =>
+    streamText({ model, messages, abortSignal: signal }),
+});
+```
+
+`chat.createLoadTranscriptAction(storage)` and `useLoadTranscript` read the conversation back the same way for every storage, and `runTranscriptStorageTests` from `@trigger.dev/sdk/ai/test` checks an implementation against the contract.
+
+Compaction summaries and `chat.inject` context now survive a continuation run, and crash recovery runs for every agent, including one that owns its own context. `hydrateMessages` is deprecated in favour of `loadContext` on a storage. The snapshot format is now version 2, which older SDK versions cannot read.

apps/webapp/app/components/runs/v3/agent/AgentView.tsx

@@ -1,9 +1,10 @@
 import type { UIMessage } from "@ai-sdk/react";
-import { ChatSnapshotV1Schema, SSEStreamSubscription } from "@trigger.dev/core/v3";
+import { SSEStreamSubscription } from "@trigger.dev/core/v3";
 import { useEffect, useMemo, useRef, useState } from "react";
 import { Paragraph } from "~/components/primitives/Paragraph";
 import { Spinner } from "~/components/primitives/Spinner";
 import { AgentMessageView } from "~/components/runs/v3/agent/AgentMessageView";
+import { seedFromTranscriptSnapshot } from "~/components/runs/v3/agent/transcriptSnapshotSeed";
 import { useAutoScrollToBottom } from "~/hooks/useAutoScrollToBottom";
 import { useEnvironment } from "~/hooks/useEnvironment";
 import { useOrganization } from "~/hooks/useOrganizations";
@@ -392,30 +393,17 @@ function useAgentSessionMessages({
         const resp = await fetch(url, { signal: abort.signal });
         if (!resp.ok) return undefined;
         const json = (await resp.json()) as unknown;
-        const parsed = ChatSnapshotV1Schema.safeParse(json);
-        if (!parsed.success) return undefined;
-        const snapshot = parsed.data;
-        // Preserve the snapshot's array order in the final render by
-        // giving each message a unique, monotonically increasing
-        // timestamp from `(savedAt - count + index)`. Real chunk
-        // timestamps from the SSE path use S2 arrival ms (positive
-        // numbers in the present), so anything below `savedAt` sorts
-        // before live chunks while preserving snapshot order among
-        // themselves.
-        const count = snapshot.messages.length;
-        snapshot.messages.forEach((raw, i) => {
-          const message = raw as UIMessage;
-          if (!message?.id) return;
+        const seed = seedFromTranscriptSnapshot(json);
+        if (!seed) return undefined;
+        for (const { id, message, timestamp } of seed.messages) {
           // The snapshot's seed wins over the task-payload seed for any
           // overlapping ids (the snapshot represents the agent's
           // canonical accumulator, post-turn).
-          pendingRef.current.set(message.id, message);
-          if (!timestampsRef.current.has(message.id)) {
-            timestampsRef.current.set(message.id, snapshot.savedAt - count + i);
-          }
-        });
+          pendingRef.current.set(id, message);
+          timestampsRef.current.set(id, timestamp);
+        }
         scheduleFlush.current();
-        return snapshot.lastOutEventId;
+        return seed.lastOutEventId;
       } catch {
         // 404 / network / parse / abort — fall back to seq=0 SSE
         return undefined;

apps/webapp/app/components/runs/v3/agent/transcriptSnapshotSeed.test.ts

@@ -0,0 +1,62 @@
+import { describe, expect, it } from "vitest";
+import { seedFromTranscriptSnapshot } from "./transcriptSnapshotSeed";
+
+const user = { id: "u-1", role: "user", parts: [{ type: "text", text: "hello" }] };
+const assistant = { id: "a-1", role: "assistant", parts: [{ type: "text", text: "world" }] };
+
+describe("seedFromTranscriptSnapshot", () => {
+  it("seeds from a version 1 snapshot in array order", () => {
+    const seed = seedFromTranscriptSnapshot({
+      version: 1,
+      savedAt: 1_000,
+      messages: [user, assistant],
+      lastOutEventId: "42",
+      lastInEventId: "7",
+    });
+
+    expect(seed).toBeDefined();
+    expect(seed!.lastOutEventId).toBe("42");
+    expect(seed!.messages.map((m) => m.id)).toEqual(["u-1", "a-1"]);
+    expect(seed!.messages.map((m) => m.timestamp)).toEqual([998, 999]);
+    expect(seed!.messages[1]!.message).toEqual(assistant);
+  });
+
+  it("seeds from a version 2 snapshot, unwrapping the message envelope", () => {
+    const seed = seedFromTranscriptSnapshot({
+      version: 2,
+      savedAt: 1_000,
+      messages: [
+        { id: "u-1", final: true, message: user },
+        { id: "a-1", final: false, message: assistant },
+      ],
+      state: { summary: "irrelevant to rendering" },
+      lastOutEventId: "42",
+      lastInEventId: "7",
+    });
+
+    expect(seed).toBeDefined();
+    expect(seed!.lastOutEventId).toBe("42");
+    expect(seed!.messages.map((m) => m.id)).toEqual(["u-1", "a-1"]);
+    expect(seed!.messages.map((m) => m.timestamp)).toEqual([998, 999]);
+    expect(seed!.messages[1]!.message).toEqual(assistant);
+  });
+
+  it("skips version 1 entries without an id but keeps the others' positions", () => {
+    const seed = seedFromTranscriptSnapshot({
+      version: 1,
+      savedAt: 1_000,
+      messages: [{ role: "user", parts: [] }, assistant],
+    });
+
+    expect(seed!.messages.map((m) => m.id)).toEqual(["a-1"]);
+    expect(seed!.messages[0]!.timestamp).toBe(999);
+    expect(seed!.lastOutEventId).toBeUndefined();
+  });
+
+  it("returns undefined for an unknown version or a non-snapshot body", () => {
+    expect(seedFromTranscriptSnapshot({ version: 3, savedAt: 1, messages: [] })).toBeUndefined();
+    expect(seedFromTranscriptSnapshot({ error: "not found" })).toBeUndefined();
+    expect(seedFromTranscriptSnapshot(null)).toBeUndefined();
+    expect(seedFromTranscriptSnapshot("[]")).toBeUndefined();
+  });
+});

apps/webapp/app/components/runs/v3/agent/transcriptSnapshotSeed.ts

@@ -0,0 +1,32 @@
+import type { UIMessage } from "@ai-sdk/react";
+import { parseTranscriptSnapshot } from "@trigger.dev/core/v3";
+
+export type TranscriptSnapshotSeed = {
+  messages: Array<{ id: string; message: UIMessage; timestamp: number }>;
+  lastOutEventId: string | undefined;
+};
+
+/**
+ * Turn a fetched chat-snapshot blob into the messages the AgentView seeds
+ * before it opens the `.out` subscription.
+ *
+ * Each message gets a unique, monotonically increasing timestamp from
+ * `(savedAt - count + index)`. Live chunk timestamps are S2 arrival
+ * milliseconds in the present, so anything below `savedAt` sorts before
+ * live chunks while preserving the snapshot's own order.
+ *
+ * Reads both snapshot versions through `parseTranscriptSnapshot`. Returns
+ * `undefined` for anything that is not a snapshot this reader understands;
+ * the caller then falls back to the seq=0 SSE.
+ */
+export function seedFromTranscriptSnapshot(json: unknown): TranscriptSnapshotSeed | undefined {
+  const snapshot = parseTranscriptSnapshot<UIMessage>(json);
+  if (!snapshot) return undefined;
+  const count = snapshot.messages.length;
+  const messages = snapshot.messages.map((entry, i) => ({
+    id: entry.id,
+    message: entry.message,
+    timestamp: snapshot.savedAt - count + i,
+  }));
+  return { messages, lastOutEventId: snapshot.lastOutEventId };
+}

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

@@ -0,0 +1,97 @@
+import { json } from "@remix-run/server-runtime";
+import { pageTranscriptEntries, parseTranscriptSnapshot } from "@trigger.dev/core/v3";
+import { z } from "zod/v4";
+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";
+import { logger } from "~/services/logger.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 })));
+}
+
+function isObjectNotFound(error: unknown): boolean {
+  if (!error) return false;
+  const name = (error as { name?: unknown }).name;
+  if (name === "NoSuchKey" || name === "NotFound") return true;
+  const status = (error as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode;
+  if (status === 404) return true;
+  const message = error instanceof Error ? error.message : String(error);
+  return /not found|nosuchkey|404|does not exist/i.test(message);
+}
+
+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 (error) {
+      // A missing blob is a valid empty transcript (a session that has not
+      // saved yet). Any other read failure must NOT look like an empty chat:
+      // return an error so the client falls back to the whole-blob read
+      // instead of rendering a saved conversation as empty.
+      if (isObjectNotFound(error)) {
+        return json({ messages: [], state: null });
+      }
+      logger.error("transcript endpoint: snapshot read failed", {
+        sessionId: session.friendlyId,
+        error: error instanceof Error ? error.message : String(error),
+      });
+      return json({ error: "Failed to read transcript" }, { status: 502 });
+    }
+
+    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,
+    });
+  }
+);

apps/webapp/test/chat-snapshot-integration.test.ts

@@ -1,27 +1,8 @@
-// Plan F.3: integration test that round-trips a `ChatSnapshotV1` blob
-// through the SDK's snapshot helpers + a real MinIO backing store. Mirrors
-// the testcontainer pattern from `objectStore.test.ts`.
-//
-// What this verifies end-to-end:
-//   - SDK's `writeChatSnapshot` calls `apiClient.createUploadPayloadUrl`
-//     to mint a presigned PUT, then PUTs JSON to it.
-//   - SDK's `readChatSnapshot` calls `apiClient.getPayloadUrl` to mint a
-//     presigned GET, then fetches and parses.
-//   - The webapp's `generatePresignedUrl` produces URLs MinIO accepts.
-//   - The blob round-trips with `version: 1` shape preserved.
-//   - 404 (no snapshot for a fresh session) returns `undefined`, not an
-//     error.
-//
-// This is the integration safety net behind the unit tests in
-// `packages/trigger-sdk/test/chat-snapshot.test.ts` — those tests mock
-// `fetch`; this one drives a real S3-compatible backend.
-
 import { postgresAndMinioTest } from "@internal/testcontainers";
-import { apiClientManager } from "@trigger.dev/core/v3";
+import { apiClientManager, type TranscriptSnapshotV2 } from "@trigger.dev/core/v3";
 import {
   __readChatSnapshotProductionPathForTests as readChatSnapshot,
   __writeChatSnapshotProductionPathForTests as writeChatSnapshot,
-  type ChatSnapshotV1,
 } from "@trigger.dev/sdk/ai";
 import type { UIMessage } from "ai";
 import { afterEach, describe, expect, vi } from "vitest";
@@ -35,22 +16,24 @@ vi.setConfig({ testTimeout: 60_000 });
 
 function makeSnapshot(
   opts: { messages?: UIMessage[]; lastOutEventId?: string } = {}
-): ChatSnapshotV1 {
+): TranscriptSnapshotV2 {
+  const messages = opts.messages ?? [
+    {
+      id: "u-1",
+      role: "user",
+      parts: [{ type: "text", text: "hello" }],
+    },
+    {
+      id: "a-1",
+      role: "assistant",
+      parts: [{ type: "text", text: "world" }],
+    },
+  ];
   return {
-    version: 1,
+    version: 2,
     savedAt: 1_700_000_000_000,
-    messages: opts.messages ?? [
-      {
-        id: "u-1",
-        role: "user",
-        parts: [{ type: "text", text: "hello" }],
-      },
-      {
-        id: "a-1",
-        role: "assistant",
-        parts: [{ type: "text", text: "world" }],
-      },
-    ],
+    messages: messages.map((message) => ({ id: message.id, final: true, message })),
+    state: null,
     lastOutEventId: opts.lastOutEventId ?? "evt-42",
   };
 }

apps/webapp/test/replay-after-crash.test.ts

@@ -24,11 +24,10 @@
 // through it), even though the replay path itself doesn't read from S3.
 
 import { postgresAndMinioTest } from "@internal/testcontainers";
-import { apiClientManager } from "@trigger.dev/core/v3";
+import { apiClientManager, type TranscriptSnapshotV2 } from "@trigger.dev/core/v3";
 import {
   __readChatSnapshotProductionPathForTests as readChatSnapshot,
   __replaySessionOutTailProductionPathForTests as replaySessionOutTail,
-  type ChatSnapshotV1,
 } from "@trigger.dev/sdk/ai";
 import type { UIMessageChunk } from "ai";
 import { afterEach, describe, expect, vi } from "vitest";
@@ -265,13 +264,26 @@ describe("replay after crash (MinIO + SDK helpers)", () => {
 
       // Pre-write a snapshot to MinIO via real apiClient stub.
       const sessionId = "sess_merge_round_trip";
-      const snapshot: ChatSnapshotV1 = {
-        version: 1,
+      const snapshot: TranscriptSnapshotV2 = {
+        version: 2,
         savedAt: 1_700_000_000_000,
         messages: [
-          { id: "u-1", role: "user", parts: [{ type: "text", text: "hi" }] },
-          { id: "a-1", role: "assistant", parts: [{ type: "text", text: "stale-assistant" }] },
+          {
+            id: "u-1",
+            final: true,
+            message: { id: "u-1", role: "user", parts: [{ type: "text", text: "hi" }] },
+          },
+          {
+            id: "a-1",
+            final: true,
+            message: {
+              id: "a-1",
+              role: "assistant",
+              parts: [{ type: "text", text: "stale-assistant" }],
+            },
+          },
         ],
+        state: null,
         lastOutEventId: "evt-prev",
       };
 

docs/ai-chat/actions.mdx

@@ -83,9 +83,9 @@ onAction: async ({ action }) => {
 
 An action that returns nothing does not fire `onTurnComplete`, and that is where an app that owns its own transcript normally writes. What that means depends on which persistence model you use.
 
-**Platform-managed** (no `hydrateMessages`): nothing to do. After an action that changed the conversation, the runtime writes the snapshot, so the edit survives the run ending. An action that returns `chat.turn()` is followed by a turn, which persists its answer the way every turn does.
+**Transcript storage** (the default, or your own `storage`): nothing to do. After an action that changed the conversation, the runtime hands the storage a changeset with `reason: "action"`. An undo is one `truncateAfter`; a regenerate is a `truncateAfter` followed by the new answer's `put` when the turn completes; an edit is a `put` for the edited id. The changeset carries the same resume cursors as the last turn. See [Transcript storage](/ai-chat/transcript-storage#what-the-runtime-saves). An action that returns `chat.turn()` is followed by a turn, which persists its answer the way every turn does.
 
-**Your own store** (`hydrateMessages` registered): the runtime deliberately does not write, because your store is the source of truth. A history edit lives only in the running worker until you persist it, and a continuation rehydrates from your store, not from what the worker had in memory. Mirror each edit in your store, not only additions: a regenerate is a delete *and* an insert. The answer that follows `chat.turn()` reaches your store through `onTurnComplete`, like any turn's answer.
+**Your own store through the deprecated `hydrateMessages`**: the runtime deliberately does not write, because your store is the source of truth. A history edit lives only in the running worker until you persist it, and a continuation rehydrates from your store, not from what the worker had in memory. Mirror each edit in your store, not only additions: a regenerate is a delete *and* an insert. The answer that follows `chat.turn()` reaches your store through `onTurnComplete`, like any turn's answer.
 
 ```ts
 onAction: async ({ action, chatId }) => {

docs/ai-chat/background-injection.mdx

@@ -250,6 +250,8 @@ chat.inject(messages: ModelMessage[]): void
 
 Queue model messages for injection at the next opportunity. Messages persist across the idle wait between turns, and are not reset when a new turn starts.
 
+Lifetime: a conversational message (`role: "user"` or `"assistant"`) becomes part of the model's context from the next turn onward, for the rest of the conversation. It is written to the [transcript storage](/ai-chat/transcript-storage)'s `state`, anchored to the message it followed, so it survives a continuation run and comes back in the same place. It does not appear in the UI transcript. A history edit that rebuilds the context drops it. This holds however the message reached the model: drained before `run()` or at a step boundary inside a multi-step turn. A message that is still queued when the run ends (injected from the last `onTurnComplete` before an exit, for example) is carried in the storage's `state` too and is queued again when the next run boots, so it reaches the next turn. A `role: "system"` message is appended to the instructions for the next turn only and is consumed once. Injecting the same notice every turn adds a copy every turn; dedupe on your side.
+
 **Parameters:**
 
 | Parameter | Type | Description |