Visitar URL original
fix(sdk): keep the persisted chat cursor at the turn start while stre… · triggerdotdev/trigger.dev@9fd1de6 · GitHub
Skip to content

Commit 9fd1de6

Browse files
ericallamTrigger.dev RepoOps
ericallam
authored and
Trigger.dev RepoOps
committed
fix(sdk): keep the persisted chat cursor at the turn start while streaming
Fix reloading a chat mid-turn after its access token was refreshed. `TriggerChatTransport` persists a resume cursor (`lastEventId`) so a reloaded page can reconnect to `.out`. Sends persist it before the turn streams, so a reload normally replays the whole turn. A token refresh during a streaming turn (the 401 retry on a steering send or on an `.out` reconnect) persisted the latest record's cursor instead, from the middle of the turn. A reload then resumed past the turn's opening chunks, and `useChat` failed to rebuild the assistant message: "No tool invocation found for tool call ID". The transport now remembers the cursor from before each turn's first record, set when the turn is sent or, after a reconnect, when its first record arrives (so it also holds for apps that don't persist `isStreaming`). It persists that cursor in place of the live one until the turn ends, then persists the latest cursor as before. `getSession()` returns the same value, and the frontend docs describe it. Mono-RevId: 13ad0c6f4d89f44707f9683a6faa7b362268d9f4
1 parent 4451f4d commit 9fd1de6

4 files changed

Lines changed: 176 additions & 14 deletions

File tree

‎.changeset/chat-turn-start-cursor.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@trigger.dev/sdk": patch
3+
---
4+
5+
Fix reloading a chat mid-turn after its access token was refreshed. A token refresh during a streaming turn persisted a resume cursor from the middle of the turn, so a reload resumed past chunks the page needed and failed to rebuild the assistant message ("No tool invocation found for tool call ID"). The persisted cursor now stays at the start of the turn until it completes, so a reload replays the whole turn.

