Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/fluffy-pans-argue.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/sdk": patch
---

Fix chat transport discarding the next turn after stopping generation. `skipToTurnComplete` is now reset when a new message or action is sent, so a message sent after `stopGeneration` streams normally instead of leaving the chat stuck in a streaming state.
8 changes: 8 additions & 0 deletions packages/trigger-sdk/src/v3/chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -870,6 +870,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
this.activeStreams.delete(chatId);
}

// A stop that never saw its TURN_COMPLETE leaves the flag set, and the new
// turn would be skipped record by record.
state.skipToTurnComplete = false;
Comment thread
matt-aitken marked this conversation as resolved.

Comment thread
coderabbitai[bot] marked this conversation as resolved.
state.isStreaming = true;
this.notifySessionChange(chatId, state);

Expand Down Expand Up @@ -1281,6 +1285,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
this.activeStreams.delete(chatId);
}

// A stop that never saw its TURN_COMPLETE leaves the flag set, and the new
// turn would be skipped record by record.
state.skipToTurnComplete = false;

// Mark streaming + persist so a reload mid-action resumes (reconnectToStream
// no-ops when the persisted session says isStreaming: false).
state.isStreaming = true;
Expand Down
64 changes: 64 additions & 0 deletions packages/trigger-sdk/test/chat-transport-events.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,70 @@ describe("transport send events", () => {
});
});

describe("stopped turn followed by a new turn", () => {
/**
* `.out` stub that honours the `Last-Event-ID` cursor like the server does, so
* a resubscribe cannot replay records the reader already consumed. A stop that
* never saw its turn-complete is therefore unrecoverable unless the new send
* clears the skip state.
*/
function cursoredOneTurnTransport() {
const frames = [
{ id: "1", data: `{"type":"text-delta","id":"t1","delta":"hello"}` },
{ id: "2", data: `{"type":"trigger:turn-complete"}` },
];

return makeTransport({
fetch: async (_url, init, ctx) => {
if (ctx.endpoint === "in") return jsonOk();

const cursor = new Headers(init.headers).get("Last-Event-ID");
const from = cursor ? frames.findIndex((f) => f.id === cursor) + 1 : 0;
const remaining = frames.slice(from);
const response = sseResponse(
remaining.map((f) => `id: ${f.id}\ndata: ${f.data}\n\n`).join("")
);
// Nothing left to send: the session is settled, so the reader stops
// instead of resubscribing.
if (remaining.length === 0) response.headers.set("X-Session-Settled", "true");
return response;
},
});
}

it("streams a sendMessages turn after a stop that never saw turn-complete", async () => {
const { transport, events } = cursoredOneTurnTransport();

expect(await transport.stopGeneration("c1")).toBe(true);
events.length = 0;

const stream = await transport.sendMessages({
trigger: "submit-message",
chatId: "c1",
messageId: undefined,
messages: [user("after stop", "u-2")],
abortSignal: undefined,
});
const chunks = await readAll(stream);

expect(chunks).toEqual([{ type: "text-delta", id: "t1", delta: "hello" }]);
expect(events.some((e) => e.type === "turn-completed")).toBe(true);
});

it("streams a sendAction turn after a stop that never saw turn-complete", async () => {
const { transport, events } = cursoredOneTurnTransport();

expect(await transport.stopGeneration("c1")).toBe(true);
events.length = 0;

const stream = await transport.sendAction("c1", { type: "undo" });
const chunks = await readAll(stream);

expect(chunks).toEqual([{ type: "text-delta", id: "t1", delta: "hello" }]);
expect(events.some((e) => e.type === "turn-completed")).toBe(true);
});
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.

describe("transport stream events", () => {
it("marks reconnectToStream subscriptions as resumed", async () => {
const { transport, events } = makeTransport({
Expand Down