Skip to content

Commit c56e3e5

Browse files
committed
fix(sdk): let a watch subscription re-establish after a reload when the last turn completed
1 parent 1a2e187 commit c56e3e5

3 files changed

Lines changed: 35 additions & 2 deletions

File tree

.changeset/watch-mode-keepalive.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22
"@trigger.dev/sdk": patch
33
---
44

5-
Watch-mode chat streams now survive quiet windows and keep delivering later turns, and a reply cut off by a lost connection now shows an error instead of appearing finished. Aborting a resumed subscription no longer stops the run — it only closes your local stream, so a viewer leaving can't cut off someone else's reply. To stop generation, call `stopGeneration(chatId)`, or pass `stopOnAbort: true` to `reconnectToStream` when that subscriber owns the turn.
5+
Watch-mode chat streams now survive quiet windows and keep delivering later turns, and a reply cut off by a lost connection now shows an error instead of appearing finished. Aborting a resumed subscription no longer stops the run — it only closes your local stream, so a viewer leaving can't cut off someone else's reply. To stop generation, call `stopGeneration(chatId)`, or pass `stopOnAbort: true` to `reconnectToStream` when that subscriber owns the turn. Watch subscriptions also re-establish after a page reload even when the last turn has completed.

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

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1056,6 +1056,37 @@ describe("TriggerChatTransport", () => {
10561056
expect(result).toBeNull();
10571057
});
10581058

1059+
it("resumes in watch mode when the session is hydrated with isStreaming=false", async () => {
1060+
let subscribeCount = 0;
1061+
global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
1062+
const urlStr = typeof url === "string" ? url : url.toString();
1063+
if (isSessionOutSubscribeUrl(urlStr)) {
1064+
subscribeCount++;
1065+
const response = defaultSseResponse([{ type: "text-delta", id: "p1", delta: "turn2" }]);
1066+
const headers = new Headers(response.headers);
1067+
headers.set("X-Session-Settled", "true");
1068+
return new Response(response.body, { status: 200, headers });
1069+
}
1070+
throw new Error(`Unexpected URL: ${urlStr}`);
1071+
});
1072+
1073+
const transport = new TriggerChatTransport({
1074+
task: "my-chat-task",
1075+
accessToken: () => "pat",
1076+
watch: true,
1077+
sessions: {
1078+
"chat-rc-watch": { publicAccessToken: "p", isStreaming: false },
1079+
},
1080+
});
1081+
1082+
const stream = await transport.reconnectToStream({ chatId: "chat-rc-watch" });
1083+
expect(stream).not.toBeNull();
1084+
const chunks = await drainChunks(stream!);
1085+
1086+
expect(subscribeCount).toBe(1);
1087+
expect(chunks).toEqual([{ type: "text-delta", id: "p1", delta: "turn2" }]);
1088+
});
1089+
10591090
it("opens an SSE subscription with the X-Peek-Settled header set", async () => {
10601091
let subscribeHeaders: Headers | undefined;
10611092
global.fetch = vi.fn().mockImplementation(async (url: string | URL, init?: RequestInit) => {

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1162,7 +1162,9 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
11621162
const state = this.sessions.get(options.chatId);
11631163
if (!state) return null;
11641164

1165-
if (state.isStreaming === false) return null;
1165+
// Watch is a standing subscription: a settled session is exactly the
1166+
// state it waits in, so a completed last turn must not block the resume.
1167+
if (state.isStreaming === false && !this.watchMode) return null;
11661168
if (this.activeStreams.has(options.chatId)) return null;
11671169

11681170
const abortController = new AbortController();

0 commit comments

Comments
 (0)