‎docs/ai-chat/frontend.mdx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -109,7 +109,7 @@ Every chat is backed by a durable Session — the row that owns the chat's runs,
109109
| Field | Type | Notes |
110110
| --- | --- | --- |
111111
| `publicAccessToken` | `string` | Session-scoped JWT (`read:sessions:{chatId} + write:sessions:{chatId}`). Refreshed automatically on 401/403 via `accessToken`. |
112-
| `lastEventId` | `string \| undefined` | Last SSE event received on `.out`. **Valid for the lifetime of the Session** — keep it across `endRun` / `requestUpgrade` / continuation-run boundaries; only clear when the Session itself closes. The cursor lets the next subscription open past the prior turn's stale `turn-complete` record. |
112+
| `lastEventId` | `string \| undefined` | Resume cursor on `.out`: the last SSE event received, or while a turn is streaming, the cursor from when that turn started, so a reload replays the whole turn. **Valid for the lifetime of the Session**: keep it across `endRun` / `requestUpgrade` / continuation-run boundaries; only clear when the Session itself closes. The cursor lets the next subscription open past the prior turn's stale `turn-complete` record. |
113113
| `isStreaming` | `boolean \| undefined` | **Optional.** The transport sets it internally, but you don't have to persist it — the server decides "nothing is streaming" via the session's [`X-Session-Settled`](/ai-chat/client-protocol#x-session-settled-fast-close-on-idle-reconnects) signal on reconnect. If you do persist it, the transport keeps the fast-path short-circuit. If you drop it, reconnects open the SSE and close fast on settled sessions. |
114114

115115
### Session cleanup (frontend)

‎packages/trigger-sdk/src/v3/chat.test.ts‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1194,6 +1194,137 @@ describe("TriggerChatTransport", () => {
11941194
expect(appendAuth).toBe("Bearer fresh-pat");
11951195
expect(transport.getSession("chat-401")?.publicAccessToken).toBe("fresh-pat");
11961196
});
1197+
1198+
function midTurnHarness() {
1199+
const encoder = new TextEncoder();
1200+
const record = (seq: number, chunk: UIMessageChunk | undefined) =>
1201+
chunk
1202+
? {
1203+
body: JSON.stringify({ data: chunk, id: `p-${seq}` }),
1204+
seq_num: seq,
1205+
timestamp: seq,
1206+
headers: [],
1207+
}
1208+
: {
1209+
body: "",
1210+
seq_num: seq,
1211+
timestamp: seq,
1212+
headers: [["trigger-control", "turn-complete"]],
1213+
};
1214+
const batch = (records: unknown[]) =>
1215+
encoder.encode(`event: batch\ndata: ${JSON.stringify({ records })}\n\n`);
1216+
1217+
let outController: ReadableStreamDefaultController<Uint8Array> | undefined;
1218+
const appends: number[] = [];
1219+
global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
1220+
const urlStr = typeof url === "string" ? url : url.toString();
1221+
if (isSessionStreamAppendUrl(urlStr)) {
1222+
appends.push(appends.length + 1);
1223+
return appends.length === harness.failAppend ? authError(401) : defaultAppendResponse();
1224+
}
1225+
if (isSessionOutSubscribeUrl(urlStr)) {
1226+
const body = new ReadableStream<Uint8Array>({
1227+
start(controller) {
1228+
outController = controller;
1229+
controller.enqueue(
1230+
batch([
1231+
record(41, { type: "start", messageId: "a-1" }),
1232+
record(42, { type: "text-start", id: "t-1" }),
1233+
record(43, { type: "text-delta", id: "t-1", delta: "Reading" }),
1234+
])
1235+
);
1236+
},
1237+
});
1238+
return new Response(body, {
1239+
status: 200,
1240+
headers: { "content-type": "text/event-stream", "X-Stream-Version": "v2" },
1241+
});
1242+
}
1243+
throw new Error(`Unexpected URL: ${urlStr}`);
1244+
});
1245+
1246+
const harness = {
1247+
failAppend: 0,
1248+
appends,
1249+
persisted: [] as ChatSessionPersistedState[],
1250+
completeTurn() {
1251+
outController!.enqueue(
1252+
batch([record(44, { type: "text-end", id: "t-1" }), record(45, undefined)])
1253+
);
1254+
outController!.close();
1255+
},
1256+
};
1257+
return harness;
1258+
}
1259+
1260+
it("persists the turn's starting cursor when a refresh lands mid-turn", async () => {
1261+
const harness = midTurnHarness();
1262+
harness.failAppend = 2;
1263+
const transport = new TriggerChatTransport({
1264+
task: "my-chat-task",
1265+
accessToken: vi.fn().mockResolvedValue("fresh-pat"),
1266+
sessions: { "chat-mid": { publicAccessToken: "stale-pat", lastEventId: "40" } },
1267+
onSessionChange: (_chatId, session) => {
1268+
if (session) harness.persisted.push(session);
1269+
},
1270+
});
1271+
1272+
const stream = await transport.sendMessages({
1273+
trigger: "submit-message",
1274+
chatId: "chat-mid",
1275+
messageId: undefined,
1276+
messages: [createUserMessage("research queues")],
1277+
abortSignal: undefined,
1278+
});
1279+
const reader = stream.getReader();
1280+
for (let i = 0; i < 3; i++) await reader.read();
1281+
1282+
await transport.sendPendingMessage("chat-mid", createUserMessage("cover cost too"));
1283+
1284+
expect(harness.appends).toHaveLength(3);
1285+
const refreshed = harness.persisted.at(-1)!;
1286+
expect(refreshed.publicAccessToken).toBe("fresh-pat");
1287+
expect(refreshed.isStreaming).toBe(true);
1288+
expect(refreshed.lastEventId).toBe("40");
1289+
expect(transport.getSession("chat-mid")?.lastEventId).toBe("40");
1290+
1291+
harness.completeTurn();
1292+
while (!(await reader.read()).done) {}
1293+
1294+
const completed = harness.persisted.at(-1)!;
1295+
expect(completed.isStreaming).toBe(false);
1296+
expect(completed.lastEventId).toBe("45");
1297+
});
1298+
1299+
it("keeps the turn's starting cursor after a reconnect when the session was saved without isStreaming", async () => {
1300+
const harness = midTurnHarness();
1301+
harness.failAppend = 1;
1302+
const transport = new TriggerChatTransport({
1303+
task: "my-chat-task",
1304+
accessToken: vi.fn().mockResolvedValue("fresh-pat"),
1305+
sessions: { "chat-reload": { publicAccessToken: "stale-pat", lastEventId: "40" } },
1306+
onSessionChange: (_chatId, session) => {
1307+
if (session) harness.persisted.push(session);
1308+
},
1309+
});
1310+
1311+
const stream = await transport.reconnectToStream({ chatId: "chat-reload" });
1312+
const reader = stream!.getReader();
1313+
for (let i = 0; i < 3; i++) await reader.read();
1314+
1315+
await transport.sendPendingMessage("chat-reload", createUserMessage("cover cost too"));
1316+
1317+
expect(harness.appends).toHaveLength(2);
1318+
const refreshed = harness.persisted.at(-1)!;
1319+
expect(refreshed.publicAccessToken).toBe("fresh-pat");
1320+
expect(refreshed.lastEventId).toBe("40");
1321+
expect(transport.getSession("chat-reload")?.lastEventId).toBe("40");
1322+
1323+
harness.completeTurn();
1324+
while (!(await reader.read()).done) {}
1325+
1326+
expect(transport.getSession("chat-reload")?.lastEventId).toBe("45");
1327+
});
11971328
});
11981329

