From c6c35d59a7608bf0f4d3a982e688080b0a7eb3a8 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 09:47:53 +0100 Subject: [PATCH 1/9] test: cover the head-start accumulator seed without hydrateMessages The seed from payload.headStartMessages had no coverage for agents that do not register hydrateMessages, and it reads unreachable: it sits inside if (!hydrateMessages && couldHavePriorState), and couldHavePriorState is false on a head-start run. It does fire, and this pins that. Records the shape a persisting app has to handle, which is the part that actually bites: by onTurnStart the accumulator is already ['user','assistant'], because the warm route's partial is spliced in before the hook, so the incoming user message is not the last one. --- .../trigger-sdk/test/chatHandover.test.ts | 62 +++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/packages/trigger-sdk/test/chatHandover.test.ts b/packages/trigger-sdk/test/chatHandover.test.ts index a101b91494f..b1aab99f076 100644 --- a/packages/trigger-sdk/test/chatHandover.test.ts +++ b/packages/trigger-sdk/test/chatHandover.test.ts @@ -632,4 +632,66 @@ describe("chat.handover", () => { await harness.close(); } }); + + it("seeds the accumulator from headStartMessages without hydrateMessages", async () => { + // The hydrate variant above gets the head-start user message through + // `incomingMessages`. Without `hydrateMessages` it arrives only via the + // boot-time seed from `payload.headStartMessages`, so this is the path + // that keeps an app with a display-only transcript from storing an + // answer with no question above it. + // + // Note the shape a persisting app has to handle: by `onTurnStart` the + // accumulator is already ["user", "assistant"], because the warm route's + // partial is spliced in before the hook fires. "The incoming message is + // the last one" is therefore false on this path. + let captured: { roles: string[]; texts: string[] } | undefined; + + const agent = chat.agent({ + id: "test-handover-seed-no-hydrate", + onTurnComplete: async ({ uiMessages }) => { + captured = { + roles: uiMessages.map((m) => m.role), + texts: uiMessages.map((m) => + m.parts + .map((p) => (p.type === "text" ? p.text : "")) + .join("") + ), + }; + }, + run: async ({ messages, signal }) => + streamText({ + model: new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("should-not-run") }), + }), + messages, + abortSignal: signal, + }), + }); + + const harness = mockChatAgent(agent, { + chatId: "test-handover-seed-no-hydrate", + mode: "handover-prepare", + headStartMessages: [ + { id: "hs-user-1", role: "user", parts: [{ type: "text", text: "say hi" }] }, + ], + }); + + try { + await harness.sendHandover({ + partialAssistantMessage: [ + { role: "assistant", content: [{ type: "text", text: "Hi there." }] }, + ], + messageId: "asst-seed-1", + isFinal: true, + }); + await new Promise((r) => setTimeout(r, 30)); + + expect(captured).toBeDefined(); + expect(captured!.roles).toEqual(["user", "assistant"]); + expect(captured!.texts[0]).toBe("say hi"); + } finally { + await harness.close(); + } + }); + }); From 5fef13fc1f9bfe9d28eaa13f59438aa3fe78929b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 09:48:17 +0100 Subject: [PATCH 2/9] fix(chat): put injected steering messages into the accumulator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit drainSteeringQueue used the injected uiMessage for span attributes, the injection-confirmation chunk, the injected-ids set and onInjected — never the accumulator. So the message reached the model and the browser, appeared in neither uiMessages nor newUIMessages, and an app persisting from onTurnComplete never learned it existed. The user steers, the agent obeys, the user reloads, and their instruction is gone from the transcript and from every later turn's context. The asymmetry is the tell: a message that finds no step boundary falls back to becoming its own turn and is accumulated normally. Only the path that worked lost data. Appended at injection time rather than turn end, so the order matches what happened: after the message that started the turn, before the response that answers it. Deduplicated by id, since a boundary can drain more than once. The injection path had no test coverage at all — shouldInject appeared only in ai.ts — because the harness had no way to deliver a message mid-turn. Adds harness.sendPendingMessage() for that, which is also what a customer needs to test steering in their own suite. --- .changeset/steering-messages-accumulator.md | 5 + packages/trigger-sdk/src/v3/ai.ts | 34 +++++ .../src/v3/test/mock-chat-agent.ts | 46 +++++- .../test/steering-accumulator.test.ts | 133 ++++++++++++++++++ 4 files changed, 217 insertions(+), 1 deletion(-) create mode 100644 .changeset/steering-messages-accumulator.md create mode 100644 packages/trigger-sdk/test/steering-accumulator.test.ts diff --git a/.changeset/steering-messages-accumulator.md b/.changeset/steering-messages-accumulator.md new file mode 100644 index 00000000000..bc3a4919cf3 --- /dev/null +++ b/.changeset/steering-messages-accumulator.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Steering messages injected mid-answer are now part of the conversation your hooks see. Previously they reached the model and the browser but not `onTurnComplete`, so an app storing its own transcript lost the instruction the answer was shaped by — it vanished from the conversation on reload, and later turns had no record of it. diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index bcc70fa9ce0..ff3f68e2422 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -3521,6 +3521,16 @@ type SteeringQueueEntry = { const chatPendingMessagesKey = locals.create("chat.pendingMessages"); /** @internal */ const chatSteeringQueueKey = locals.create("chat.steeringQueue"); + +/** + * This turn's new messages, as `onTurnComplete.newUIMessages` will see them. + * + * Held in locals because `drainSteeringQueue` runs outside the turn closure and + * has to append the messages it injects. Without that, an injected message + * reaches the model and the browser but no hook, so an app persisting from + * `onTurnComplete` never learns it existed. + */ +const chatTurnNewUIMessagesKey = locals.create("chat.turnNewUIMessages"); /** @internal — IDs of messages that were successfully injected via prepareStep */ const chatInjectedMessageIdsKey = locals.create>("chat.injectedMessageIds"); /** @internal — non-transient data parts queued via chat.response or writer.write() for accumulation into the response message */ @@ -4224,6 +4234,29 @@ async function drainSteeringQueue( for (const m of claimedUIMessages) injectedIds.add(m.id); } + // Record them as part of the conversation. + // + // The model has them and the browser has them; without this the + // accumulator does not, so they reach neither `uiMessages` nor + // `newUIMessages` on `onTurnComplete` and an app that persists from there + // silently loses the instruction the answer was shaped by. Appending here + // rather than at turn end keeps them in the order they happened: after the + // message that started the turn, before the response that answers it. + // + // De-duplicated by id because a step boundary can drain more than once per + // turn, and because a message that failed to inject falls back to becoming + // its own turn, where it is accumulated the normal way. + const currentUIMessages = locals.get(chatCurrentUIMessagesKey); + const turnNew = locals.get(chatTurnNewUIMessagesKey); + for (const m of uiMessages) { + if (currentUIMessages && !currentUIMessages.some((existing) => existing.id === m.id)) { + currentUIMessages.push(m); + } + if (turnNew && !turnNew.some((existing) => existing.id === m.id)) { + turnNew.push(m); + } + } + // Write injection confirmation chunk to the stream so the frontend // knows which messages were injected and where in the response. if (injected.length > 0) { @@ -7490,6 +7523,7 @@ function chatAgent< // Track new messages for this turn (user input + assistant response). const turnNewModelMessages: ModelMessage[] = []; const turnNewUIMessages: TUIMessage[] = []; + locals.set(chatTurnNewUIMessagesKey, turnNewUIMessages); // ── Action handling ────────────────────────────────────── // Actions arrive on the same input stream but with diff --git a/packages/trigger-sdk/src/v3/test/mock-chat-agent.ts b/packages/trigger-sdk/src/v3/test/mock-chat-agent.ts index 63768b9b3f2..e50df390fb5 100644 --- a/packages/trigger-sdk/src/v3/test/mock-chat-agent.ts +++ b/packages/trigger-sdk/src/v3/test/mock-chat-agent.ts @@ -1,5 +1,5 @@ import type { UIMessage, UIMessageChunk } from "ai"; -import { resourceCatalog } from "@trigger.dev/core/v3"; +import { resourceCatalog, sessionStreams } from "@trigger.dev/core/v3"; import type { LocalsKey } from "@trigger.dev/core/v3"; import { runInMockTaskContext, type MockTaskContextOptions } from "@trigger.dev/core/v3/test"; import { __setSessionOpenImplForTests, __setSessionStartImplForTests } from "../sessions.js"; @@ -186,6 +186,17 @@ export type MockChatAgentHarness = { /** Send a custom action and wait for the next turn-complete. */ sendAction(action: unknown): Promise; + /** + * Deliver a message mid-turn without waiting for it, the way the browser's + * steering path does. With a `pendingMessages` config the agent routes it into + * the steering queue for injection at the next step boundary; without one it + * buffers as the next turn. + * + * Send it while a turn is in flight — start the turn without awaiting it, then + * call this. Awaiting the turn first leaves nothing to steer. + */ + sendPendingMessage(message: UIMessage): Promise; + /** Fire a stop signal. Does not wait for the turn — the task keeps running. */ sendStop(message?: string): Promise; @@ -618,6 +629,39 @@ export function mockChatAgent( }); }, + async sendPendingMessage(message) { + await harnessReady; + + const seqBefore = sessionStreams.lastSeqNum(chatId, "in") ?? -1; + + await sendSessionInput(sessionId, { + kind: "message", + payload: { + message, + chatId, + trigger: "submit-message", + metadata: clientData, + }, + }); + + /** + * Wait for the record to be observable on the channel, not merely for the + * send call to return. A test that continues on the send alone is racing the + * append: the message can still be in flight when the step boundary runs, so + * the injection it was meant to trigger silently does not happen and the test + * passes while proving nothing. + */ + const deadline = Date.now() + 5_000; + while ((sessionStreams.lastSeqNum(chatId, "in") ?? -1) <= seqBefore) { + if (Date.now() > deadline) { + throw new Error( + `sendPendingMessage: append for ${message.id} never landed on session.in` + ); + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + }, + async sendStop(message) { await harnessReady; await sendSessionInput(sessionId, { kind: "stop", message }); diff --git a/packages/trigger-sdk/test/steering-accumulator.test.ts b/packages/trigger-sdk/test/steering-accumulator.test.ts new file mode 100644 index 00000000000..aec36a14b39 --- /dev/null +++ b/packages/trigger-sdk/test/steering-accumulator.test.ts @@ -0,0 +1,133 @@ +// Import the test harness FIRST — installs the resource catalog so +// `chat.agent()` below registers its task functions correctly. +import { mockChatAgent } from "../src/v3/test/index.js"; + +import { describe, expect, it } from "vitest"; +import { chat } from "../src/v3/ai.js"; +import { simulateReadableStream, stepCountIs, streamText, tool } from "ai"; +import type { UIMessage } from "ai"; +import { MockLanguageModelV3 } from "ai/test"; +import type { LanguageModelV3StreamPart } from "@ai-sdk/provider"; +import { z } from "zod"; + +const usage = { + inputTokens: { total: 10, noCache: 10, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 10, text: 10, reasoning: undefined }, +}; + +function userMessage(text: string, id: string): UIMessage { + return { id, role: "user", parts: [{ type: "text", text }] }; +} + +function textOf(message: UIMessage): string { + return message.parts.map((part) => (part.type === "text" ? part.text : "")).join(""); +} + +/** + * Two steps with a tool call in between, so there is a step boundary for the + * steering queue to drain at. Step 1 calls the tool, step 2 answers. + */ +function twoStepModel(onFirstStep: () => Promise) { + let call = 0; + return new MockLanguageModelV3({ + doStream: async () => { + call += 1; + if (call === 1) { + const chunks: LanguageModelV3StreamPart[] = [ + { type: "tool-input-start", id: "c1", toolName: "lookup" }, + { type: "tool-input-delta", id: "c1", delta: "{}" }, + { type: "tool-input-end", id: "c1" }, + { type: "tool-call", toolCallId: "c1", toolName: "lookup", input: "{}" }, + { type: "finish", finishReason: { unified: "tool-calls", raw: "tool-calls" }, usage }, + ]; + // Land the steering message while step 1 is streaming, so it is queued + // before the boundary that drains it. + await onFirstStep(); + return { stream: simulateReadableStream({ chunks }) }; + } + return { + stream: simulateReadableStream({ + chunks: [ + { type: "text-start", id: "t1" }, + { type: "text-delta", id: "t1", delta: "done" }, + { type: "text-end", id: "t1" }, + { type: "finish", finishReason: { unified: "stop", raw: "stop" }, usage }, + ], + }), + }; + }, + }); +} + +describe("injected steering messages (TRI-13388)", () => { + it("enter the accumulator, so onTurnComplete can see them", async () => { + let captured: { ui: string[]; newUi: string[] } | undefined; + let injectedCount = 0; + + const send = { fn: async () => {} }; + + const agent = chat.agent({ + id: "steering-accumulator", + tools: { + lookup: tool({ + description: "look something up", + inputSchema: z.object({}), + execute: async () => ({ ok: true }), + }), + }, + pendingMessages: { + shouldInject: ({ steps }) => steps.length > 0, + onInjected: ({ messages }) => { + injectedCount = messages.length; + }, + }, + onTurnComplete: async ({ uiMessages, newUIMessages }) => { + captured = { + ui: uiMessages.map(textOf), + newUi: newUIMessages.map(textOf), + }; + }, + run: async ({ messages, tools, signal }) => + streamText({ + ...chat.toStreamTextOptions({ tools }), + model: twoStepModel(() => send.fn()), + messages, + abortSignal: signal, + stopWhen: stepCountIs(5), + }), + }); + + const harness = mockChatAgent(agent, { chatId: "steering-accumulator" }); + + send.fn = async () => { + await harness.sendPendingMessage(userMessage("actually, only the platform one", "steer-1")); + }; + + try { + await harness.sendMessage(userMessage("summarise every project", "u1")); + await new Promise((resolve) => setTimeout(resolve, 50)); + + // The injection happened — this is the SDK's own bookkeeping. + expect(injectedCount).toBe(1); + + expect(captured).toBeDefined(); + + /** + * The steering message reached the model and the browser. Before this fix it + * reached neither `uiMessages` nor `newUIMessages`, so an app persisting from + * `onTurnComplete` stored an answer shaped by an instruction it never saw, and + * rebuilt the next turn's context without it. + */ + expect(captured!.ui).toContain("actually, only the platform one"); + expect(captured!.newUi).toContain("actually, only the platform one"); + + // And in the order it happened: after the question, before the answer. + expect(captured!.ui.indexOf("actually, only the platform one")).toBeGreaterThan( + captured!.ui.indexOf("summarise every project") + ); + expect(captured!.ui.at(-1)).toBe("done"); + } finally { + await harness.close(); + } + }); +}); From 4b0f52a3114a97157f4b1503ff1e940360087c0f Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 09:49:36 +0100 Subject: [PATCH 3/9] fix(chat): persist history an action rolled back MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The snapshot is written on the turn-complete path, and an action is not a turn — the block literally ends 'if (!isAction)'. So a chat.history mutation from onAction lived only in the running worker's memory. Undo worked while that worker stayed warm, then the next continuation booted from a snapshot still holding the undone exchange and the messages came back. onAction is exactly where the docs tell you to call rollbackTo, so this is the documented path silently not persisting. Writes the snapshot right after the action's override is applied, awaited for the same reason as the turn-complete write: the agent may suspend straight after, and in-flight promises do not reliably survive that. An action has no turn cursor, so the write reuses the last one rather than writing undefined — that would drop the resume point and make the next boot replay from further back to rebuild what it could have read. --- .../persist-action-history-mutations.md | 5 ++ packages/trigger-sdk/src/v3/ai.ts | 73 ++++++++++++++++- .../test/action-snapshot-cursor.test.ts | 80 +++++++++++++++++++ .../trigger-sdk/test/action-snapshot.test.ts | 80 +++++++++++++++++++ 4 files changed, 237 insertions(+), 1 deletion(-) create mode 100644 .changeset/persist-action-history-mutations.md create mode 100644 packages/trigger-sdk/test/action-snapshot-cursor.test.ts create mode 100644 packages/trigger-sdk/test/action-snapshot.test.ts diff --git a/.changeset/persist-action-history-mutations.md b/.changeset/persist-action-history-mutations.md new file mode 100644 index 00000000000..d29338ba6a0 --- /dev/null +++ b/.changeset/persist-action-history-mutations.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Undo, edit and regenerate now survive a run ending. History rolled back from `onAction` was only kept in the running worker's memory, so the rollback held while that worker stayed warm and then reverted on the next continuation — the undone messages came back, minutes later, with no error. diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index ff3f68e2422..f5d932e4f6c 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -6551,6 +6551,16 @@ function chatAgent< // swallow errors internally; the agent stays available either way. const sessionIdForSnapshot = payload.sessionId ?? payload.chatId; let bootSnapshot: ChatSnapshotV1 | undefined; + + /** + * The `lastOutEventId` the most recent snapshot carried. + * + * A snapshot written outside a turn — after an action mutates history — has + * no turn cursor of its own, and writing `undefined` there would drop the + * resume point and make the next boot replay from further back. Retaining it + * keeps an action's write cursor-neutral. + */ + let lastSnapshotOutEventId: string | undefined; let replayedSettled: TUIMessage[] = []; let replayedPartial: TUIMessage | undefined; let replayedPartialRaw: TUIMessage | undefined; @@ -6601,6 +6611,8 @@ function chatAgent< // Without seeding, the new worker would emit no trim on its first // turn (chain self-bootstraps from turn 2), so this is purely an // optimization to keep continuation runs bounded from the first turn. + lastSnapshotOutEventId = bootSnapshot?.lastOutEventId; + if (bootSnapshot?.lastOutEventId !== undefined) { const seeded = Number.parseInt(bootSnapshot.lastOutEventId, 10); if (Number.isFinite(seeded)) { @@ -7607,6 +7619,63 @@ function chatAgent< accumulatedUIMessages = [...actionOverride] as TUIMessage[]; accumulatedMessages = await toModelMessages(actionOverride); locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages); + + /** + * Persist it. An action is not a turn, so it never reaches the + * turn-complete path below where the snapshot is normally + * written — and `onAction` is exactly where `chat.history` + * rollbacks are meant to happen. + * + * Without this the rollback lives only in this worker's + * memory: undo works while the worker stays warm, and the + * next continuation boots from a snapshot that still holds + * the undone exchange, so the undo silently reverts minutes + * later with no error. Awaited for the same reason as the + * turn-complete write — the agent may suspend immediately + * after, and in-flight promises do not reliably survive that. + */ + if (!hydrateMessages) { + try { + await tracer.startActiveSpan( + "snapshot.write", + async () => { + // The resume floor, not the dispatched high-water. An + // action can run while records are still queued, and the + // floor is held back below the earliest of those — writing + // the high-water instead would advance the cursor past + // records a replay still has to recover, losing them. + const snapshotInCursor = chatInputRouter().resumeFloor(); + await writeChatSnapshot(sessionIdForSnapshot, { + version: 1, + savedAt: Date.now(), + messages: accumulatedUIMessages, + lastOutEventId: lastSnapshotOutEventId, + lastInEventId: + snapshotInCursor !== undefined + ? String(snapshotInCursor) + : undefined, + }); + }, + { + attributes: { + [SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart", + [SemanticInternalAttributes.COLLAPSED]: true, + "chat.id": currentWirePayload.chatId, + "chat.snapshot.reason": "action", + "chat.messages.count": accumulatedUIMessages.length, + }, + } + ); + } catch (error) { + logger.warn( + "chat.agent: snapshot write after action failed; the mutation may not survive a continuation", + { + error: error instanceof Error ? error.message : String(error), + sessionId: sessionIdForSnapshot, + } + ); + } + } } } else { warnMissingOnActionOnce(); @@ -8683,11 +8752,13 @@ function chatAgent< "snapshot.write", async () => { const snapshotInCursor = chatInputRouter().resumeFloor(); + lastSnapshotOutEventId = + turnCompleteResult?.lastEventId ?? lastSnapshotOutEventId; await writeChatSnapshot(sessionIdForSnapshot, { version: 1, savedAt: Date.now(), messages: accumulatedUIMessages, - lastOutEventId: turnCompleteResult?.lastEventId, + lastOutEventId: lastSnapshotOutEventId, lastInEventId: snapshotInCursor !== undefined ? String(snapshotInCursor) : undefined, }); diff --git a/packages/trigger-sdk/test/action-snapshot-cursor.test.ts b/packages/trigger-sdk/test/action-snapshot-cursor.test.ts new file mode 100644 index 00000000000..8f07db20f68 --- /dev/null +++ b/packages/trigger-sdk/test/action-snapshot-cursor.test.ts @@ -0,0 +1,80 @@ +// Import the test harness FIRST — installs the resource catalog so +// `chat.agent()` below registers its task functions correctly. +import { mockChatAgent } from "../src/v3/test/index.js"; + +import { describe, expect, it } from "vitest"; +import { chat } from "../src/v3/ai.js"; +import { simulateReadableStream, streamText } from "ai"; +import { MockLanguageModelV3 } from "ai/test"; +import type { LanguageModelV3StreamPart } from "@ai-sdk/provider"; +import { z } from "zod"; + +function textStream(text: string): ReadableStream { + return simulateReadableStream({ + chunks: [ + { type: "text-start", id: "t1" }, + { type: "text-delta", id: "t1", delta: text }, + { type: "text-end", id: "t1" }, + { + type: "finish", + finishReason: { unified: "stop", raw: "stop" }, + usage: { + inputTokens: { total: 10, noCache: 10, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 10, text: 10, reasoning: undefined }, + }, + }, + ], + }); +} + +describe("the snapshot an action writes", () => { + it("keeps the resume cursor the last turn established", async () => { + const agent = chat.agent({ + id: "action-snapshot-cursor", + actionSchema: z.discriminatedUnion("type", [z.object({ type: z.literal("undo") })]), + onAction: async ({ action }) => { + if (action.type === "undo") chat.history.slice(0, -2); + }, + run: async ({ messages, signal }) => + streamText({ + model: new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("answer") }), + }), + messages, + abortSignal: signal, + }), + }); + + const harness = mockChatAgent(agent, { chatId: "action-snapshot-cursor" }); + + try { + await harness.sendMessage({ + id: "u1", + role: "user", + parts: [{ type: "text", text: "first" }], + }); + await new Promise((r) => setTimeout(r, 30)); + + const afterTurn = harness.getSnapshot(); + expect(afterTurn?.lastOutEventId).toBeDefined(); + + await harness.sendAction({ type: "undo" }); + await new Promise((r) => setTimeout(r, 30)); + + const afterAction = harness.getSnapshot(); + + /** + * An action has no turn cursor of its own. Writing the snapshot with + * `lastOutEventId: undefined` would drop the resume point the last turn + * established, and the next boot would replay from further back to rebuild + * what it could have read — so an action's write has to be cursor-neutral. + */ + expect(afterAction?.lastOutEventId).toBe(afterTurn?.lastOutEventId); + + // And the mutation itself landed, which is the point of writing at all. + expect(afterAction?.messages ?? []).toEqual([]); + } finally { + await harness.close(); + } + }); +}); diff --git a/packages/trigger-sdk/test/action-snapshot.test.ts b/packages/trigger-sdk/test/action-snapshot.test.ts new file mode 100644 index 00000000000..2445875e816 --- /dev/null +++ b/packages/trigger-sdk/test/action-snapshot.test.ts @@ -0,0 +1,80 @@ +// Import the test harness FIRST — installs the resource catalog so +// `chat.agent()` calls below register their task functions correctly. +import { mockChatAgent } from "../src/v3/test/index.js"; + +import { describe, expect, it } from "vitest"; +import { chat } from "../src/v3/ai.js"; +import { simulateReadableStream, streamText } from "ai"; +import { MockLanguageModelV3 } from "ai/test"; +import type { LanguageModelV3StreamPart } from "@ai-sdk/provider"; +import { z } from "zod"; + +function textStream(text: string): ReadableStream { + return simulateReadableStream({ + chunks: [ + { type: "text-start", id: "t1" }, + { type: "text-delta", id: "t1", delta: text }, + { type: "text-end", id: "t1" }, + { + type: "finish", + finishReason: { unified: "stop", raw: "stop" }, + usage: { + inputTokens: { total: 10, noCache: 10, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 10, text: 10, reasoning: undefined }, + }, + }, + ], + }); +} + +function agentWithUndo(id: string) { + return chat.agent({ + id, + actionSchema: z.discriminatedUnion("type", [z.object({ type: z.literal("undo") })]), + onAction: async ({ action }) => { + if (action.type === "undo") { + // The documented way to roll history back — see /ai-chat/actions. + chat.history.slice(0, -2); + } + }, + run: async ({ messages, signal }) => + streamText({ + model: new MockLanguageModelV3({ doStream: async () => ({ stream: textStream("answer") }) }), + messages, + abortSignal: signal, + }), + }); +} + +describe("snapshot durability of history mutated by an action", () => { + it("persists an undo, so a continuation does not resurrect the undone turn", async () => { + const harness = mockChatAgent(agentWithUndo("action-snapshot-undo"), { + chatId: "action-snapshot-undo", + }); + + try { + await harness.sendMessage({ + id: "u1", + role: "user", + parts: [{ type: "text", text: "first" }], + }); + await new Promise((r) => setTimeout(r, 30)); + + // After a turn the snapshot holds the exchange. + expect(harness.getSnapshot()?.messages.map((m) => m.role)).toEqual(["user", "assistant"]); + + await harness.sendAction({ type: "undo" }); + await new Promise((r) => setTimeout(r, 30)); + + /** + * An action is not a turn, so it never reaches the turn-complete path where + * the snapshot is written. The rollback lives in the accumulator only, and + * the next continuation boots from a snapshot that still holds the undone + * exchange — the user's undo silently reverts, minutes later, with no error. + */ + expect(harness.getSnapshot()?.messages ?? []).toEqual([]); + } finally { + await harness.close(); + } + }); +}); From e399b60e123237043e1e72c1c98ca08c2026973b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 15:25:50 +0100 Subject: [PATCH 4/9] fix(chat): make a response streamed from onAction part of the conversation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Returning a StreamTextResult from onAction piped it to the browser and stopped there. The accumulator never saw it, no snapshot recorded it, and actions fire no onTurnComplete — so the user read a good answer that the model had no memory of, and the next turn carried on from the answer regenerate had just replaced. The disagreement between the screen and the conversation was invisible until that next turn contradicted it. The action branch now captures what it pipes, using the pipeChatAndCapture that already existed for exactly this, and appends the message to the accumulator. Persistence beyond the snapshot is still the app's job, since an action fires no turn hook — pipeAndCapture hands back the same message for that. Also folds the snapshot write added for rolled-back history into one helper used by both action paths, so a regenerate that both rolls back and answers writes once rather than twice, and the cursor-preservation rule lives in one place. The two fixes needed each other: with the rollback persisted but the response dropped, a regenerate left the snapshot empty rather than stale — still wrong, just differently. --- .changeset/action-stream-into-conversation.md | 5 + packages/trigger-sdk/src/v3/ai.ts | 160 ++++++++++-------- .../test/action-stream-accumulator.test.ts | 98 +++++++++++ 3 files changed, 196 insertions(+), 67 deletions(-) create mode 100644 .changeset/action-stream-into-conversation.md create mode 100644 packages/trigger-sdk/test/action-stream-accumulator.test.ts diff --git a/.changeset/action-stream-into-conversation.md b/.changeset/action-stream-into-conversation.md new file mode 100644 index 00000000000..a3e6b50ce1c --- /dev/null +++ b/.changeset/action-stream-into-conversation.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +A response streamed back from `onAction` is now part of the conversation. Returning a `StreamTextResult` from an action sent it to the browser and nowhere else, so a regenerate showed the user a new answer that the model had no memory of — the next turn carried on from the answer that had just been replaced. diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index f5d932e4f6c..e0ed4a36161 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -6561,6 +6561,56 @@ function chatAgent< * keeps an action's write cursor-neutral. */ let lastSnapshotOutEventId: string | undefined; + + /** + * Persist the accumulator outside a turn. + * + * An action is not a turn, so it never reaches the turn-complete path where + * the snapshot is normally written — but it can change the conversation in + * two ways: a `chat.history` mutation, and a response streamed back from + * `onAction`. Both have to survive, and one write at the end of the action + * covers both rather than writing twice for a regenerate that does both. + * + * Cursor-neutral: an action has no turn cursor of its own, and writing + * `undefined` would drop the resume point the last turn established and make + * the next boot replay from further back. + */ + const writeSnapshotOutsideTurn = async (reason: string) => { + if (hydrateMessages) return; + try { + await tracer.startActiveSpan( + "snapshot.write", + async () => { + const snapshotInCursor = chatInputRouter().resumeFloor(); + await writeChatSnapshot(sessionIdForSnapshot, { + version: 1, + savedAt: Date.now(), + messages: accumulatedUIMessages, + lastOutEventId: lastSnapshotOutEventId, + lastInEventId: + snapshotInCursor !== undefined ? String(snapshotInCursor) : undefined, + }); + }, + { + attributes: { + [SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart", + [SemanticInternalAttributes.COLLAPSED]: true, + "chat.snapshot.reason": reason, + "chat.messages.count": accumulatedUIMessages.length, + }, + } + ); + } catch (error) { + logger.warn( + "chat.agent: snapshot write outside a turn failed; the change may not survive a continuation", + { + error: error instanceof Error ? error.message : String(error), + sessionId: sessionIdForSnapshot, + reason, + } + ); + } + }; let replayedSettled: TUIMessage[] = []; let replayedPartial: TUIMessage | undefined; let replayedPartialRaw: TUIMessage | undefined; @@ -7548,6 +7598,13 @@ function chatAgent< // string, or UIMessage from `onAction`. Turn counter // does not advance. let actionStreamResult: unknown = undefined; + /** + * Whether this action changed the conversation, by rolling history + * back or by streaming a response. Drives the single snapshot write + * at the end — an action never reaches the turn-complete path that + * normally does it. + */ + let actionChangedHistory = false; if (isAction) { // Parse and validate the action payload const parsedAction = parseAction @@ -7620,62 +7677,7 @@ function chatAgent< accumulatedMessages = await toModelMessages(actionOverride); locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages); - /** - * Persist it. An action is not a turn, so it never reaches the - * turn-complete path below where the snapshot is normally - * written — and `onAction` is exactly where `chat.history` - * rollbacks are meant to happen. - * - * Without this the rollback lives only in this worker's - * memory: undo works while the worker stays warm, and the - * next continuation boots from a snapshot that still holds - * the undone exchange, so the undo silently reverts minutes - * later with no error. Awaited for the same reason as the - * turn-complete write — the agent may suspend immediately - * after, and in-flight promises do not reliably survive that. - */ - if (!hydrateMessages) { - try { - await tracer.startActiveSpan( - "snapshot.write", - async () => { - // The resume floor, not the dispatched high-water. An - // action can run while records are still queued, and the - // floor is held back below the earliest of those — writing - // the high-water instead would advance the cursor past - // records a replay still has to recover, losing them. - const snapshotInCursor = chatInputRouter().resumeFloor(); - await writeChatSnapshot(sessionIdForSnapshot, { - version: 1, - savedAt: Date.now(), - messages: accumulatedUIMessages, - lastOutEventId: lastSnapshotOutEventId, - lastInEventId: - snapshotInCursor !== undefined - ? String(snapshotInCursor) - : undefined, - }); - }, - { - attributes: { - [SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart", - [SemanticInternalAttributes.COLLAPSED]: true, - "chat.id": currentWirePayload.chatId, - "chat.snapshot.reason": "action", - "chat.messages.count": accumulatedUIMessages.length, - }, - } - ); - } catch (error) { - logger.warn( - "chat.agent: snapshot write after action failed; the mutation may not survive a continuation", - { - error: error instanceof Error ? error.message : String(error), - sessionId: sessionIdForSnapshot, - } - ); - } - } + actionChangedHistory = true; } } else { warnMissingOnActionOnce(); @@ -7959,17 +7961,37 @@ function chatAgent< isUIMessageStreamable(actionStreamResult) ) { try { - const resolvedOptions = resolveUIMessageStreamOptions(); - const uiStream = ( - actionStreamResult as UIMessageStreamable - ).toUIMessageStream({ - ...resolvedOptions, - generateMessageId: resolvedOptions.generateMessageId ?? generateMessageId, - }); - await pipeChat(uiStream, { - signal: combinedSignal, - spanName: "stream response", - }); + /** + * Captured, not just piped. The stream reaching the browser was + * never the problem — the problem was that it stopped there, so + * the user read an answer the accumulator had no record of and + * the next turn contradicted the screen. Worst on regenerate, + * which removes the old answer and used to leave nothing in its + * place. + * + * Persistence beyond the snapshot is still the app's job: an + * action fires no `onTurnComplete`, so an app owning its own + * store has to write the row itself — `chat.pipeAndCapture` + * hands back the same message for that. + */ + const { message: actionResponse } = await pipeChatAndCapture( + actionStreamResult as UIMessageStreamable, + { signal: combinedSignal, spanName: "stream response" } + ); + + if (actionResponse) { + const existingIdx = actionResponse.id + ? accumulatedUIMessages.findIndex((m) => m.id === actionResponse.id) + : -1; + if (existingIdx !== -1) { + accumulatedUIMessages[existingIdx] = actionResponse as TUIMessage; + } else { + accumulatedUIMessages.push(actionResponse as TUIMessage); + } + accumulatedMessages = await toModelMessages(accumulatedUIMessages); + locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages); + actionChangedHistory = true; + } } catch (error) { if ( error instanceof Error && @@ -7982,6 +8004,10 @@ function chatAgent< } } + if (actionChangedHistory) { + await writeSnapshotOutsideTurn("action"); + } + await writeTurnCompleteChunk(currentWirePayload.chatId); // Don't consume a turn iteration — actions aren't turns. diff --git a/packages/trigger-sdk/test/action-stream-accumulator.test.ts b/packages/trigger-sdk/test/action-stream-accumulator.test.ts new file mode 100644 index 00000000000..b10d830d292 --- /dev/null +++ b/packages/trigger-sdk/test/action-stream-accumulator.test.ts @@ -0,0 +1,98 @@ +// Import the test harness FIRST — installs the resource catalog so +// `chat.agent()` below registers its task functions correctly. +import { mockChatAgent } from "../src/v3/test/index.js"; + +import { describe, expect, it } from "vitest"; +import { chat } from "../src/v3/ai.js"; +import { simulateReadableStream, streamText } from "ai"; +import type { UIMessage } from "ai"; +import { MockLanguageModelV3 } from "ai/test"; +import type { LanguageModelV3StreamPart } from "@ai-sdk/provider"; +import { z } from "zod"; + +function textStream(text: string): ReadableStream { + return simulateReadableStream({ + chunks: [ + { type: "text-start", id: "t1" }, + { type: "text-delta", id: "t1", delta: text }, + { type: "text-end", id: "t1" }, + { + type: "finish", + finishReason: { unified: "stop", raw: "stop" }, + usage: { + inputTokens: { total: 10, noCache: 10, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 10, text: 10, reasoning: undefined }, + }, + }, + ], + }); +} + +function textOf(message: UIMessage): string { + return message.parts.map((part) => (part.type === "text" ? part.text : "")).join(""); +} + +describe("a StreamTextResult returned from onAction (TRI-13378)", () => { + it("becomes part of the conversation, not just something the browser saw", async () => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("regenerated answer") }), + }); + + const agent = chat.agent({ + id: "action-stream-accumulator", + actionSchema: z.discriminatedUnion("type", [z.object({ type: z.literal("regenerate") })]), + + /** + * The bare shape the docs show: return the stream and let the runtime pipe + * it. The alternative — consuming it with `chat.pipeAndCapture` — is the + * workaround, so testing that instead would prove nothing about this path. + */ + onAction: async ({ action, messages }) => { + if (action.type !== "regenerate") return; + chat.history.slice(0, -1); + return streamText({ model, messages }); + }, + + run: async ({ messages, signal }) => + streamText({ + model: new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("first answer") }), + }), + messages, + abortSignal: signal, + }), + }); + + const harness = mockChatAgent(agent, { chatId: "action-stream-accumulator" }); + + try { + await harness.sendMessage({ + id: "u1", + role: "user", + parts: [{ type: "text", text: "ask" }], + }); + await new Promise((r) => setTimeout(r, 30)); + + const turn = await harness.sendAction({ type: "regenerate" }); + await new Promise((r) => setTimeout(r, 50)); + + // The browser did see it — that part was never broken. + const streamed = turn.chunks + .filter((c) => c.type === "text-delta") + .map((c) => (c as { delta: string }).delta) + .join(""); + expect(streamed).toBe("regenerated answer"); + + /** + * And the conversation agrees with the screen. Before the fix the response + * was piped and dropped: absent from the accumulator, absent from the + * snapshot, so the next turn's model context contained the question and the + * *old* answer that regenerate had just removed. + */ + const snapshot = harness.getSnapshot(); + expect(snapshot?.messages.map(textOf)).toEqual(["ask", "regenerated answer"]); + } finally { + await harness.close(); + } + }); +}); From 52bd98d342b303fcc0ea0e56eff00371c10eab51 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Aug 2026 16:42:12 +0100 Subject: [PATCH 5/9] fix(chat): route a system-role injection to the instructions lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chat.inject with role 'system' put the message into the conversation, which ai@7 rejects for every provider: standardizePrompt throws before any provider is called. The next turn died with an error chunk reading 'An error occurred.' and persisted an assistant message with no parts, so from the app's side the agent had simply stopped answering. The error message names the fix — use the instructions option — and Instructions is string | SystemModelMessage | Array, so an injected system block has a correct home. It is appended after the base prompt, which keeps the prompt's position for caching and reads as a later amendment. This makes the documented examples right rather than rewriting them to a workaround. It also answers whether trusted mid-conversation context is supportable: it is, and only this way. A message injected as 'user' is untrusted by construction, and a well-aligned model says so and re-derives the answer from tools instead. The docs now state which lane to use for facts and which for directives. A new instruction block changes the cached prefix, so the first call carrying it misses the prompt cache. Only turns that actually injected pay it. --- .changeset/inject-instructions-shape.md | 5 + .changeset/inject-system-to-instructions.md | 5 + docs/ai-chat/background-injection.mdx | 43 ++++- packages/trigger-sdk/src/v3/ai.ts | 91 +++++++++- .../test/inject-system-instructions.test.ts | 171 ++++++++++++++++++ 5 files changed, 311 insertions(+), 4 deletions(-) create mode 100644 .changeset/inject-instructions-shape.md create mode 100644 .changeset/inject-system-to-instructions.md create mode 100644 packages/trigger-sdk/test/inject-system-instructions.test.ts diff --git a/.changeset/inject-instructions-shape.md b/.changeset/inject-instructions-shape.md new file mode 100644 index 00000000000..85129f9cb57 --- /dev/null +++ b/.changeset/inject-instructions-shape.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Injected system context is merged into a single instruction block, so it works on every supported AI SDK version. Note that a cached system prompt gives up its cache entry for as long as an injection is live, since the cached prefix has changed. diff --git a/.changeset/inject-system-to-instructions.md b/.changeset/inject-system-to-instructions.md new file mode 100644 index 00000000000..c28af52c8ce --- /dev/null +++ b/.changeset/inject-system-to-instructions.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +`chat.inject()` with `role: "system"` now works. It previously put the system message into the conversation, which AI SDK 7 rejects for every provider — the next turn died with a generic "An error occurred." and persisted an empty assistant message, so the agent looked like it had simply stopped answering. System-role context is now appended to the model's instructions, which is also the only way to inject context the agent will treat as trusted. diff --git a/docs/ai-chat/background-injection.mdx b/docs/ai-chat/background-injection.mdx index f84336ff4de..86e7c227f0f 100644 --- a/docs/ai-chat/background-injection.mdx +++ b/docs/ai-chat/background-injection.mdx @@ -189,9 +189,50 @@ export const myChat = chat.agent({ | **Source** | Backend task code | Frontend user input | | **Triggered by** | Your code (e.g. `onTurnComplete` + `chat.defer()`) | User sending a message during streaming | | **Injection point** | Start of next turn, or next `prepareStep` boundary | Next `prepareStep` boundary only | -| **Message role** | Any (`system`, `user`, `assistant`) | Typically `user` | +| **Message role** | Any — `system` becomes an instruction, others join the conversation (see below) | Typically `user` | | **Frontend visibility** | Not visible unless you write custom `data-*` chunks | Visible via `usePendingMessages` hook | +## Two lanes: trusted and untrusted + +The role you inject with decides more than position — it decides whether the model +treats the content as trustworthy. + +**`role: "system"` goes to the instructions lane.** The block is appended to the +system instructions for subsequent inference calls, so it carries the same standing +as your system prompt. This is the lane for context the agent should simply believe: +entitlements, plan changes, operational notices. + +It has to work this way. On AI SDK 7 a system message inside `messages` is rejected +for every provider — `standardizePrompt` throws before any provider is called, and +its own advice is to use the instructions option. `Instructions` accepts +`Array`, so the injected block is appended there rather than +smuggled into the transcript. + +Two things worth knowing: + +- A new instruction block changes the cached prefix, so the first call carrying it + misses the prompt cache. Only the turns where something was actually injected pay + that. +- The injected text is merged into a single instruction rather than added as a + second block, because AI SDK 5 rejects an array of system blocks while accepting + one structured block. That means a cached system prompt loses its cache entry for + as long as an injection is live — the prefix changed, so there is nothing to hit. + If you rely on prompt caching, inject sparingly and prefer facts that go stale, so + the injection clears. + +**Any other role joins the conversation, and is untrusted by construction.** A +message injected as `user` is indistinguishable from something the user typed, and a +well-aligned model treats it accordingly — it may say so and re-derive the answer +from tools instead of taking it at face value: + +> "that text arrived embedded in your message, not from a tool I called, so I +> verified it myself rather than trusting it" + +That is correct behaviour, not a bug. So inject **checkable facts** in the +conversational lane and put **directives** in the instructions lane. A conclusion +injected as a user message is the worst of both: the model neither trusts it nor +ignores it, and may contradict it in front of the user. + ## API reference ### chat.inject() diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index e0ed4a36161..2f063a6c9f8 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -46,6 +46,7 @@ import type { FinishReason, LanguageModelUsage, ModelMessage, + SystemModelMessage, ProviderMetadata, Tool, ToolSet, @@ -2695,6 +2696,24 @@ function spliceHandoverPartial( */ const chatBackgroundQueueKey = locals.create("chat.backgroundQueue"); +/** + * System-role context injected mid-conversation, held for the instructions lane. + * + * Kept apart from the message queue because ai@7 rejects a system message inside + * `messages` for every provider — `standardizePrompt` throws upstream of any + * provider call, and its own advice is to use the instructions option. Instructions + * accept `Array`, so a system-role injection has a correct + * home: appended as another system block rather than smuggled into the transcript. + * + * This is also the only way to inject *trusted* context. A message injected as + * `user` is untrusted by construction, and a well-aligned model treats it that + * way — it will say so, and re-derive the answer from tools instead. + */ +const chatInjectedInstructionsKey = locals.create( + "chat.injectedInstructions" +); + + /** * Run-scoped pipe counter. Stored in locals so concurrent runs in the * same worker don't share state. @@ -4691,6 +4710,59 @@ function toStreamTextOptions(options?: ToStreamTextOptionsOptions): Record`. This package's peer + * range still spans all three, so emitting an array unconditionally would break + * v5 consumers — for whom a system-role injection used to work, since v5 accepted + * a system message inside `messages` that v7 rejects. + * + * So: concatenate into one string when the base is a plain string, which every + * version accepts and which loses nothing (separate blocks only matter for + * per-block `providerOptions`). Use the array form only when the base is already + * a structured message — that path requires v6+ regardless, because it is how + * prompt caching marks the system block, and flattening it would silently throw + * the cache away. + * + * Either way the injected text goes last: the base prompt keeps its position for + * caching, and the addition reads as a later amendment. A changed prefix does + * cost the first call its cache hit, on turns that actually injected. + */ + const injectedInstructions = locals.get(chatInjectedInstructionsKey); + if (injectedInstructions && injectedInstructions.length > 0) { + const injectedText = injectedInstructions + .map((block) => (typeof block.content === "string" ? block.content : "")) + .filter(Boolean) + .join("\n\n"); + + const base = result.system; + + if (base === undefined) { + result.system = injectedText; + } else if (typeof base === "string") { + result.system = [base, injectedText].filter(Boolean).join("\n\n"); + } else { + // Merged into the existing block rather than added as a second one. An array + // of system blocks would keep the base block's cache entry, but ai@5 rejects + // it outright ("Invalid prompt: system must be a string") while accepting a + // single structured block, and this package's peer range still spans v5. + // Choosing per version would mean resolving the installed version at runtime, + // which is not something to build on: `import.meta.url` is illegal in this + // package's CommonJS output, and a bundled task may have no resolvable `ai` + // to read. One shape that works everywhere beats a cache hit. + const baseBlock = base as SystemModelMessage; + result.system = { + ...baseBlock, + content: [typeof baseBlock.content === "string" ? baseBlock.content : "", injectedText] + .filter(Boolean) + .join("\n\n"), + }; + } + } + // Prompt-related options (only if chat.prompt.set() was called) if (prompt) { // Resolve model via registry if both are present @@ -9944,9 +10016,22 @@ function chatDefer(promiseOrFn: Promise | (() => Promise)): vo * ``` */ function injectBackgroundContext(messages: ModelMessage[]): void { - const queue = locals.get(chatBackgroundQueueKey) ?? []; - queue.push(...messages); - locals.set(chatBackgroundQueueKey, queue); + const systemBlocks = messages.filter( + (message): message is SystemModelMessage => message.role === "system" + ); + const conversational = messages.filter((message) => message.role !== "system"); + + if (systemBlocks.length > 0) { + const instructions = locals.get(chatInjectedInstructionsKey) ?? []; + instructions.push(...systemBlocks); + locals.set(chatInjectedInstructionsKey, instructions); + } + + if (conversational.length > 0) { + const queue = locals.get(chatBackgroundQueueKey) ?? []; + queue.push(...conversational); + locals.set(chatBackgroundQueueKey, queue); + } } // --------------------------------------------------------------------------- diff --git a/packages/trigger-sdk/test/inject-system-instructions.test.ts b/packages/trigger-sdk/test/inject-system-instructions.test.ts new file mode 100644 index 00000000000..618055c8055 --- /dev/null +++ b/packages/trigger-sdk/test/inject-system-instructions.test.ts @@ -0,0 +1,171 @@ +// Import the test harness FIRST — installs the resource catalog so +// `chat.agent()` below registers its task functions correctly. +import { mockChatAgent } from "../src/v3/test/index.js"; + +import { describe, expect, it } from "vitest"; +import { chat } from "../src/v3/ai.js"; +import { simulateReadableStream, streamText } from "ai"; +import { MockLanguageModelV3 } from "ai/test"; +import type { LanguageModelV3StreamPart } from "@ai-sdk/provider"; + +function textStream(text: string): ReadableStream { + return simulateReadableStream({ + chunks: [ + { type: "text-start", id: "t1" }, + { type: "text-delta", id: "t1", delta: text }, + { type: "text-end", id: "t1" }, + { + type: "finish", + finishReason: { unified: "stop", raw: "stop" }, + usage: { + inputTokens: { total: 10, noCache: 10, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 10, text: 10, reasoning: undefined }, + }, + }, + ], + }); +} + +describe("chat.inject with a system role (TRI-13380)", () => { + it("goes to the instructions lane instead of poisoning the prompt", async () => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }); + + let injected = false; + + const agent = chat.agent({ + id: "inject-system-instructions", + onBoot: async () => { + chat.prompt.set({ + promptId: "base", + version: 1, + labels: ["local"], + text: "You are a helpful assistant.", + model: undefined, + config: undefined, + toAISDKTelemetry: () => ({ experimental_telemetry: { isEnabled: true, metadata: {} } }), + }); + }, + onTurnComplete: async () => { + if (injected) return; + injected = true; + /** + * The shape every docs example uses. On ai@7 a system message inside + * `messages` is rejected by `standardizePrompt` for every provider, so + * this used to kill the next turn — an error chunk reading "An error + * occurred." and an assistant message with no parts. + */ + chat.inject([{ role: "system", content: "The user just upgraded to Pro." }]); + }, + run: async ({ messages, signal }) => + streamText({ + ...chat.toStreamTextOptions(), + model, + messages, + abortSignal: signal, + }), + }); + + const harness = mockChatAgent(agent, { chatId: "inject-system-instructions" }); + + try { + await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "one" }] }); + await new Promise((r) => setTimeout(r, 40)); + + const turn = await harness.sendMessage({ + id: "u2", + role: "user", + parts: [{ type: "text", text: "two" }], + }); + await new Promise((r) => setTimeout(r, 40)); + + // The turn survives. + const errors = turn.rawChunks.filter((c) => (c as { type?: string })?.type === "error"); + expect(errors).toEqual([]); + + // The injected context arrives as a system block, alongside the base prompt, + // and never as a system message inside `messages`. + const prompt = model.doStreamCalls.at(-1)!.prompt; + const systemBlocks = prompt.filter((m) => m.role === "system"); + const asText = JSON.stringify(systemBlocks); + + expect(asText).toContain("You are a helpful assistant."); + expect(asText).toContain("The user just upgraded to Pro."); + + const nonSystem = prompt.filter((m) => m.role !== "system"); + expect(JSON.stringify(nonSystem)).not.toContain("upgraded to Pro"); + } finally { + await harness.close(); + } + }); + it("emits one system value whether or not the base block is cached", async () => { + /** + * Never an array. ai@6+ accepts `Array` and would let a + * cached base block keep its cache entry, but ai@5 rejects an array outright + * ("Invalid prompt: system must be a string") while accepting a single + * structured block — and the peer range still spans v5. So a plain base + * concatenates into a string, and a cached base absorbs the injection into its + * own content, keeping its provider options. + */ + const shapes: unknown[] = []; + + function agentFor(id: string, cacheControl: boolean) { + let injected = false; + return chat.agent({ + id, + onBoot: async () => { + chat.prompt.set({ + promptId: "base", + version: 1, + labels: ["local"], + text: "Base instructions.", + model: undefined, + config: undefined, + toAISDKTelemetry: () => ({ + experimental_telemetry: { isEnabled: true, metadata: {} }, + }), + }); + }, + onTurnComplete: async () => { + if (injected) return; + injected = true; + chat.inject([{ role: "system", content: "Amendment." }]); + }, + run: async ({ messages, signal }) => { + const options = cacheControl + ? chat.toStreamTextOptions({ cacheControl: { type: "ephemeral" } }) + : chat.toStreamTextOptions(); + shapes.push(Array.isArray(options.system) ? "array" : typeof options.system); + return streamText({ + ...options, + model: new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }), + messages, + abortSignal: signal, + }); + }, + }); + } + + for (const [id, cacheControl] of [ + ["shape-plain", false], + ["shape-cached", true], + ] as const) { + const harness = mockChatAgent(agentFor(id, cacheControl), { chatId: id }); + try { + await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "one" }] }); + await new Promise((r) => setTimeout(r, 40)); + await harness.sendMessage({ id: "u2", role: "user", parts: [{ type: "text", text: "two" }] }); + await new Promise((r) => setTimeout(r, 40)); + } finally { + await harness.close(); + } + } + + // [plain turn 1, plain turn 2 (injected), cached turn 1, cached turn 2 (injected)] + expect(shapes).toEqual(["string", "string", "object", "object"]); + }); + +}); From 47f2f8f2322aa9267d14596b5003f8ec58c4bb69 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 28 Aug 2026 15:06:51 +0100 Subject: [PATCH 6/9] fix(chat): address review on the accumulator, instructions and action paths Record only the messages a steering drain actually claimed. The loop used the offered batch, so a record another consumer took while shouldInject() awaited was written into the accumulator for a turn it was never part of. Drain the injected instructions once applied, matching the conversational lane. Left in place they were re-applied by every later toStreamTextOptions() call in the run, growing the prompt and changing its cached prefix each turn. Clean a stopped action's partial response before it is committed, and skip committing at all once the run is cancelled. --- packages/trigger-sdk/src/v3/ai.ts | 18 ++++- .../trigger-sdk/test/action-snapshot.test.ts | 4 +- .../test/action-stream-accumulator.test.ts | 2 +- .../trigger-sdk/test/chatHandover.test.ts | 5 +- .../test/inject-system-instructions.test.ts | 70 ++++++++++++++++++- .../test/steering-accumulator.test.ts | 2 +- 6 files changed, 88 insertions(+), 13 deletions(-) diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 2f063a6c9f8..5b65d61e2b7 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -2713,7 +2713,6 @@ const chatInjectedInstructionsKey = locals.create( "chat.injectedInstructions" ); - /** * Run-scoped pipe counter. Stored in locals so concurrent runs in the * same worker don't share state. @@ -4267,7 +4266,7 @@ async function drainSteeringQueue( // its own turn, where it is accumulated the normal way. const currentUIMessages = locals.get(chatCurrentUIMessagesKey); const turnNew = locals.get(chatTurnNewUIMessagesKey); - for (const m of uiMessages) { + for (const m of claimedUIMessages) { if (currentUIMessages && !currentUIMessages.some((existing) => existing.id === m.id)) { currentUIMessages.push(m); } @@ -4734,6 +4733,7 @@ function toStreamTextOptions(options?: ToStreamTextOptionsOptions): Record 0) { const injectedText = injectedInstructions + .splice(0) .map((block) => (typeof block.content === "string" ? block.content : "")) .filter(Boolean) .join("\n\n"); @@ -8046,11 +8046,23 @@ function chatAgent< * store has to write the row itself — `chat.pipeAndCapture` * hands back the same message for that. */ - const { message: actionResponse } = await pipeChatAndCapture( + const captured = await pipeChatAndCapture( actionStreamResult as UIMessageStreamable, { signal: combinedSignal, spanName: "stream response" } ); + if (runSignal.aborted) return "exit"; + + /** + * A stopped action still commits what streamed, cleaned: + * incomplete tool and text parts left mid-flight are what + * strand the UI on a spinner forever once persisted. + */ + const actionResponse = + captured.status === "complete" || !captured.message + ? captured.message + : cleanupAbortedParts(captured.message); + if (actionResponse) { const existingIdx = actionResponse.id ? accumulatedUIMessages.findIndex((m) => m.id === actionResponse.id) diff --git a/packages/trigger-sdk/test/action-snapshot.test.ts b/packages/trigger-sdk/test/action-snapshot.test.ts index 2445875e816..39e2ecfe280 100644 --- a/packages/trigger-sdk/test/action-snapshot.test.ts +++ b/packages/trigger-sdk/test/action-snapshot.test.ts @@ -39,7 +39,9 @@ function agentWithUndo(id: string) { }, run: async ({ messages, signal }) => streamText({ - model: new MockLanguageModelV3({ doStream: async () => ({ stream: textStream("answer") }) }), + model: new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("answer") }), + }), messages, abortSignal: signal, }), diff --git a/packages/trigger-sdk/test/action-stream-accumulator.test.ts b/packages/trigger-sdk/test/action-stream-accumulator.test.ts index b10d830d292..b57b7de8022 100644 --- a/packages/trigger-sdk/test/action-stream-accumulator.test.ts +++ b/packages/trigger-sdk/test/action-stream-accumulator.test.ts @@ -32,7 +32,7 @@ function textOf(message: UIMessage): string { return message.parts.map((part) => (part.type === "text" ? part.text : "")).join(""); } -describe("a StreamTextResult returned from onAction (TRI-13378)", () => { +describe("a StreamTextResult returned from onAction", () => { it("becomes part of the conversation, not just something the browser saw", async () => { const model = new MockLanguageModelV3({ doStream: async () => ({ stream: textStream("regenerated answer") }), diff --git a/packages/trigger-sdk/test/chatHandover.test.ts b/packages/trigger-sdk/test/chatHandover.test.ts index b1aab99f076..65dd802d26f 100644 --- a/packages/trigger-sdk/test/chatHandover.test.ts +++ b/packages/trigger-sdk/test/chatHandover.test.ts @@ -652,9 +652,7 @@ describe("chat.handover", () => { captured = { roles: uiMessages.map((m) => m.role), texts: uiMessages.map((m) => - m.parts - .map((p) => (p.type === "text" ? p.text : "")) - .join("") + m.parts.map((p) => (p.type === "text" ? p.text : "")).join("") ), }; }, @@ -693,5 +691,4 @@ describe("chat.handover", () => { await harness.close(); } }); - }); diff --git a/packages/trigger-sdk/test/inject-system-instructions.test.ts b/packages/trigger-sdk/test/inject-system-instructions.test.ts index 618055c8055..296b13fb168 100644 --- a/packages/trigger-sdk/test/inject-system-instructions.test.ts +++ b/packages/trigger-sdk/test/inject-system-instructions.test.ts @@ -26,7 +26,7 @@ function textStream(text: string): ReadableStream { }); } -describe("chat.inject with a system role (TRI-13380)", () => { +describe("chat.inject with a system role", () => { it("goes to the instructions lane instead of poisoning the prompt", async () => { const model = new MockLanguageModelV3({ doStream: async () => ({ stream: textStream("ok") }), @@ -155,9 +155,17 @@ describe("chat.inject with a system role (TRI-13380)", () => { ] as const) { const harness = mockChatAgent(agentFor(id, cacheControl), { chatId: id }); try { - await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "one" }] }); + await harness.sendMessage({ + id: "u1", + role: "user", + parts: [{ type: "text", text: "one" }], + }); await new Promise((r) => setTimeout(r, 40)); - await harness.sendMessage({ id: "u2", role: "user", parts: [{ type: "text", text: "two" }] }); + await harness.sendMessage({ + id: "u2", + role: "user", + parts: [{ type: "text", text: "two" }], + }); await new Promise((r) => setTimeout(r, 40)); } finally { await harness.close(); @@ -168,4 +176,60 @@ describe("chat.inject with a system role (TRI-13380)", () => { expect(shapes).toEqual(["string", "string", "object", "object"]); }); + it("applies an injection to the next turn only, not to every later turn", async () => { + /** + * `chat.inject()` is a queue consumed at the next injection opportunity, so + * the instructions lane has to drain like the conversational one does. Left + * undrained, every later turn in the run repeats every earlier injection — + * the prompt grows without bound and its cached prefix changes each turn. + */ + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }); + + let injectedOnce = false; + + const agent = chat.agent({ + id: "inject-system-drains", + onTurnComplete: async () => { + if (injectedOnce) return; + injectedOnce = true; + chat.inject([{ role: "system", content: "SENTINEL-ONE-SHOT" }]); + }, + run: async ({ messages, signal }) => + streamText({ + ...chat.toStreamTextOptions(), + model, + messages, + abortSignal: signal, + }), + }); + + const harness = mockChatAgent(agent, { chatId: "inject-system-drains" }); + + try { + await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "one" }] }); + await new Promise((r) => setTimeout(r, 40)); + + await harness.sendMessage({ id: "u2", role: "user", parts: [{ type: "text", text: "two" }] }); + await new Promise((r) => setTimeout(r, 40)); + + await harness.sendMessage({ + id: "u3", + role: "user", + parts: [{ type: "text", text: "three" }], + }); + await new Promise((r) => setTimeout(r, 40)); + + const systemOf = (i: number) => + JSON.stringify(model.doStreamCalls[i]!.prompt.filter((m) => m.role === "system")); + + // Turn 1 injected nothing yet, turn 2 carries it, turn 3 must not repeat it. + expect(systemOf(0)).not.toContain("SENTINEL-ONE-SHOT"); + expect(systemOf(1)).toContain("SENTINEL-ONE-SHOT"); + expect(systemOf(2)).not.toContain("SENTINEL-ONE-SHOT"); + } finally { + await harness.close(); + } + }); }); diff --git a/packages/trigger-sdk/test/steering-accumulator.test.ts b/packages/trigger-sdk/test/steering-accumulator.test.ts index aec36a14b39..41f0e734068 100644 --- a/packages/trigger-sdk/test/steering-accumulator.test.ts +++ b/packages/trigger-sdk/test/steering-accumulator.test.ts @@ -59,7 +59,7 @@ function twoStepModel(onFirstStep: () => Promise) { }); } -describe("injected steering messages (TRI-13388)", () => { +describe("injected steering messages", () => { it("enter the accumulator, so onTurnComplete can see them", async () => { let captured: { ui: string[]; newUi: string[] } | undefined; let injectedCount = 0; From 52772b5ea5f0014bda7ea2caabff14faf5d434b4 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 28 Aug 2026 15:45:20 +0100 Subject: [PATCH 7/9] fix(chat): report a failed action stream instead of committing it as finished MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pipeChatAndCapture returns a stream failure rather than throwing it, so a mid-stream failure in a response returned from onAction was committed as a complete answer, snapshotted, and followed by a normal turn-complete with no error — the browser saw the stream stop and the next turn built on the truncated text. The partial is still kept; the failure is now surfaced with it. Document that the instructions lane is delivered by chat.toStreamTextOptions(), and that an injection applies to the next inference call only. --- docs/ai-chat/background-injection.mdx | 12 +++ packages/trigger-sdk/src/v3/ai.ts | 9 +++ .../test/action-stream-accumulator.test.ts | 74 +++++++++++++++++++ 3 files changed, 95 insertions(+) diff --git a/docs/ai-chat/background-injection.mdx b/docs/ai-chat/background-injection.mdx index 86e7c227f0f..e2760b80dd1 100644 --- a/docs/ai-chat/background-injection.mdx +++ b/docs/ai-chat/background-injection.mdx @@ -208,8 +208,20 @@ its own advice is to use the instructions option. `Instructions` accepts `Array`, so the injected block is appended there rather than smuggled into the transcript. + + The instructions lane is delivered by `chat.toStreamTextOptions()`, because that + is the only place the SDK can set `streamText`'s instructions for you. If your + `run()` calls `streamText({ model, messages, abortSignal })` without spreading + `chat.toStreamTextOptions()`, a `role: "system"` injection never reaches the + model. The conversational lane has no such requirement — it arrives through + `messages` either way. + + Two things worth knowing: +- An injection applies to the next inference call only. The lane is drained once + applied, so a block injected in `onTurnComplete` shapes the following turn and is + not repeated on every turn after it. - A new instruction block changes the cached prefix, so the first call carrying it misses the prompt cache. Only the turns where something was actually injected pay that. diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 5b65d61e2b7..20d360e844c 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -8076,6 +8076,15 @@ function chatAgent< locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages); actionChangedHistory = true; } + + /** + * Reported after the partial is committed, not instead of it. + * `pipeChatAndCapture` returns a stream failure rather than + * throwing, so without this a mid-stream failure writes a + * normal turn-complete and the truncated answer is persisted + * as if it were finished — the next turn then builds on it. + */ + if (captured.status === "error") throw captured.error; } catch (error) { if ( error instanceof Error && diff --git a/packages/trigger-sdk/test/action-stream-accumulator.test.ts b/packages/trigger-sdk/test/action-stream-accumulator.test.ts index b57b7de8022..f3a6234883f 100644 --- a/packages/trigger-sdk/test/action-stream-accumulator.test.ts +++ b/packages/trigger-sdk/test/action-stream-accumulator.test.ts @@ -95,4 +95,78 @@ describe("a StreamTextResult returned from onAction", () => { await harness.close(); } }); + + it("reports a mid-stream failure instead of committing a truncated answer as finished", async () => { + /** + * `pipeChatAndCapture` returns a stream failure as `status: "error"` rather + * than throwing it. Unchecked, the action commits whatever streamed, writes a + * normal turn-complete, and the browser just sees the stream stop — so the + * user reads a half-finished answer presented as complete and the next turn + * builds on it. The partial is still kept, as on the turn path; what changes + * is that the failure is surfaced alongside it. + */ + let stage = 0; + const failsMidStream = new MockLanguageModelV3({ + doStream: async () => ({ + stream: new ReadableStream({ + async pull(controller) { + await new Promise((r) => setTimeout(r, 25)); + if (stage === 0) { + controller.enqueue({ type: "text-start", id: "t1" }); + stage++; + return; + } + if (stage === 1) { + controller.enqueue({ type: "text-delta", id: "t1", delta: "half an answer" }); + stage++; + return; + } + controller.error(new Error("provider exploded mid-stream")); + }, + }), + }), + }); + + const agent = chat.agent({ + id: "action-stream-error", + actionSchema: z.discriminatedUnion("type", [z.object({ type: z.literal("regenerate") })]), + onAction: async ({ action, messages }) => { + if (action.type !== "regenerate") return; + chat.history.slice(0, -1); + return streamText({ model: failsMidStream, messages }); + }, + run: async ({ messages, signal }) => + streamText({ + model: new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("first answer") }), + }), + messages, + abortSignal: signal, + }), + }); + + const harness = mockChatAgent(agent, { chatId: "action-stream-error" }); + + try { + await harness.sendMessage({ + id: "u1", + role: "user", + parts: [{ type: "text", text: "ask" }], + }); + await new Promise((r) => setTimeout(r, 40)); + + await harness.sendAction({ type: "regenerate" }).catch(() => {}); + await new Promise((r) => setTimeout(r, 300)); + + const errors = (harness.allRawChunks as { type?: string }[]).filter( + (c) => c.type === "error" + ); + expect(errors.length).toBeGreaterThan(0); + + // The partial is still kept rather than discarded. + expect(harness.getSnapshot()?.messages.map(textOf).at(-1)).toContain("half an answer"); + } finally { + await harness.close(); + } + }); }); From c6b0dbd0f4e2231c6bcdbb56d9cbb050934686d8 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 29 Aug 2026 09:00:24 +0100 Subject: [PATCH 8/9] docs(chat): note the instructions delivery path and one-shot injection in the changesets --- .changeset/action-stream-into-conversation.md | 2 ++ .changeset/inject-system-to-instructions.md | 2 ++ 2 files changed, 4 insertions(+) diff --git a/.changeset/action-stream-into-conversation.md b/.changeset/action-stream-into-conversation.md index a3e6b50ce1c..621a2cb83fb 100644 --- a/.changeset/action-stream-into-conversation.md +++ b/.changeset/action-stream-into-conversation.md @@ -3,3 +3,5 @@ --- A response streamed back from `onAction` is now part of the conversation. Returning a `StreamTextResult` from an action sent it to the browser and nowhere else, so a regenerate showed the user a new answer that the model had no memory of — the next turn carried on from the answer that had just been replaced. + +A stream that fails part-way through is also no longer committed as though it finished. Whatever streamed is still kept, but the failure is reported instead of the truncated text being stored, and built on, as a complete answer. diff --git a/.changeset/inject-system-to-instructions.md b/.changeset/inject-system-to-instructions.md index c28af52c8ce..fe4e4ff1ce9 100644 --- a/.changeset/inject-system-to-instructions.md +++ b/.changeset/inject-system-to-instructions.md @@ -3,3 +3,5 @@ --- `chat.inject()` with `role: "system"` now works. It previously put the system message into the conversation, which AI SDK 7 rejects for every provider — the next turn died with a generic "An error occurred." and persisted an empty assistant message, so the agent looked like it had simply stopped answering. System-role context is now appended to the model's instructions, which is also the only way to inject context the agent will treat as trusted. + +Two things to know. Instructions are delivered by `chat.toStreamTextOptions()`, so a `run()` that calls `streamText` without spreading it will not receive a system-role injection — the conversational lane has no such requirement. And an injection applies to the next inference call only, rather than repeating on every turn that follows it. From e3318cf4fa286e4b0abe89908e67bc711c23526b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 29 Aug 2026 09:20:47 +0100 Subject: [PATCH 9/9] docs(ai-chat): say what an action persists under each persistence model The actions page said only that persistence was your responsibility inside onAction, which is now wrong for platform-managed agents (the runtime writes the snapshot) and too vague for app-owned ones, where a rollback and a streamed replacement both need storing and there is no onTurnComplete to do it in. --- docs/ai-chat/actions.mdx | 29 ++++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/docs/ai-chat/actions.mdx b/docs/ai-chat/actions.mdx index 956e5090aef..27ea4310414 100644 --- a/docs/ai-chat/actions.mdx +++ b/docs/ai-chat/actions.mdx @@ -70,7 +70,34 @@ onAction: async ({ action, messages }) => { } ``` -This is useful for actions that both mutate state and want a fresh model response (regenerate-from-here, retry-with-different-style). Persistence is your responsibility inside `onAction` itself; you have access to the streamed response object. +This is useful for actions that both mutate state and want a fresh model response (regenerate-from-here, retry-with-different-style). + +### Actions and persistence + +An action is not a turn, so `onTurnComplete` never fires — and that is where an app that owns its own transcript normally writes. What that means depends on which persistence model you use. + +**Platform-managed** (no `hydrateMessages`): nothing to do. After an action that changed the conversation — a `chat.history` mutation, a response returned from `onAction`, or both — the runtime writes the snapshot, so the change survives the run ending. + +**Your own store** (`hydrateMessages` registered): the runtime deliberately does not write, because your store is the source of truth. A history mutation and a returned response both live only in the running worker until you persist them, and a continuation rehydrates from your store, not from what the worker had in memory. `chat.pipeAndCapture` hands you the same assistant message the runtime would have captured: + +```ts +onAction: async ({ action, messages }) => { + if (action.type === "undo") { + chat.history.slice(0, -2); + await db.deleteLastExchange(chatId); // the rollback is yours to persist + } + + if (action.type === "regenerate") { + chat.history.slice(0, -1); + const { message } = await chat.pipeAndCapture( + streamText({ model: anthropic("claude-sonnet-4-5"), messages }) + ); + if (message) await db.saveMessage(message); // and so is the replacement + } +}, +``` + +Returning the stream instead of piping it yourself still works and still reaches the browser — you just have no message to store, so the next run will not know about it. ## Gating actions on HITL state