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/.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/.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/.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/docs/ai-chat/background-injection.mdx b/docs/ai-chat/background-injection.mdx index f84336ff4de..e2760b80dd1 100644 --- a/docs/ai-chat/background-injection.mdx +++ b/docs/ai-chat/background-injection.mdx @@ -189,9 +189,62 @@ 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. + + + 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. +- 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 bcc70fa9ce0..20d360e844c 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,23 @@ 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. @@ -3521,6 +3539,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 +4252,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 claimedUIMessages) { + 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) { @@ -4658,6 +4709,60 @@ 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 + .splice(0) + .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 @@ -6518,6 +6623,66 @@ 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; + + /** + * 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; @@ -6568,6 +6733,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)) { @@ -7490,6 +7657,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 @@ -7502,6 +7670,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 @@ -7573,6 +7748,8 @@ function chatAgent< accumulatedUIMessages = [...actionOverride] as TUIMessage[]; accumulatedMessages = await toModelMessages(actionOverride); locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages); + + actionChangedHistory = true; } } else { warnMissingOnActionOnce(); @@ -7856,17 +8033,58 @@ 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 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) + : -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; + } + + /** + * 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 && @@ -7879,6 +8097,10 @@ function chatAgent< } } + if (actionChangedHistory) { + await writeSnapshotOutsideTurn("action"); + } + await writeTurnCompleteChunk(currentWirePayload.chatId); // Don't consume a turn iteration — actions aren't turns. @@ -8649,11 +8871,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, }); @@ -9813,9 +10037,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/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/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..39e2ecfe280 --- /dev/null +++ b/packages/trigger-sdk/test/action-snapshot.test.ts @@ -0,0 +1,82 @@ +// 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(); + } + }); +}); 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..f3a6234883f --- /dev/null +++ b/packages/trigger-sdk/test/action-stream-accumulator.test.ts @@ -0,0 +1,172 @@ +// 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", () => { + 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(); + } + }); + + 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(); + } + }); +}); diff --git a/packages/trigger-sdk/test/chatHandover.test.ts b/packages/trigger-sdk/test/chatHandover.test.ts index a101b91494f..65dd802d26f 100644 --- a/packages/trigger-sdk/test/chatHandover.test.ts +++ b/packages/trigger-sdk/test/chatHandover.test.ts @@ -632,4 +632,63 @@ 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(); + } + }); }); 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..296b13fb168 --- /dev/null +++ b/packages/trigger-sdk/test/inject-system-instructions.test.ts @@ -0,0 +1,235 @@ +// 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", () => { + 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"]); + }); + + 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 new file mode 100644 index 00000000000..41f0e734068 --- /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", () => { + 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(); + } + }); +});