11991330
describe("stopGeneration", () => {

‎packages/trigger-sdk/src/v3/chat.ts‎

Lines changed: 39 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -767,6 +767,15 @@ type ChatSessionState = {
767767
supersededInputSeq?: number;
768768
/** Whether the agent is currently streaming a response. Set on first chunk, cleared on turn-complete. */
769769
isStreaming?: boolean;
770+
/**
771+
* The resume cursor from before the current turn's first record. Persisted
772+
* in place of `lastEventId` until the turn ends, so a reload replays the
773+
* whole turn: a cursor from the middle of a turn resumes past chunks the
774+
* client needs to rebuild the message. Set when a turn is sent or its first
775+
* record arrives (which covers reconnects and sessions hydrated without
776+
* `isStreaming`), cleared wherever streaming ends.
777+
*/
778+
turnStart?: { lastEventId: string | undefined };
770779
/** Set once the outstanding turn is declared dead: a later stop must not gate the next turn on it. */
771780
outstandingTurnAbandoned?: boolean;
772781
/** Set once the session is closed. Terminal — sends and reconnects stop. */
@@ -859,6 +868,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
859868
lastEventId: session.lastEventId,
860869
activeInputSeq: session.activeInputSeq,
861870
isStreaming: session.isStreaming,
871+
...(session.isStreaming ? { turnStart: { lastEventId: session.lastEventId } } : {}),
862872
closed: session.closed,
863873
closedReason: session.closedReason,
864874
});
@@ -1028,6 +1038,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
10281038

10291039
state.activeInputSeq = inSeq;
10301040
state.isStreaming = true;
1041+
state.turnStart = { lastEventId: state.lastEventId };
10311042
state.outstandingTurnAbandoned = false;
10321043
this.notifySessionChange(chatId, state);
10331044

@@ -1120,6 +1131,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
11201131
const state: ChatSessionState = {
11211132
publicAccessToken: accessToken,
11221133
isStreaming: true,
1134+
turnStart: { lastEventId: undefined },
11231135
};
11241136
this.sessions.set(chatId, state);
11251137
this.notifySessionChange(chatId, state);
@@ -1160,6 +1172,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
11601172
const state = sessions.get(chatId);
11611173
if (state && state.isStreaming) {
11621174
state.isStreaming = false;
1175+
state.turnStart = undefined;
11631176
notifyChange(chatId, state);
11641177
}
11651178
};
@@ -1411,6 +1424,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
14111424
// otherwise a reload resumes mid-turn and replays the chunks the user
14121425
// explicitly stopped.
14131426
state.isStreaming = false;
1427+
state.turnStart = undefined;
14141428
this.notifySessionChange(chatId, state);
14151429
return true;
14161430
};
@@ -1485,6 +1499,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
14851499
// no-ops when the persisted session says isStreaming: false).
14861500
state.activeInputSeq = inSeq;
14871501
state.isStreaming = true;
1502+
state.turnStart = { lastEventId: state.lastEventId };
14881503
state.outstandingTurnAbandoned = false;
14891504
this.notifySessionChange(chatId, state);
14901505

@@ -1507,16 +1522,15 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
15071522
};
15081523

