Skip to content

Commit ed95a9e

Browse files
ericallamclaude
andcommitted
feat(sdk,core,webapp): transcript storage for chat.agent
Give chat.agent a pluggable TranscriptStorage seam so a run can own its conversation history across continuations: the version 2 transcript snapshot and dual-version dashboard reader, the storage option with a read API and conformance suite, run-tail recovery, compaction and injected-context persistence, and a dashboard TranscriptStorage over the agent's message rows. Includes the continuation-boot recovery hardening and the compaction/injection persistence fix. Rebased onto main and migrated to zod v4. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VG39FXXkFFU24U5EtJMwPi
1 parent 67b203d commit ed95a9e

51 files changed

Lines changed: 5698 additions & 414 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.changeset/transcript-storage.md

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
---
2+
"@trigger.dev/sdk": minor
3+
"@trigger.dev/core": minor
4+
---
5+
6+
`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).
7+
8+
```ts
9+
chat.agent({
10+
id: "my-chat",
11+
storage: myTranscriptStorage,
12+
run: async ({ messages, signal, streamText }) =>
13+
streamText({ model, messages, abortSignal: signal }),
14+
});
15+
```
16+
17+
`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.
18+
19+
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

Lines changed: 9 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,10 @@
11
import type { UIMessage } from "@ai-sdk/react";
2-
import { ChatSnapshotV1Schema, SSEStreamSubscription } from "@trigger.dev/core/v3";
2+
import { SSEStreamSubscription } from "@trigger.dev/core/v3";
33
import { useEffect, useMemo, useRef, useState } from "react";
44
import { Paragraph } from "~/components/primitives/Paragraph";
55
import { Spinner } from "~/components/primitives/Spinner";
66
import { AgentMessageView } from "~/components/runs/v3/agent/AgentMessageView";
7+
import { seedFromTranscriptSnapshot } from "~/components/runs/v3/agent/transcriptSnapshotSeed";
78
import { useAutoScrollToBottom } from "~/hooks/useAutoScrollToBottom";
89
import { useEnvironment } from "~/hooks/useEnvironment";
910
import { useOrganization } from "~/hooks/useOrganizations";
@@ -392,30 +393,17 @@ function useAgentSessionMessages({
392393
const resp = await fetch(url, { signal: abort.signal });
393394
if (!resp.ok) return undefined;
394395
const json = (await resp.json()) as unknown;
395-
const parsed = ChatSnapshotV1Schema.safeParse(json);
396-
if (!parsed.success) return undefined;
397-
const snapshot = parsed.data;
398-
// Preserve the snapshot's array order in the final render by
399-
// giving each message a unique, monotonically increasing
400-
// timestamp from `(savedAt - count + index)`. Real chunk
401-
// timestamps from the SSE path use S2 arrival ms (positive
402-
// numbers in the present), so anything below `savedAt` sorts
403-
// before live chunks while preserving snapshot order among
404-
// themselves.
405-
const count = snapshot.messages.length;
406-
snapshot.messages.forEach((raw, i) => {
407-
const message = raw as UIMessage;
408-
if (!message?.id) return;
396+
const seed = seedFromTranscriptSnapshot(json);
397+
if (!seed) return undefined;
398+
for (const { id, message, timestamp } of seed.messages) {
409399
// The snapshot's seed wins over the task-payload seed for any
410400
// overlapping ids (the snapshot represents the agent's
411401
// canonical accumulator, post-turn).
412-
pendingRef.current.set(message.id, message);
413-
if (!timestampsRef.current.has(message.id)) {
414-
timestampsRef.current.set(message.id, snapshot.savedAt - count + i);
415-
}
416-
});
402+
pendingRef.current.set(id, message);
403+
timestampsRef.current.set(id, timestamp);
404+
}
417405
scheduleFlush.current();
418-
return snapshot.lastOutEventId;
406+
return seed.lastOutEventId;
419407
} catch {
420408
// 404 / network / parse / abort — fall back to seq=0 SSE
421409
return undefined;
Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
import { describe, expect, it } from "vitest";
2+
import { seedFromTranscriptSnapshot } from "./transcriptSnapshotSeed";
3+
4+
const user = { id: "u-1", role: "user", parts: [{ type: "text", text: "hello" }] };
5+
const assistant = { id: "a-1", role: "assistant", parts: [{ type: "text", text: "world" }] };
6+
7+
describe("seedFromTranscriptSnapshot", () => {
8+
it("seeds from a version 1 snapshot in array order", () => {
9+
const seed = seedFromTranscriptSnapshot({
10+
version: 1,
11+
savedAt: 1_000,
12+
messages: [user, assistant],
13+
lastOutEventId: "42",
14+
lastInEventId: "7",
15+
});
16+
17+
expect(seed).toBeDefined();
18+
expect(seed!.lastOutEventId).toBe("42");
19+
expect(seed!.messages.map((m) => m.id)).toEqual(["u-1", "a-1"]);
20+
expect(seed!.messages.map((m) => m.timestamp)).toEqual([998, 999]);
21+
expect(seed!.messages[1]!.message).toEqual(assistant);
22+
});
23+
24+
it("seeds from a version 2 snapshot, unwrapping the message envelope", () => {
25+
const seed = seedFromTranscriptSnapshot({
26+
version: 2,
27+
savedAt: 1_000,
28+
messages: [
29+
{ id: "u-1", final: true, message: user },
30+
{ id: "a-1", final: false, message: assistant },
31+
],
32+
state: { summary: "irrelevant to rendering" },
33+
lastOutEventId: "42",
34+
lastInEventId: "7",
35+
});
36+
37+
expect(seed).toBeDefined();
38+
expect(seed!.lastOutEventId).toBe("42");
39+
expect(seed!.messages.map((m) => m.id)).toEqual(["u-1", "a-1"]);
40+
expect(seed!.messages.map((m) => m.timestamp)).toEqual([998, 999]);
41+
expect(seed!.messages[1]!.message).toEqual(assistant);
42+
});
43+
44+
it("skips version 1 entries without an id but keeps the others' positions", () => {
45+
const seed = seedFromTranscriptSnapshot({
46+
version: 1,
47+
savedAt: 1_000,
48+
messages: [{ role: "user", parts: [] }, assistant],
49+
});
50+
51+
expect(seed!.messages.map((m) => m.id)).toEqual(["a-1"]);
52+
expect(seed!.messages[0]!.timestamp).toBe(999);
53+
expect(seed!.lastOutEventId).toBeUndefined();
54+
});
55+
56+
it("returns undefined for an unknown version or a non-snapshot body", () => {
57+
expect(seedFromTranscriptSnapshot({ version: 3, savedAt: 1, messages: [] })).toBeUndefined();
58+
expect(seedFromTranscriptSnapshot({ error: "not found" })).toBeUndefined();
59+
expect(seedFromTranscriptSnapshot(null)).toBeUndefined();
60+
expect(seedFromTranscriptSnapshot("[]")).toBeUndefined();
61+
});
62+
});
Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
import type { UIMessage } from "@ai-sdk/react";
2+
import { parseTranscriptSnapshot } from "@trigger.dev/core/v3";
3+
4+
export type TranscriptSnapshotSeed = {
5+
messages: Array<{ id: string; message: UIMessage; timestamp: number }>;
6+
lastOutEventId: string | undefined;
7+
};
8+
9+
/**
10+
* Turn a fetched chat-snapshot blob into the messages the AgentView seeds
11+
* before it opens the `.out` subscription.
12+
*
13+
* Each message gets a unique, monotonically increasing timestamp from
14+
* `(savedAt - count + index)`. Live chunk timestamps are S2 arrival
15+
* milliseconds in the present, so anything below `savedAt` sorts before
16+
* live chunks while preserving the snapshot's own order.
17+
*
18+
* Reads both snapshot versions through `parseTranscriptSnapshot`. Returns
19+
* `undefined` for anything that is not a snapshot this reader understands;
20+
* the caller then falls back to the seq=0 SSE.
21+
*/
22+
export function seedFromTranscriptSnapshot(json: unknown): TranscriptSnapshotSeed | undefined {
23+
const snapshot = parseTranscriptSnapshot<UIMessage>(json);
24+
if (!snapshot) return undefined;
25+
const count = snapshot.messages.length;
26+
const messages = snapshot.messages.map((entry, i) => ({
27+
id: entry.id,
28+
message: entry.message,
29+
timestamp: snapshot.savedAt - count + i,
30+
}));
31+
return { messages, lastOutEventId: snapshot.lastOutEventId };
32+
}
Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
import { json } from "@remix-run/server-runtime";
2+
import { pageTranscriptEntries, parseTranscriptSnapshot } from "@trigger.dev/core/v3";
3+
import { z } from "zod/v4";
4+
import { $replica } from "~/db.server";
5+
import { chatSnapshotStorageKey } from "~/services/realtime/chatSnapshot.server";
6+
import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
7+
import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
8+
import { downloadPacketFromObjectStore } from "~/v3/objectStore.server";
9+
import { logger } from "~/services/logger.server";
10+
11+
const ParamsSchema = z.object({
12+
sessionId: z.string(),
13+
});
14+
15+
const SearchParamsSchema = z.object({
16+
limit: z.coerce.number().int().min(1).max(1000).optional(),
17+
before: z.string().optional(),
18+
});
19+
20+
function sessionResource(
21+
paramId: string,
22+
session: { friendlyId: string; externalId: string | null } | null | undefined
23+
) {
24+
const ids = new Set<string>([paramId]);
25+
if (session) {
26+
ids.add(session.friendlyId);
27+
if (session.externalId) ids.add(session.externalId);
28+
}
29+
return anyResource([...ids].map((id) => ({ type: "sessions" as const, id })));
30+
}
31+
32+
function isObjectNotFound(error: unknown): boolean {
33+
if (!error) return false;
34+
const name = (error as { name?: unknown }).name;
35+
if (name === "NoSuchKey" || name === "NotFound") return true;
36+
const status = (error as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode;
37+
if (status === 404) return true;
38+
const message = error instanceof Error ? error.message : String(error);
39+
return /not found|nosuchkey|404|does not exist/i.test(message);
40+
}
41+
42+
export const loader = createLoaderApiRoute(
43+
{
44+
params: ParamsSchema,
45+
searchParams: SearchParamsSchema,
46+
corsStrategy: "none",
47+
findResource: async (params, auth) =>
48+
resolveSessionByIdOrExternalId($replica, auth.environment.id, params.sessionId),
49+
authorization: {
50+
action: "read",
51+
resource: (session, params) => sessionResource(params.sessionId, session),
52+
},
53+
},
54+
async ({ authentication, resource: session, searchParams }) => {
55+
if (!session) {
56+
return json({ error: "Session not found" }, { status: 404 });
57+
}
58+
59+
let body: unknown;
60+
try {
61+
const packet = await downloadPacketFromObjectStore(
62+
{ dataType: "application/store", data: chatSnapshotStorageKey(session) },
63+
authentication.environment
64+
);
65+
body = typeof packet.data === "string" ? JSON.parse(packet.data) : undefined;
66+
} catch (error) {
67+
// A missing blob is a valid empty transcript (a session that has not
68+
// saved yet). Any other read failure must NOT look like an empty chat:
69+
// return an error so the client falls back to the whole-blob read
70+
// instead of rendering a saved conversation as empty.
71+
if (isObjectNotFound(error)) {
72+
return json({ messages: [], state: null });
73+
}
74+
logger.error("transcript endpoint: snapshot read failed", {
75+
sessionId: session.friendlyId,
76+
error: error instanceof Error ? error.message : String(error),
77+
});
78+
return json({ error: "Failed to read transcript" }, { status: 502 });
79+
}
80+
81+
const snapshot = parseTranscriptSnapshot(body);
82+
if (!snapshot) {
83+
return json({ messages: [], state: null });
84+
}
85+
86+
const page = pageTranscriptEntries(snapshot.messages, searchParams);
87+
return json({
88+
messages: page.entries.map((entry) => entry.message),
89+
state: snapshot.state,
90+
cursors: {
91+
lastOutEventId: snapshot.lastOutEventId,
92+
lastInEventId: snapshot.lastInEventId,
93+
},
94+
nextCursor: page.nextCursor,
95+
});
96+
}
97+
);

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

Lines changed: 17 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,8 @@
1-
// Plan F.3: integration test that round-trips a `ChatSnapshotV1` blob
2-
// through the SDK's snapshot helpers + a real MinIO backing store. Mirrors
3-
// the testcontainer pattern from `objectStore.test.ts`.
4-
//
5-
// What this verifies end-to-end:
6-
// - SDK's `writeChatSnapshot` calls `apiClient.createUploadPayloadUrl`
7-
// to mint a presigned PUT, then PUTs JSON to it.
8-
// - SDK's `readChatSnapshot` calls `apiClient.getPayloadUrl` to mint a
9-
// presigned GET, then fetches and parses.
10-
// - The webapp's `generatePresignedUrl` produces URLs MinIO accepts.
11-
// - The blob round-trips with `version: 1` shape preserved.
12-
// - 404 (no snapshot for a fresh session) returns `undefined`, not an
13-
// error.
14-
//
15-
// This is the integration safety net behind the unit tests in
16-
// `packages/trigger-sdk/test/chat-snapshot.test.ts` — those tests mock
17-
// `fetch`; this one drives a real S3-compatible backend.
18-
191
import { postgresAndMinioTest } from "@internal/testcontainers";
20-
import { apiClientManager } from "@trigger.dev/core/v3";
2+
import { apiClientManager, type TranscriptSnapshotV2 } from "@trigger.dev/core/v3";
213
import {
224
__readChatSnapshotProductionPathForTests as readChatSnapshot,
235
__writeChatSnapshotProductionPathForTests as writeChatSnapshot,
24-
type ChatSnapshotV1,
256
} from "@trigger.dev/sdk/ai";
267
import type { UIMessage } from "ai";
278
import { afterEach, describe, expect, vi } from "vitest";
@@ -35,22 +16,24 @@ vi.setConfig({ testTimeout: 60_000 });
3516

3617
function makeSnapshot(
3718
opts: { messages?: UIMessage[]; lastOutEventId?: string } = {}
38-
): ChatSnapshotV1 {
19+
): TranscriptSnapshotV2 {
20+
const messages = opts.messages ?? [
21+
{
22+
id: "u-1",
23+
role: "user",
24+
parts: [{ type: "text", text: "hello" }],
25+
},
26+
{
27+
id: "a-1",
28+
role: "assistant",
29+
parts: [{ type: "text", text: "world" }],
30+
},
31+
];
3932
return {
40-
version: 1,
33+
version: 2,
4134
savedAt: 1_700_000_000_000,
42-
messages: opts.messages ?? [
43-
{
44-
id: "u-1",
45-
role: "user",
46-
parts: [{ type: "text", text: "hello" }],
47-
},
48-
{
49-
id: "a-1",
50-
role: "assistant",
51-
parts: [{ type: "text", text: "world" }],
52-
},
53-
],
35+
messages: messages.map((message) => ({ id: message.id, final: true, message })),
36+
state: null,
5437
lastOutEventId: opts.lastOutEventId ?? "evt-42",
5538
};
5639
}

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

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,10 @@
2424
// through it), even though the replay path itself doesn't read from S3.
2525

2626
import { postgresAndMinioTest } from "@internal/testcontainers";
27-
import { apiClientManager } from "@trigger.dev/core/v3";
27+
import { apiClientManager, type TranscriptSnapshotV2 } from "@trigger.dev/core/v3";
2828
import {
2929
__readChatSnapshotProductionPathForTests as readChatSnapshot,
3030
__replaySessionOutTailProductionPathForTests as replaySessionOutTail,
31-
type ChatSnapshotV1,
3231
} from "@trigger.dev/sdk/ai";
3332
import type { UIMessageChunk } from "ai";
3433
import { afterEach, describe, expect, vi } from "vitest";
@@ -265,13 +264,26 @@ describe("replay after crash (MinIO + SDK helpers)", () => {
265264

266265
// Pre-write a snapshot to MinIO via real apiClient stub.
267266
const sessionId = "sess_merge_round_trip";
268-
const snapshot: ChatSnapshotV1 = {
269-
version: 1,
267+
const snapshot: TranscriptSnapshotV2 = {
268+
version: 2,
270269
savedAt: 1_700_000_000_000,
271270
messages: [
272-
{ id: "u-1", role: "user", parts: [{ type: "text", text: "hi" }] },
273-
{ id: "a-1", role: "assistant", parts: [{ type: "text", text: "stale-assistant" }] },
271+
{
272+
id: "u-1",
273+
final: true,
274+
message: { id: "u-1", role: "user", parts: [{ type: "text", text: "hi" }] },
275+
},
276+
{
277+
id: "a-1",
278+
final: true,
279+
message: {
280+
id: "a-1",
281+
role: "assistant",
282+
parts: [{ type: "text", text: "stale-assistant" }],
283+
},
284+
},
274285
],
286+
state: null,
275287
lastOutEventId: "evt-prev",
276288
};
277289

docs/ai-chat/actions.mdx

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -83,9 +83,9 @@ onAction: async ({ action }) => {
8383

8484
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.
8585

86-
**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.
86+
**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.
8787

88-
**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.
88+
**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.
8989

9090
```ts
9191
onAction: async ({ action, chatId }) => {

docs/ai-chat/background-injection.mdx

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -250,6 +250,8 @@ chat.inject(messages: ModelMessage[]): void
250250

251251
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.
252252

253+
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.
254+
253255
**Parameters:**
254256

255257
| Parameter | Type | Description |

0 commit comments

Comments
 (0)