feat(sdk): route chat.agent transcript persistence through a TranscriptStorage seam (#4893)

## Summary

Introduces the `TranscriptStorage` seam inside `chat.agent` and makes
the built-in snapshot writer its default implementation. No public
option yet; the default path behaves as before, apart from the blob now
being written as version 2.

## Design

A storage has `load` (called once when a run boots to continue a
conversation) and `save` (called after every turn, failed turn and
history-changing action). `save` receives a changeset of id-addressed
operations: `put` (upsert by message id), `remove`, `truncateAfter` and
`state`, plus the stream cursors the next boot resumes from.

The runtime keeps a shadow of the transcript it last handed to `save`
(ids in order plus a fingerprint per message) and diffs the accumulator
against it. A changed message in the common prefix is an in-place `put`;
anything past the prefix is one `truncateAfter` on the last common id
followed by `put`s in order. Applying the result reproduces the
accumulator exactly for any edit, and the common cases come out as the
natural operations: a turn is two `put`s, an undo is one
`truncateAfter`, a regenerate is a `truncateAfter` and a `put`. The
shadow only advances when `save` resolves, so a failed save is folded
into the next changeset; every operation is idempotent, so a retried
changeset converges.

Every changeset also carries the whole transcript as it stands after the
changes (entries with a `final` flag, plus the runtime's `state`). A
row-per-message store applies the changes; a store that keeps the
conversation as one document writes the transcript as-is and needs no
state of its own between saves. The default storage is the second kind:
it serialises the transcript and rewrites the blob, so a turn still
costs one PUT and no GET, and nothing is held in memory between saves.
The snapshot read and write helpers move to their own module so the
storage can import them without a cycle; the test seams keep their
import path.

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: ericallam

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",
       };
 

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

@@ -42,7 +42,7 @@ export type ChatSnapshotV1<TUIMessage = unknown> = {
  */
 export const ChatSnapshotV1Schema = z.object({
   version: z.literal(1),
-  savedAt: z.number(),
+  savedAt: z.number().optional(),
   messages: z.array(z.unknown()),
   lastOutEventId: z.string().optional(),
   lastInEventId: z.string().optional(),
@@ -79,7 +79,7 @@ export type TranscriptSnapshotV2<TUIMessage extends UIMessage = UIMessage> = {
 
 export const TranscriptSnapshotV2Schema = z.object({
   version: z.literal(2),
-  savedAt: z.number(),
+  savedAt: z.number().optional(),
   messages: z.array(
     z.object({
       id: z.string(),
@@ -100,6 +100,9 @@ export const TranscriptSnapshotV2Schema = z.object({
  * `message` are dropped; a version 2 entry whose `message.id` disagrees with
  * the envelope `id` is dropped too, since a reader keys by one and renders by
  * the other. A caller never sees an entry it would crash on or mis-order.
+ * A missing `savedAt` defaults to `0` rather than rejecting the whole blob:
+ * the field only orders snapshot history before live chunks, and dropping a
+ * whole conversation over an absent timestamp is the wrong failure mode.
  * Returns `undefined` for an unknown version or a body that is not a
  * snapshot; callers treat that as "no snapshot".
  */
@@ -117,7 +120,7 @@ export function parseTranscriptSnapshot<TUIMessage extends UIMessage = UIMessage
     }
     return {
       version: 2,
-      savedAt: v2.data.savedAt,
+      savedAt: v2.data.savedAt ?? 0,
       messages,
       state: v2.data.state ?? null,
       lastOutEventId: v2.data.lastOutEventId,
@@ -135,7 +138,7 @@ export function parseTranscriptSnapshot<TUIMessage extends UIMessage = UIMessage
     }
     return {
       version: 2,
-      savedAt: v1.data.savedAt,
+      savedAt: v1.data.savedAt ?? 0,
       messages,
       state: null,
       lastOutEventId: v1.data.lastOutEventId,

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

@@ -220,7 +220,11 @@ export class TestSessionStreamManager implements SessionStreamManager {
   }
 
   setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
-    this.seqNums.set(keyFor(sessionId, io), seqNum);
+    const key = keyFor(sessionId, io);
+    const current = this.seqNums.get(key);
+    if (current === undefined || seqNum > current) {
+      this.seqNums.set(key, seqNum);
+    }
   }
 
   consumeRecord(sessionId: string, io: SessionChannelIO, seqNum: number): void {