15091524
setSession(chatId: string, session: ChatSessionPersistedState): void {
1510-
this.sessions.set(
1511-
chatId,
1512-
this.applyPendingResumeCursor(chatId, {
1513-
publicAccessToken: session.publicAccessToken,
1514-
lastEventId: session.lastEventId,
1515-
activeInputSeq: session.activeInputSeq,
1516-
isStreaming: session.isStreaming,
1517-
})
1518-
);
1519-
this.notifySessionChange(chatId, this.toPersisted(this.sessions.get(chatId)!));
1525+
const state = this.applyPendingResumeCursor(chatId, {
1526+
publicAccessToken: session.publicAccessToken,
1527+
lastEventId: session.lastEventId,
1528+
activeInputSeq: session.activeInputSeq,
1529+
isStreaming: session.isStreaming,
1530+
});
1531+
if (state.isStreaming) state.turnStart = { lastEventId: state.lastEventId };
1532+
this.sessions.set(chatId, state);
1533+
this.notifySessionChange(chatId, state);
15201534
}
15211535

15221536
/**
@@ -1532,7 +1546,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
15321546
if (existing?.publicAccessToken) {
15331547
if (existing.lastEventId === undefined) {
15341548
existing.lastEventId = lastEventId;
1535-
this.notifySessionChange(chatId, this.toPersisted(existing));
1549+
if (existing.turnStart?.lastEventId === undefined && existing.isStreaming) {
1550+
existing.turnStart = { lastEventId };
1551+
}
1552+
this.notifySessionChange(chatId, existing);
15361553
}
15371554
this.pendingResumeCursors.delete(chatId);
15381555
return;
@@ -1700,7 +1717,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
17001717

17011718
private toPersisted = (state: ChatSessionState): ChatSessionPersistedState => ({
17021719
publicAccessToken: state.publicAccessToken,
1703-
lastEventId: state.lastEventId,
1720+
lastEventId: state.turnStart ? state.turnStart.lastEventId : state.lastEventId,
17041721
activeInputSeq: state.activeInputSeq,
17051722
isStreaming: state.isStreaming,
17061723
closed: state.closed,
@@ -1722,6 +1739,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
17221739
state.closed = true;
17231740
if (reason) state.closedReason = reason;
17241741
state.isStreaming = false;
1742+
state.turnStart = undefined;
17251743
state.activeInputSeq = undefined;
17261744
state.skipToTurnComplete = false;
17271745
state.supersededInputSeq = undefined;
@@ -2015,6 +2033,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
20152033
state.publicAccessToken = publicAccessToken;
20162034
state.lastEventId = undefined;
20172035
state.isStreaming = false;
2036+
state.turnStart = undefined;
20182037
this.sessions.set(chatId, state);
20192038
this.notifySessionChange(chatId, state);
20202039
}
@@ -2219,6 +2238,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
22192238
// leaving it spinning on a stream nobody will finish.
22202239
if (state.isStreaming && this.activeStreams.get(chatId) === internalAbort) {
22212240
state.isStreaming = false;
2241+
state.turnStart = undefined;
22222242
this.notifySessionChange(chatId, state);
22232243
}
22242244
return null;
@@ -2293,7 +2313,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
22932313
return;
22942314
}
22952315

2296-
if (value.id) state.lastEventId = value.id;
2316+
if (value.id) {
2317+
state.turnStart ??= { lastEventId: state.lastEventId };
2318+
state.lastEventId = value.id;
2319+
}
22972320
// Any record — including the first of a resumed connection, which
22982321
// arrives via `primed` — re-earns the resubscribe budget, so a long
22992322
// turn spanning many long-poll windows keeps streaming.
@@ -2371,6 +2394,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
23712394
}
23722395
state.activeInputSeq = undefined;
23732396
state.isStreaming = false;
2397+
state.turnStart = undefined;
23742398
this.notifySessionChange(chatId, state);
23752399
throw new Error("The previous turn's output was lost. Try again.");
23762400
}
@@ -2450,6 +2474,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
24502474
state.activeInputSeq = undefined;
24512475
sinceInSeq = undefined;
24522476
state.isStreaming = false;
2477+
state.turnStart = undefined;
24532478
this.notifySessionChange(chatId, state);
24542479
this.coordinator?.release(chatId);
24552480
this.coordinator?.broadcastSession(chatId, {
@@ -2526,6 +2551,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
25262551
// A superseded stream cannot settle the replacement stream.
25272552
if (this.activeStreams.get(chatId) === internalAbort) {
25282553
state.isStreaming = false;
2554+
state.turnStart = undefined;
25292555
this.notifySessionChange(chatId, state);
25302556
}
25312557
this.emitEvent({

0 commit comments

Comments
 (0)