Skip to content

Commit 035e432

Browse files
committed
feat(core,webapp): version 2 transcript snapshot and a dual-version dashboard reader
Adds TranscriptSnapshotV2 (entries of {id, final, message} plus an opaque state record) and parseTranscriptSnapshot(), which reads a version 1 or version 2 blob into the version 2 shape. The Sessions dashboard seeds its transcript through that reader, so it keeps rendering history once the SDK starts writing version 2. The SDK still writes version 1 in this change.
1 parent bf66457 commit 035e432

5 files changed

Lines changed: 281 additions & 20 deletions

File tree

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

Lines changed: 10 additions & 20 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,19 @@ 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);
402+
pendingRef.current.set(id, message);
403+
if (!timestampsRef.current.has(id)) {
404+
timestampsRef.current.set(id, timestamp);
415405
}
416-
});
406+
}
417407
scheduleFlush.current();
418-
return snapshot.lastOutEventId;
408+
return seed.lastOutEventId;
419409
} catch {
420410
// 404 / network / parse / abort — fall back to seq=0 SSE
421411
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+
}

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

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,98 @@ export const ChatSnapshotV1Schema = z.object({
5050
lastInEventId: z.string().optional(),
5151
});
5252

53+
/**
54+
* One transcript entry in a version 2 snapshot. `id` duplicates
55+
* `message.id` so a reader can address entries without inspecting the
56+
* message body; `final` is false for a partial assistant message captured
57+
* by an errored or stopped turn.
58+
*/
59+
export type TranscriptSnapshotEntry<TUIMessage extends UIMessage = UIMessage> = {
60+
id: string;
61+
final: boolean;
62+
message: TUIMessage;
63+
};
64+
65+
/**
66+
* Version 2 of the persisted transcript blob. Entries are ordered by array
67+
* position. `state` is an opaque record the runtime uses for compaction and
68+
* other cross-run bookkeeping; `null` when nothing has been recorded.
69+
*
70+
* Readers must accept version 1 as well; writers only emit version 2. Use
71+
* {@link parseTranscriptSnapshot} to read either.
72+
*/
73+
export type TranscriptSnapshotV2<TUIMessage extends UIMessage = UIMessage> = {
74+
version: 2;
75+
savedAt: number;
76+
messages: TranscriptSnapshotEntry<TUIMessage>[];
77+
state: unknown | null;
78+
lastOutEventId?: string;
79+
lastInEventId?: string;
80+
};
81+
82+
export const TranscriptSnapshotV2Schema = z.object({
83+
version: z.literal(2),
84+
savedAt: z.number(),
85+
messages: z.array(
86+
z.object({
87+
id: z.string(),
88+
final: z.boolean(),
89+
message: z.unknown(),
90+
})
91+
),
92+
state: z.unknown().nullable(),
93+
lastOutEventId: z.string().optional(),
94+
lastInEventId: z.string().optional(),
95+
});
96+
97+
/**
98+
* Parse a fetched snapshot blob of any known version into the version 2
99+
* shape. A version 1 blob is upgraded in memory: every message becomes a
100+
* `final: true` entry keyed by its `id` (entries without a string `id` are
101+
* dropped) and `state` is `null`. Returns `undefined` for an unknown
102+
* version or a body that is not a snapshot; callers treat that as
103+
* "no snapshot".
104+
*/
105+
export function parseTranscriptSnapshot<TUIMessage extends UIMessage = UIMessage>(
106+
input: unknown
107+
): TranscriptSnapshotV2<TUIMessage> | undefined {
108+
const v2 = TranscriptSnapshotV2Schema.safeParse(input);
109+
if (v2.success) {
110+
return {
111+
version: 2,
112+
savedAt: v2.data.savedAt,
113+
messages: v2.data.messages.map((entry) => ({
114+
id: entry.id,
115+
final: entry.final,
116+
message: entry.message as TUIMessage,
117+
})),
118+
state: v2.data.state ?? null,
119+
lastOutEventId: v2.data.lastOutEventId,
120+
lastInEventId: v2.data.lastInEventId,
121+
};
122+
}
123+
124+
const v1 = ChatSnapshotV1Schema.safeParse(input);
125+
if (v1.success) {
126+
const messages: TranscriptSnapshotEntry<TUIMessage>[] = [];
127+
for (const raw of v1.data.messages) {
128+
const id = (raw as { id?: unknown } | null)?.id;
129+
if (typeof id !== "string" || id.length === 0) continue;
130+
messages.push({ id, final: true, message: raw as TUIMessage });
131+
}
132+
return {
133+
version: 2,
134+
savedAt: v1.data.savedAt,
135+
messages,
136+
state: null,
137+
lastOutEventId: v1.data.lastOutEventId,
138+
lastInEventId: v1.data.lastInEventId,
139+
};
140+
}
141+
142+
return undefined;
143+
}
144+
53145
/**
54146
* S3 key suffix for a session's snapshot blob. The webapp's presigned
55147
* URL routes prefix this with `packets/{projectRef}/{envSlug}/`.
Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
import { describe, expect, it } from "vitest";
2+
import {
3+
parseTranscriptSnapshot,
4+
type ChatSnapshotV1,
5+
type TranscriptSnapshotV2,
6+
} from "../src/v3/sessionStreams/chatSnapshot.js";
7+
8+
const user = { id: "u-1", role: "user" as const, parts: [{ type: "text" as const, text: "hi" }] };
9+
const assistant = {
10+
id: "a-1",
11+
role: "assistant" as const,
12+
parts: [{ type: "text" as const, text: "hello" }],
13+
};
14+
15+
describe("parseTranscriptSnapshot", () => {
16+
it("returns a version 2 blob unchanged", () => {
17+
const blob: TranscriptSnapshotV2 = {
18+
version: 2,
19+
savedAt: 10,
20+
messages: [
21+
{ id: "u-1", final: true, message: user },
22+
{ id: "a-1", final: false, message: assistant },
23+
],
24+
state: { summary: "s", through: "u-1" },
25+
lastOutEventId: "9",
26+
lastInEventId: "3",
27+
};
28+
29+
expect(parseTranscriptSnapshot(blob)).toEqual(blob);
30+
});
31+
32+
it("normalises a version 2 blob with no state to state: null", () => {
33+
const parsed = parseTranscriptSnapshot({
34+
version: 2,
35+
savedAt: 10,
36+
messages: [],
37+
state: undefined,
38+
});
39+
40+
expect(parsed?.state).toBeNull();
41+
});
42+
43+
it("upgrades a version 1 blob: every message final, state null, cursors kept", () => {
44+
const blob: ChatSnapshotV1 = {
45+
version: 1,
46+
savedAt: 10,
47+
messages: [user, assistant],
48+
lastOutEventId: "9",
49+
lastInEventId: "3",
50+
};
51+
52+
expect(parseTranscriptSnapshot(blob)).toEqual({
53+
version: 2,
54+
savedAt: 10,
55+
messages: [
56+
{ id: "u-1", final: true, message: user },
57+
{ id: "a-1", final: true, message: assistant },
58+
],
59+
state: null,
60+
lastOutEventId: "9",
61+
lastInEventId: "3",
62+
});
63+
});
64+
65+
it("drops version 1 messages that have no string id", () => {
66+
const parsed = parseTranscriptSnapshot({
67+
version: 1,
68+
savedAt: 10,
69+
messages: [{ role: "user", parts: [] }, { id: 7, role: "user", parts: [] }, assistant, null],
70+
});
71+
72+
expect(parsed?.messages.map((m) => m.id)).toEqual(["a-1"]);
73+
});
74+
75+
it("returns undefined for unknown versions and non-snapshot bodies", () => {
76+
expect(parseTranscriptSnapshot({ version: 3, savedAt: 1, messages: [] })).toBeUndefined();
77+
expect(
78+
parseTranscriptSnapshot({ version: 2, savedAt: 1, messages: [{ id: "x" }] })
79+
).toBeUndefined();
80+
expect(parseTranscriptSnapshot({ version: 1, savedAt: "1", messages: [] })).toBeUndefined();
81+
expect(parseTranscriptSnapshot({ message: "Not Found" })).toBeUndefined();
82+
expect(parseTranscriptSnapshot(undefined)).toBeUndefined();
83+
expect(parseTranscriptSnapshot([])).toBeUndefined();
84+
});
85+
});

0 commit comments

Comments
 (0)