diff --git a/.changeset/managed-streamtext-in-run.md b/.changeset/managed-streamtext-in-run.md new file mode 100644 index 00000000000..e530f974b31 --- /dev/null +++ b/.changeset/managed-streamtext-in-run.md @@ -0,0 +1,20 @@ +--- +"@trigger.dev/sdk": minor +--- + +`run()` now receives a `streamText` with your agent's managed options already applied, so they cannot be lost by leaving out the spread: + +```ts +run: async ({ messages, signal, streamText }) => + streamText({ model, messages, abortSignal: signal }); +``` + +Spreading `chat.toStreamTextOptions()` still works and is equivalent. The difference is what happens when your options collide with the managed ones. Passing `tools` after the spread replaces the skill tools, and passing your own `prepareStep` replaces the managed one, which silently switches off steering, compaction and injected context. The managed `streamText` merges tools and composes `prepareStep` instead, so neither can be turned off by accident. + +`system` can be set at the call site, on `chat.agent({ system })`, or through `chat.prompt.set()`, but only in one of them: setting it in two places throws, because no single shape merges two system values across every supported AI SDK version, and dropping one silently is the failure this seam exists to prevent. Injected instructions append to whichever one is in play. + +`chat.agent()` also takes `registry`, `cacheControl` and `systemProviderOptions` now, so a managed prompt's model and its cache breakpoint no longer have to be passed at the call site. + +`onAction` receives the same `streamText`, so a response produced from an action, a regenerate especially, answers with the agent's own system prompt and tools. Built with the `streamText` imported from `ai` it answered with none, and the reply still looked fine, which is what made the difference easy to miss. + +`chat.headStart` and `chat.startHeadStart` hand their `run` the same thing, carrying the four options the handover protocol depends on. There it matters more: re-setting `messages`, `stopWhen` or `abortSignal` after a spread breaks the handover rather than degrading a feature, and nothing caught it. On the managed one those keys are a type error. diff --git a/docs/ai-chat/actions.mdx b/docs/ai-chat/actions.mdx index a3b5787b71e..c62cf9b2f83 100644 --- a/docs/ai-chat/actions.mdx +++ b/docs/ai-chat/actions.mdx @@ -44,7 +44,7 @@ export const myChat = chat.agent({ // returning void → side-effect-only, no model call }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -56,8 +56,10 @@ export const myChat = chat.agent({ `onAction` can return a `StreamTextResult`, `string`, or `UIMessage` to produce a response. The returned stream is auto-piped to the frontend just like a normal turn, but the rest of the turn machinery (`onTurnStart`, `onTurnComplete`, etc.) still does not fire. +Build it with the `streamText` from `onAction`'s own argument, the same one `run()` receives. It carries the agent's system prompt, skill tools, resolved model and telemetry, so a regenerated answer is produced under the same configuration as every other turn. The `streamText` imported from `ai` carries none of that, and the reply still looks fine, which is what makes the difference easy to miss. + ```ts -onAction: async ({ action, messages }) => { +onAction: async ({ action, messages, streamText }) => { if (action.type === "regenerate") { chat.history.slice(0, -1); // drop the last assistant return streamText({ @@ -81,7 +83,7 @@ An action is not a turn, so `onTurnComplete` never fires, and that is where an a **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 }) => { +onAction: async ({ action, messages, streamText }) => { if (action.type === "undo") { chat.history.slice(0, -2); await db.deleteLastExchange(chatId); // the rollback is yours to persist @@ -107,7 +109,7 @@ Returning the stream instead of piping it yourself still works and still reaches If you have a [human-in-the-loop](/ai-chat/patterns/human-in-the-loop) tool waiting on `addToolOutput`, you usually want to refuse competing actions like `regenerate` until the answer arrives. [`chat.history.getPendingToolCalls()`](/ai-chat/backend#chat-history) gives you exactly that signal: ```ts -onAction: async ({ action, messages, signal }) => { +onAction: async ({ action, messages, signal, streamText }) => { if (action.type === "regenerate") { if (chat.history.getPendingToolCalls().length > 0) return; // gated chat.history.slice(0, -1); diff --git a/docs/ai-chat/anatomy.mdx b/docs/ai-chat/anatomy.mdx index 3f7cf876cb8..235da7f218d 100644 --- a/docs/ai-chat/anatomy.mdx +++ b/docs/ai-chat/anatomy.mdx @@ -18,7 +18,7 @@ Everything below maps onto one annotated agent: ```ts trigger/my-agent.ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText, stepCountIs } from "ai"; +import { stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; export const myAgent = chat.agent({ @@ -36,9 +36,9 @@ export const myAgent = chat.agent({ // The turn loop. Messages arrive accumulated; you stream back. // Options, levels, and alternatives — see Backend. - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools }), + tools, model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, diff --git a/docs/ai-chat/backend.mdx b/docs/ai-chat/backend.mdx index c055570ef16..b59553c8c0c 100644 --- a/docs/ai-chat/backend.mdx +++ b/docs/ai-chat/backend.mdx @@ -30,14 +30,13 @@ Return the `streamText` result from `run` and it's automatically piped to the fr ```ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText, stepCountIs } from "ai"; +import { stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; export const simpleChat = chat.agent({ id: "simple-chat", - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ - ...chat.toStreamTextOptions(), // prepareStep, system, telemetry (see note below) model: anthropic("claude-sonnet-4-5"), system: "You are a helpful assistant.", messages, @@ -48,9 +47,43 @@ export const simpleChat = chat.agent({ }); ``` - - **Always spread `chat.toStreamTextOptions()` first** (as above) so your explicit overrides win. It wires up the `prepareStep` callback behind [compaction](/ai-chat/compaction), [steering](/ai-chat/pending-messages), and [background injection](/ai-chat/background-injection), all of which silently no-op without it, and injects the system prompt from `chat.prompt()`, the resolved model (when you pass a `registry`), and telemetry metadata. Examples below keep the spread implicit for brevity, so include it in real code. - + + The `streamText` destructured from `run`'s argument is the SDK's, not the one + imported from `ai`. It carries the agent's managed options, so nothing has to be + spread in. [The managed streamText](#the-managed-streamtext) covers what those + options are and what happens when yours collide with them. + + +### The managed streamText + +`run()` is handed a `streamText` that already carries everything the spread provides, so the managed state cannot be lost by leaving the spread out: + +```ts +export const simpleChat = chat.agent({ + id: "simple-chat", + run: async ({ messages, signal, streamText }) => + streamText({ + model: anthropic("claude-sonnet-4-5"), + messages, + abortSignal: signal, + stopWhen: stepCountIs(15), + }), +}); +``` + +Note the destructured `streamText`: it shadows the one imported from `ai` inside `run`, so the managed options apply without a spread. Spreading `chat.toStreamTextOptions()` into the imported `streamText` is still supported and equivalent. + +It differs from the spread in three ways, all of them about what happens when your options collide with the managed ones: + +| Option | Spread | Managed `streamText` | +| --- | --- | --- | +| `tools` | Passing `tools` after the spread replaces the skill tools | Merged, so skill tools survive | +| `prepareStep` | Passing your own after the spread replaces the managed one, silently disabling steering, compaction and injection | Composed, yours runs after the managed one | +| `system` | Yours replaces the managed prompt and any injected instructions | Throws | + +`system` throws rather than merging because there is no shape that combines two system values on every supported AI SDK version: v5 rejects an array of blocks, and a structured block carries the provider options that make [prompt caching](/ai-chat/prompt-caching) work, so concatenating discards the cache entry. Set a static prompt with [`chat.prompt.set()`](#using-prompts) and add per-turn context with [`chat.inject()`](/ai-chat/background-injection). + +If the managed prompt names a model, pass a registry on the agent so the runtime can resolve it: `chat.agent({ registry, run })`. ### Using chat.pipe() for complex flows @@ -58,13 +91,12 @@ For complex agent flows where `streamText` is called deep inside your code, use ```ts trigger/agent-chat.ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; import type { ModelMessage } from "ai"; export const agentChat = chat.agent({ id: "agent-chat", - run: async ({ messages }) => { + run: async ({ messages, streamText }) => { // Don't return anything — chat.pipe is called inside await runAgentLoop(messages); }, @@ -102,7 +134,7 @@ export const myChat = chat.agent({ // responseMessage.parts includes the data-metadata part await db.messages.save(responseMessage); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { // Also works from run() via chat.response chat.response.write({ type: "data-context", @@ -177,9 +209,9 @@ const tools = { searchDocs }; export const myChat = chat.agent({ id: "my-chat", tools, - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools }), + tools, model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, @@ -200,12 +232,12 @@ See [Tools](/ai-chat/tools) for `toModelOutput` across turns, per-turn dynamic t ### Using prompts -Use [AI Prompts](/ai/prompts) to manage your system prompt as versioned, overridable config. Store the resolved prompt in a lifecycle hook with `chat.prompt.set()`, then spread `chat.toStreamTextOptions()` into `streamText` — it includes the system prompt, model, config, and telemetry automatically. +Use [AI Prompts](/ai/prompts) to manage your system prompt as versioned, overridable config. Store the resolved prompt in a lifecycle hook with `chat.prompt.set()`. The `streamText` from `run`'s argument picks it up: system prompt, model, config and telemetry. ```ts import { chat } from "@trigger.dev/sdk/ai"; import { prompts } from "@trigger.dev/sdk"; -import { streamText, createProviderRegistry } from "ai"; +import { createProviderRegistry } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; import { z } from "zod"; @@ -221,15 +253,15 @@ const systemPrompt = prompts.define({ export const myChat = chat.agent({ id: "my-chat", + registry, clientDataSchema: z.object({ userId: z.string() }), onChatStart: async ({ clientData }) => { const user = await db.user.findUnique({ where: { id: clientData.userId } }); const resolved = await systemPrompt.resolve({ name: user.name }); chat.prompt.set(resolved); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ - ...chat.toStreamTextOptions({ registry }), // system, model, config, telemetry messages, abortSignal: signal, stopWhen: stepCountIs(15), @@ -238,16 +270,9 @@ export const myChat = chat.agent({ }); ``` -`chat.toStreamTextOptions()` returns an object with `system`, `model` (resolved via the registry), `temperature`, and `experimental_telemetry` — all from the stored prompt. Properties you set after the spread (like a client-selected model) take precedence. - -**Which form to call:** +The managed `streamText` carries the stored prompt's `system`, `model` (resolved through the agent's `registry`), sampling config, and `experimental_telemetry`. Options you pass at the call site win, apart from `system`, which throws when the prompt already set one. -| Form | Use when | -|---|---| -| `chat.toStreamTextOptions()` | Default. Wires up `prepareStep` (compaction, steering, background injection), the stored prompt's `system` / `model` / `config`, and telemetry metadata. | -| `chat.toStreamTextOptions({ registry })` | You're using [Prompts](/ai/prompts) with a provider-prefixed model string (e.g. `"anthropic:claude-sonnet-4-5"`). The registry resolves the prefix to a real model instance via `createProviderRegistry({ anthropic, openai, ... })`. | -| `chat.toStreamTextOptions({ tools })` | You want HITL tool approvals — pass the same `tools` object you give to `streamText`. The SDK then knows which tool calls need to pause on `needsApproval: true`. | -| `chat.toStreamTextOptions({ registry, tools })` | Both of the above. | +`chat.toStreamTextOptions()` remains available for the same job, and is the only option in a [custom agent](#custom-agents) or a `chat.headStart` route, where there is no `run` argument to take it from. Pass `{ registry }` when a prompt names a provider-prefixed model, and `{ tools }` when you want HITL tool approvals, so the SDK knows which calls pause on `needsApproval`. See [Prompts](/ai/prompts) for the full guide — defining templates, variable schemas, dashboard @@ -273,7 +298,7 @@ The `run` function receives three abort signals: ```ts export const myChat = chat.agent({ id: "my-chat", - run: async ({ messages, signal, stopSignal, cancelSignal }) => { + run: async ({ messages, signal, stopSignal, cancelSignal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, @@ -302,7 +327,7 @@ export const myChat = chat.agent({ data: { messages: uiMessages, lastStoppedAt: stopped ? new Date() : undefined }, }); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -312,11 +337,10 @@ You can also check stop status from **anywhere** during a turn using `chat.isSto ```ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText } from "ai"; export const myChat = chat.agent({ id: "my-chat", - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, @@ -369,7 +393,7 @@ const sendEmail = tool({ export const myChat = chat.agent({ id: "my-chat", - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, @@ -405,12 +429,12 @@ Users can send messages while the agent is executing tool calls. With `pendingMe ```ts export const myChat = chat.agent({ id: "my-chat", + registry, pendingMessages: { shouldInject: ({ steps }) => steps.length > 0, }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ - ...chat.toStreamTextOptions({ registry }), messages, tools: { /* ... */ @@ -436,6 +460,7 @@ Inject context from background work into the conversation using `chat.inject()`. ```ts export const myChat = chat.agent({ id: "my-chat", + registry, onTurnComplete: async ({ messages }) => { chat.defer( (async () => { @@ -453,8 +478,8 @@ export const myChat = chat.agent({ })() ); }, - run: async ({ messages, signal }) => { - return streamText({ ...chat.toStreamTextOptions({ registry }), messages, abortSignal: signal }); + run: async ({ messages, signal, streamText }) => { + return streamText({ messages, abortSignal: signal }); }, }); ``` @@ -565,7 +590,7 @@ export const myChat = chat.agent({ }, ]; }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -590,7 +615,7 @@ By default, a chat agent stays idle after each turn waiting for the next user me ```ts chat.agent({ id: "one-shot", - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { // Single-response agent — exit after this turn. chat.endRun(); return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); @@ -613,7 +638,7 @@ Use this when the agent knows its work is done (budget exhausted, goal achieved, Override how long the run stays suspended waiting for the next message. Call from inside `run()`: ```ts -run: async ({ messages, signal }) => { +run: async ({ messages, signal, streamText }) => { chat.setTurnTimeout("2h"); // Wait longer for this conversation return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, @@ -624,7 +649,7 @@ run: async ({ messages, signal }) => { Override how long the run stays idle (active, using compute) after each turn: ```ts -run: async ({ messages, signal }) => { +run: async ({ messages, signal, streamText }) => { chat.setIdleTimeoutInSeconds(60); // Stay idle for 1 minute return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, @@ -659,7 +684,7 @@ export const myChat = chat.agent({ return "Something went wrong. Please try again."; }, }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -690,7 +715,7 @@ export const myChat = chat.agent({ sendReasoning: true, // Forward model reasoning (default: true) sendSources: true, // Forward source citations (default: false) }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -708,7 +733,7 @@ export const myChat = chat.agent({ uiMessageStreamOptions: { generateMessageId: () => uuidv7(), }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -728,7 +753,7 @@ export const myChat = chat }) .agent({ id: "my-chat", - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -746,7 +771,7 @@ export const myChat = chat Override per-turn with `chat.setUIMessageStreamOptions()` — per-turn values merge with the static config (per-turn wins on conflicts). The override is cleared automatically after each turn. ```ts -run: async ({ messages, clientData, signal }) => { +run: async ({ messages, clientData, signal, streamText }) => { // Enable reasoning only for certain models if (clientData.model?.includes("claude")) { chat.setUIMessageStreamOptions({ sendReasoning: true }); @@ -772,7 +797,6 @@ If you need full control over task options, use the standard `task()` with `Chat ```ts import { task } from "@trigger.dev/sdk"; import { chat, type ChatTaskPayload } from "@trigger.dev/sdk/ai"; -import { streamText } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; export const manualChat = task({ diff --git a/docs/ai-chat/background-injection.mdx b/docs/ai-chat/background-injection.mdx index 92fa3dd1336..6d26e0d9e09 100644 --- a/docs/ai-chat/background-injection.mdx +++ b/docs/ai-chat/background-injection.mdx @@ -33,6 +33,7 @@ The most powerful pattern combines `chat.defer()` (background work) with `chat.i ```ts export const myChat = chat.agent({ id: "my-chat", + registry, onTurnComplete: async ({ messages }) => { // Kick off background analysis, doesn't block the turn chat.defer( @@ -47,9 +48,8 @@ export const myChat = chat.agent({ })() ); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ - ...chat.toStreamTextOptions({ registry }), messages, abortSignal: signal, stopWhen: stepCountIs(15), @@ -77,7 +77,7 @@ A cheap model reviews the agent's response after each turn and injects coaching ```ts import { chat } from "@trigger.dev/sdk/ai"; import { prompts } from "@trigger.dev/sdk"; -import { streamText, generateObject, createProviderRegistry, stepCountIs } from "ai"; +import { generateObject, createProviderRegistry, stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; import { z } from "zod"; @@ -98,6 +98,7 @@ Be concise. Only flag issues worth fixing.`, export const myChat = chat.agent({ id: "my-chat", + registry, onTurnComplete: async ({ messages }) => { chat.defer( (async () => { @@ -139,9 +140,8 @@ export const myChat = chat.agent({ })() ); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ - ...chat.toStreamTextOptions({ registry }), messages, abortSignal: signal, stopWhen: stepCountIs(15), @@ -170,7 +170,7 @@ export const myChat = chat.agent({ // Analytics: fire-and-forget, irrelevant to resume. chat.defer(analytics.track("turn_started", { chatId, runId })); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); diff --git a/docs/ai-chat/compaction.mdx b/docs/ai-chat/compaction.mdx index 0a8ed48c7ee..c78999ac9d9 100644 --- a/docs/ai-chat/compaction.mdx +++ b/docs/ai-chat/compaction.mdx @@ -19,11 +19,12 @@ Provide `shouldCompact` to decide when to compact and `summarize` to generate th ```ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText, generateText, stepCountIs } from "ai"; +import { generateText, stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; export const myChat = chat.agent({ id: "my-chat", + registry, compaction: { shouldCompact: ({ totalTokens }) => (totalTokens ?? 0) > 80_000, summarize: async ({ messages }) => { @@ -34,9 +35,8 @@ export const myChat = chat.agent({ return result.text; }, }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ - ...chat.toStreamTextOptions({ registry }), messages, abortSignal: signal, stopWhen: stepCountIs(15), @@ -91,7 +91,7 @@ export const myChat = chat.agent({ ...uiMessages.slice(-4), // Keep the last 4 messages ], }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -185,7 +185,7 @@ export const myChat = chat.agent({ data: { chatId, summary, totalTokens, messageCount }, }); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); @@ -201,7 +201,7 @@ Define a `compact` action that reuses your existing `summarize` function: ```ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText, generateText, generateId, convertToModelMessages } from "ai"; +import { generateText, generateId, convertToModelMessages } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; import { z } from "zod"; @@ -243,7 +243,7 @@ export const myChat = chat.agent({ ]); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }); }, }); diff --git a/docs/ai-chat/fast-starts.mdx b/docs/ai-chat/fast-starts.mdx index 61d9c468f25..9bcb1ae0490 100644 --- a/docs/ai-chat/fast-starts.mdx +++ b/docs/ai-chat/fast-starts.mdx @@ -239,11 +239,11 @@ This is an **import-chain** problem, not a runtime one. A "we'll strip the execu export const myChat = chat.agent({ id: "my-chat", - run: async ({ messages, signal }) => + run: async ({ messages, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools: chatTools }), model: anthropic("claude-sonnet-4-6"), messages, + tools: chatTools, stopWhen: stepCountIs(10), abortSignal: signal, }), @@ -261,18 +261,23 @@ This is an **import-chain** problem, not a runtime one. A "we'll strip the execu export const chatHandler = chat.headStart({ agentId: "my-chat", - run: async ({ chat: helper }) => + run: async ({ streamText }) => streamText({ - ...helper.toStreamTextOptions({ tools: headStartTools }), model: anthropic("claude-sonnet-4-6"), system: "You are a helpful assistant.", + tools: headStartTools, }), }); ``` - - Don't set `stopWhen` here. The spread pins it to `stepCountIs(1)`, and overriding it makes the handler run steps the agent is supposed to own — the handover then splices a stream that has already moved past step 1. - + + That `streamText` is the SDK's, not the one from `ai`. It pins `messages`, + `stopWhen: stepCountIs(1)` and `abortSignal`, which the handover depends on: + running past step 1 would splice a stream the agent is supposed to own. Setting + any of the three at the call site is a type error, and a throw if you get past + the types, rather than breaking the handover quietly. `chat.toStreamTextOptions()` is still there if you want to build the + options yourself. + Use the **same model** on both sides (route handler and `chat.agent`) to avoid a tone or style shift between step 1 and step 2+. Your LLM provider keys stay server-side in your warm process — Trigger.dev never holds them in this design. @@ -625,8 +630,7 @@ chat.headStart({ export const chatHandler = chat.headStart({ agentId: "my-chat", triggerConfig: { tags: ["org:acme"], queue: "chat", machine: "small-2x" }, - run: async ({ chat: helper }) => - streamText({ ...helper.toStreamTextOptions({ tools: headStartTools }), model, system }), + run: async ({ streamText }) => streamText({ model, system, tools: headStartTools }), }); ``` @@ -684,11 +688,11 @@ export async function POST(req: Request) { agentId: "my-chat", chatId, // session externalId; reuse it on the destination page messages, // first-turn user history - run: async ({ chat: helper }) => + run: async ({ streamText }) => streamText({ - ...helper.toStreamTextOptions({ tools: headStartTools }), model: anthropic("claude-sonnet-4-6"), system: "You are a helpful assistant.", + tools: headStartTools, }), }); diff --git a/docs/ai-chat/migrating-from-a-route-handler.mdx b/docs/ai-chat/migrating-from-a-route-handler.mdx index 9e4f41e2011..eee3059e6a7 100644 --- a/docs/ai-chat/migrating-from-a-route-handler.mdx +++ b/docs/ai-chat/migrating-from-a-route-handler.mdx @@ -56,7 +56,7 @@ Then make these changes: - Create a `chat.agent` task in `trigger/chat.ts`. Move the existing `streamText` call into its `run` function UNCHANGED — same model, same `system`, same `temperature`, same `stopWhen`, same provider options. Do not rewrite the prompt or swap the model. -- Spread `...chat.toStreamTextOptions({ tools })` as the FIRST property of that +- Take `streamText` from the `run` argument, so the managed options apply to that `streamText` call, so the explicit options after it still win. - `run` receives `ModelMessage[]` already. Delete the `convertToModelMessages` call. - Forward the `signal` from `run` as `abortSignal` so Stop works. @@ -145,10 +145,10 @@ import { tools } from "@/lib/tools"; export const myChat = chat.agent({ id: "my-chat", tools, - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ // Spread first, so every option below still wins. - ...chat.toStreamTextOptions({ tools }), + tools, model: anthropic("claude-sonnet-4-5"), system: "You are a helpful assistant.", messages, @@ -163,10 +163,10 @@ Four things changed inside the `streamText` call, and `tools` moved onto the age - **`messages` arrives as `ModelMessage[]`.** The runtime converts the frontend's `UIMessage[]` for you, so `convertToModelMessages` is gone. - **`abortSignal` comes from `signal` on the payload**, not `req.signal`. It fires on stop and on cancel. - **Return the `StreamTextResult`.** It's piped to the frontend automatically — no `toUIMessageStreamResponse`. If `streamText` is buried in a helper, call `await chat.pipe(result)` from anywhere in the task instead and let `run` resolve `void`. -- **`...chat.toStreamTextOptions()` is spread first.** It wires up the `prepareStep` callback behind [compaction](/ai-chat/compaction), [mid-turn steering](/ai-chat/pending-messages), and [background injection](/ai-chat/background-injection), plus the system prompt set via [`chat.prompt()`](/ai-chat/backend#using-prompts) and telemetry. +- **`streamText` comes from the `run` argument, not from `ai`.** It carries the `prepareStep` behind [compaction](/ai-chat/compaction), [mid-turn steering](/ai-chat/pending-messages) and [background injection](/ai-chat/background-injection), plus the system prompt set via [`chat.prompt()`](/ai-chat/backend#using-prompts) and telemetry. - Omitting `...chat.toStreamTextOptions()` throws no error — compaction, steering, and background injection just silently never run. Spread it first so any explicit override you write after it takes precedence. + Importing `streamText` from `ai` instead throws no error: compaction, steering and background injection never run. Spreading `chat.toStreamTextOptions()` into the imported one is the equivalent, and is what a `chat.headStart` route has to do, since it has no `run` argument. There's no `maxDuration` equivalent to set. A turn isn't bounded by a function timeout; a run stays alive across turns and suspends when nothing is happening. @@ -357,9 +357,9 @@ export const myChat = chat.agent({ }), ]); }, - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools }), + tools, model: anthropic("claude-sonnet-4-5"), system: "You are a helpful assistant.", messages, @@ -468,18 +468,21 @@ Head Start brings the route handler back for exactly that first turn. It runs st export const chatHandler = chat.headStart({ agentId: "my-chat", - run: async ({ chat: helper }) => + run: async ({ streamText }) => streamText({ - ...helper.toStreamTextOptions({ tools: headStartTools }), model: anthropic("claude-sonnet-4-5"), system: "You are a helpful assistant.", + tools: headStartTools, }), }); ``` - - Spread `toStreamTextOptions()` first and add only your own keys after it. It owns `messages`, `tools`, `abortSignal`, and `stopWhen` — and unlike the agent-side spread, re-setting any of those breaks the handover rather than degrading it. `stopWhen` in particular is pinned to `stepCountIs(1)`: the agent, not the handler, runs step 2 onward. - + + That `streamText` owns `messages`, `abortSignal` and `stopWhen`. Passing one is + a type error, and a runtime throw if you get past the types. Unlike the agent side, re-setting one breaks the + handover rather than degrading it: `stopWhen` is pinned to `stepCountIs(1)` + because the agent, not the handler, runs step 2 onward. + Your provider keys never leave your server — the first-turn model call runs in your process, so that environment needs whatever the model requires. @@ -558,7 +561,7 @@ The shape is identical outside Next.js. The agent task and the React component d **The head-start route dies mid-turn on Vercel.** The handler holds the SSE response open until the agent signals turn-complete, so the function timeout has to cover the whole turn, not just step 1. Set `maxDuration` on that route segment. -**Compaction and steering do nothing.** The `...chat.toStreamTextOptions()` spread is missing, or something before it in the object is overwriting `prepareStep`. Spread it as the first property. +**Compaction and steering do nothing.** `run` is calling the `streamText` imported from `ai` rather than the one in its argument. Destructure `streamText` from the `run` argument, or spread `chat.toStreamTextOptions()` into the imported one. **`toModelOutput` works on the first turn, then stops.** Tools are declared only on `streamText`. Declare the same set on `chat.agent({ tools })` too, and read it back off the `run` payload. diff --git a/docs/ai-chat/patterns/skills.mdx b/docs/ai-chat/patterns/skills.mdx index 68930b89f1d..f5da0861793 100644 --- a/docs/ai-chat/patterns/skills.mdx +++ b/docs/ai-chat/patterns/skills.mdx @@ -79,7 +79,7 @@ The **body** is loaded on demand via the `loadSkill` tool when the agent decides ```ts trigger/chat.ts import { chat } from "@trigger.dev/sdk/ai"; import { skills } from "@trigger.dev/sdk"; -import { streamText, stepCountIs } from "ai"; +import { stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; const timeUtilsSkill = skills.define({ @@ -92,12 +92,11 @@ export const agent = chat.agent({ onChatStart: async () => { chat.skills.set([await timeUtilsSkill.local()]); }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, - ...chat.toStreamTextOptions(), stopWhen: stepCountIs(15), }); }, @@ -111,7 +110,7 @@ export const agent = chat.agent({ `skill.local()` reads the bundled `SKILL.md` from disk and returns a `ResolvedSkill` with the parsed frontmatter + body + on-disk path. -`chat.skills.set([...])` stores the resolved skills for the current run. `chat.toStreamTextOptions()` spreads them into `streamText` automatically: +`chat.skills.set([...])` stores the resolved skills for the current run. The `streamText` from `run`'s argument picks them up automatically: - The frontmatter `description` lands in the system prompt under "Available skills:". - Three tools are added: `loadSkill`, `readFile`, `bash` — scoped per skill. @@ -169,12 +168,10 @@ return streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, - ...chat.toStreamTextOptions({ - tools: { - webFetch, // your tool - deepResearch, // your tool - }, - }), + tools: { + webFetch, // your tool + deepResearch, // your tool + }, stopWhen: stepCountIs(15), }); ``` diff --git a/docs/ai-chat/pending-messages.mdx b/docs/ai-chat/pending-messages.mdx index a4dec357a47..cee8495035f 100644 --- a/docs/ai-chat/pending-messages.mdx +++ b/docs/ai-chat/pending-messages.mdx @@ -30,18 +30,18 @@ Add `pendingMessages` to your `chat.agent` configuration: ```ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText, stepCountIs } from "ai"; +import { stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; export const myChat = chat.agent({ id: "my-chat", + registry, pendingMessages: { // Only inject when there are completed steps (tool calls happened) shouldInject: ({ steps }) => steps.length > 0, }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ - ...chat.toStreamTextOptions({ registry }), messages, tools: { /* ... */ }, abortSignal: signal, diff --git a/docs/ai-chat/prompt-caching.mdx b/docs/ai-chat/prompt-caching.mdx index 49d43dc0397..1a30b475f15 100644 --- a/docs/ai-chat/prompt-caching.mdx +++ b/docs/ai-chat/prompt-caching.mdx @@ -16,7 +16,7 @@ A request renders as `tools` → `system` → `messages`. There are three prefix | Region | How to cache it | Stability | | --- | --- | --- | -| System prompt (+ tools) | `cacheControl` / `systemProviderOptions` on `chat.toStreamTextOptions()`, or `providerOptions` on `chat.prompt.set()` | Set once, never changes — the highest-value target | +| System prompt (+ tools) | `cacheControl` / `systemProviderOptions` on `chat.agent()`, or `providerOptions` on `chat.prompt.set()` | Set once, never changes — the highest-value target | | Conversation history | `prepareMessages` adds a breakpoint to the last message | Grows append-only across turns | | Tool definitions | Stable as long as your tool set doesn't change between turns | Render at position 0 — changing them invalidates everything | @@ -32,23 +32,22 @@ The system prompt (your `chat.prompt` text plus any skills preamble) is usually Three ways to opt in, depending on where you'd rather express it. -**`cacheControl` at the `streamText` call site** — the Anthropic-flavored one-liner: +**`cacheControl` on the agent** — the Anthropic-flavored one-liner: ```ts /trigger/chat.ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; export const myChat = chat.agent({ id: "my-chat", + cacheControl: { type: "ephemeral" }, onChatStart: async () => { chat.prompt.set(SYSTEM_PROMPT); // a large, stable instruction block }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-6"), // Caches the system block with a 5-minute breakpoint. - ...chat.toStreamTextOptions({ cacheControl: { type: "ephemeral" } }), messages, abortSignal: signal, }); @@ -59,17 +58,19 @@ export const myChat = chat.agent({ **`systemProviderOptions`** is the provider-agnostic form — pass the raw `providerOptions` so it composes with any provider: ```ts /trigger/chat.ts -return streamText({ - model: anthropic("claude-sonnet-4-6"), - ...chat.toStreamTextOptions({ - systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } }, - }), - messages, - abortSignal: signal, +export const myChat = chat.agent({ + id: "my-chat", + systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } }, + run: async ({ messages, signal, streamText }) => + streamText({ + model: anthropic("claude-sonnet-4-6"), + messages, + abortSignal: signal, + }), }); ``` -**`providerOptions` on `chat.prompt.set()`** co-locates the intent with where the prompt is defined. It carries through to `toStreamTextOptions()` with no call-site change: +**`providerOptions` on `chat.prompt.set()`** co-locates the intent with where the prompt is defined. It carries through to the managed `streamText` with no call-site change: ```ts /trigger/chat.ts onChatStart: async () => { @@ -77,17 +78,16 @@ onChatStart: async () => { providerOptions: { anthropic: { cacheControl: { type: "ephemeral" } } }, }); }, -run: async ({ messages, signal }) => { +run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-6"), - ...chat.toStreamTextOptions(), // already cached messages, abortSignal: signal, }); }, ``` -If more than one is set, the call-site option wins: `systemProviderOptions` overrides `cacheControl`, and both override `chat.prompt.set`'s `providerOptions`. There's no deep merge — the most specific option replaces the rest. +If more than one is set, the most specific wins: `systemProviderOptions` overrides `cacheControl`, and both override `chat.prompt.set`'s `providerOptions`. There's no deep merge — the most specific option replaces the rest. Use the 1-hour cache for prefixes that sit idle longer than 5 minutes between turns: `cacheControl: { type: "ephemeral", ttl: "1h" }`. Writes cost more (2× vs 1.25×), so it pays off only when reads span the longer window. @@ -100,6 +100,7 @@ Place a breakpoint on the last message and the entire conversation prefix up to ```ts /trigger/chat.ts export const myChat = chat.agent({ id: "my-chat", + cacheControl: { type: "ephemeral" }, prepareMessages: async ({ messages }) => { if (messages.length === 0) return messages; const last = messages[messages.length - 1]; @@ -114,10 +115,9 @@ export const myChat = chat.agent({ }, ]; }, - run: async ({ messages, signal }) => { + run: async ({ messages, signal, streamText }) => { return streamText({ model: anthropic("claude-sonnet-4-6"), - ...chat.toStreamTextOptions({ cacheControl: { type: "ephemeral" } }), messages, abortSignal: signal, }); @@ -149,11 +149,10 @@ Caching is provider-specific, and most providers don't use per-block breakpoints ```ts /trigger/chat.ts // Amazon Bedrock -return streamText({ - ...chat.toStreamTextOptions({ - systemProviderOptions: { bedrock: { cachePoint: { type: "default" } } }, - }), - messages, +export const myChat = chat.agent({ + id: "my-chat", + systemProviderOptions: { bedrock: { cachePoint: { type: "default" } } }, + run: async ({ messages, streamText }) => streamText({ messages }), }); ``` @@ -166,14 +165,13 @@ Usage reporting is normalized. Each provider reports cache tokens under its own The turn's usage carries cache token counts. `chat.agent` accumulates them across turns and hands them to `run` as `previousTurnUsage` (last turn) and `totalUsage` (whole chat), both `LanguageModelUsage`: ```ts /trigger/chat.ts -run: async ({ messages, signal, previousTurnUsage }) => { +run: async ({ messages, signal, previousTurnUsage, streamText }) => { // After turn 1, cacheReadTokens should be > 0 on a stable prefix. console.log("cache read", previousTurnUsage?.inputTokenDetails?.cacheReadTokens); console.log("cache write", previousTurnUsage?.inputTokenDetails?.cacheWriteTokens); return streamText({ model: anthropic("claude-sonnet-4-6"), - ...chat.toStreamTextOptions({ cacheControl: { type: "ephemeral" } }), messages, abortSignal: signal, }); diff --git a/docs/ai-chat/quick-start.mdx b/docs/ai-chat/quick-start.mdx index 52526b6b602..0838eb18102 100644 --- a/docs/ai-chat/quick-start.mdx +++ b/docs/ai-chat/quick-start.mdx @@ -16,19 +16,16 @@ The chat surface works with Vercel AI SDK **v5, v6, or v7**; install whichever m ```ts trigger/chat.ts import { chat } from "@trigger.dev/sdk/ai"; - import { streamText, stepCountIs } from "ai"; + import { stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; export const myChat = chat.agent({ id: "my-chat", - run: async ({ messages, signal }) => { + // `streamText` here is the SDK's, not the one from `ai`: it carries + // compaction, steering, background injection, the system prompt and + // telemetry, so none of them have to be wired up by hand. + run: async ({ messages, signal, streamText }) => { return streamText({ - // Spread chat.toStreamTextOptions() FIRST — it wires up - // prepareStep (compaction, steering, background injection), - // the system prompt set via chat.prompt(), and telemetry. - // Skipping this is the single most common cause of subtle - // bugs (silent broken compaction, missing steering, etc.). - ...chat.toStreamTextOptions(), model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, @@ -38,9 +35,12 @@ The chat surface works with Vercel AI SDK **v5, v6, or v7**; install whichever m }); ``` - - **Always spread `chat.toStreamTextOptions()` into your `streamText` call.** It wires up the `prepareStep` callback that drives compaction, mid-turn steering, and background injection — features that silently no-op if the spread is missing. Spread it **first** so any explicit overrides (e.g. a custom `prepareStep`) win. - + + Take `streamText` from `run`'s argument rather than importing it from `ai`. The + imported one drives no `prepareStep`, so compaction, mid-turn steering and + background injection never run, and nothing reports it. Spreading + `chat.toStreamTextOptions()` into the imported one does the same job by hand. + For a **custom** [`UIMessage`](https://sdk.vercel.ai/docs/reference/ai-sdk-core/ui-message) subtype (typed `data-*` parts, tool map, etc.), define the agent with [`chat.withUIMessage<...>().agent({...})`](/ai-chat/types) instead of `chat.agent`. diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx index 18ee0c58190..5a7b7302fd7 100644 --- a/docs/ai-chat/reference.mdx +++ b/docs/ai-chat/reference.mdx @@ -52,6 +52,10 @@ Options for `chat.agent()`. | `onTurnComplete` | `(event: TurnCompleteEvent) => Promise \| void` | — | Fires after each turn completes (stream closed) | | `onCompacted` | `(event: CompactedEvent) => Promise \| void` | — | Fires when compaction occurs. Includes `writer`. See [Compaction](/ai-chat/compaction) | | `compaction` | `ChatAgentCompactionOptions` | — | Automatic context compaction. See [Compaction](/ai-chat/compaction) | +| `registry` | `{ languageModel(id: string): unknown }` | — | A provider registry, so the managed `streamText` can resolve a model set through `chat.prompt.set()` | +| `system` | `string \| SystemModelMessage` | — | The agent's system prompt. Injected instructions append to it. Set it here, at the `streamText` call site, or through `chat.prompt.set()`, but only in one of them | +| `cacheControl` | `SystemCacheControl` | — | Mark the system prompt for provider-side caching. See [Prompt caching](/ai-chat/prompt-caching) | +| `systemProviderOptions` | `ProviderMetadata` | — | Raw provider options for the system block. Takes precedence over `cacheControl` | | `pendingMessages` | `PendingMessagesOptions` | — | Mid-execution message injection. See [Pending Messages](/ai-chat/pending-messages) | | `prepareMessages` | `(event: PrepareMessagesEvent) => ModelMessage[]` | — | Transform model messages before use (cache breaks, context injection, etc.) | | `tools` | `ToolSet \| ((event: ResolveToolsEvent) => ToolSet \| Promise)` | — | Tools for this agent. Threads each tool's `toModelOutput` through cross-turn history re-conversion, and hands the resolved set back on the run payload. Static set or per-turn function. See [Tools](/ai-chat/tools). | @@ -98,6 +102,7 @@ The payload passed to the `run` function. | `ctx` | `TaskRunContext` | Full task run context — same as `task` `run`’s `{ ctx }` | | `messages` | `ModelMessage[]` | Model-ready messages — pass directly to `streamText` | | `tools` | `ToolSet` | Resolved tools declared on the agent config (empty object when none). Pass straight to `streamText`. See [Tools](/ai-chat/tools). | +| `streamText` | `typeof streamText` | The AI SDK's `streamText` with this agent's managed options already applied: the prompt, skill tools, telemetry, and the `prepareStep` that delivers steering, compaction and injected context. Prefer it over importing `streamText` from `ai`. See [The managed streamText](/ai-chat/backend#the-managed-streamtext). | | `chatId` | `string` | Your conversation ID (the session's `externalId`) | | `sessionId` | `string` | Friendly ID of the backing Session (`session_*`). Use with `sessions.open()` for advanced cases. Always set — every chat.agent run is bound to a Session. | | `trigger` | `"submit-message" \| "regenerate-message"` | What triggered the request | @@ -240,6 +245,7 @@ Passed to the `onAction` callback. See [Actions](/ai-chat/actions). | `clientData` | Typed by `clientDataSchema` | Custom data from the frontend | | `uiMessages` | `UIMessage[]` | Accumulated UI messages (after hydration, if set) | | `messages` | `ModelMessage[]` | Accumulated model messages (after hydration, if set) | +| `streamText` | `ChatStreamText` | `streamText` with the agent's managed options applied, the same one `run()` receives. Use it for a response produced from an action so the answer carries the agent's own prompt and tools | ## TurnStartEvent diff --git a/docs/ai-chat/tools.mdx b/docs/ai-chat/tools.mdx index 7fa91d4beb8..d9c99510a40 100644 --- a/docs/ai-chat/tools.mdx +++ b/docs/ai-chat/tools.mdx @@ -8,7 +8,7 @@ description: "Declare tools on chat.agent so toModelOutput survives across turns ```ts import { chat } from "@trigger.dev/sdk/ai"; -import { streamText, stepCountIs, tool } from "ai"; +import { stepCountIs, tool } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; import { z } from "zod"; @@ -23,9 +23,9 @@ const tools = { export const myChat = chat.agent({ id: "my-chat", tools, // ← declare here - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools }), // ← the same set, handed back on the payload + tools, model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, @@ -80,9 +80,9 @@ const tools = { export const chartChat = chat.agent({ id: "chart-chat", tools, // ← without this, the image is "remembered" on turn 1 and gone from turn 2 - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools }), + tools, model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, @@ -104,9 +104,9 @@ export const myChat = chat searchDocs, ...(clientData?.plan === "pro" ? { deepResearch } : {}), }), - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools }), + tools, model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, @@ -131,10 +131,10 @@ The resolved set is what lands on the `run()` payload's `tools`. The `run()` payload's `tools` is typed to whatever you declared, so you can pass it straight through without re-importing the map: ```ts -run: async ({ messages, tools, signal }) => { +run: async ({ messages, tools, signal, streamText }) => { // `tools` is typed as your tool set, not a broad `ToolSet` return streamText({ - ...chat.toStreamTextOptions({ tools }), + tools, model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, diff --git a/packages/cli-v3/skills/trigger-authoring-chat-agent/SKILL.md b/packages/cli-v3/skills/trigger-authoring-chat-agent/SKILL.md index 2935baae22d..8eadc075c4b 100644 --- a/packages/cli-v3/skills/trigger-authoring-chat-agent/SKILL.md +++ b/packages/cli-v3/skills/trigger-authoring-chat-agent/SKILL.md @@ -2,7 +2,7 @@ name: trigger-authoring-chat-agent description: > Author and run a durable AI chat agent with chat.agent from @trigger.dev/sdk/ai: the per-turn - run loop, why you MUST spread ...chat.toStreamTextOptions() first, returning a StreamTextResult + run loop, why you MUST take streamText from the run argument rather than importing it from ai, returning a StreamTextResult vs calling chat.pipe(), the two server actions (chat.createStartSessionAction + auth.createPublicToken), and wiring useChat to useTriggerChatTransport. Load this when building, modifying, or debugging a chat backend (the agent task or its lifecycle hooks) or its React @@ -16,26 +16,29 @@ library: trigger.dev The full, version-pinned reference ships **inside your installed `@trigger.dev/sdk`**. Read it before writing code — it always matches the SDK version in this project, so it never drifts: -- **Skill:** `node_modules/@trigger.dev/sdk/skills/trigger-authoring-chat-agent/SKILL.md` — the per-turn run loop, `chat.toStreamTextOptions()`, the two server actions, typed tools/data parts, and the React transport. +- **Skill:** `node_modules/@trigger.dev/sdk/skills/trigger-authoring-chat-agent/SKILL.md` — the per-turn run loop, the managed `streamText`, the two server actions, typed tools/data parts, and the React transport. - **Docs:** the full, version-pinned docs ship bundled at `node_modules/@trigger.dev/sdk/docs/ai-chat/`; the skill above lists the exact pages it draws from in its `sources:` frontmatter. Grep for an API, e.g. `grep -rl "toStreamTextOptions" node_modules/@trigger.dev/sdk/docs/`. If those paths don't exist, `@trigger.dev/sdk` isn't installed yet — install it first. In a non-hoisted layout, resolve the package with `node -p "require.resolve('@trigger.dev/sdk/package.json')"` and read `skills/` + `docs/` beside it. ## Common mistakes -- **CRITICAL: forgetting `...chat.toStreamTextOptions()`.** +- **CRITICAL: calling the `streamText` imported from `ai`.** ```ts // Wrong - compaction / steering / background injection silently no-op - return streamText({ model, messages, abortSignal: signal }); - // Correct - spread FIRST so explicit overrides win - return streamText({ ...chat.toStreamTextOptions(), model, messages, abortSignal: signal }); + import { streamText } from "ai"; + run: async ({ messages, signal }) => streamText({ model, messages, abortSignal: signal }); + // Correct - the run argument's streamText carries the managed options + run: async ({ messages, signal, streamText }) => streamText({ model, messages, abortSignal: signal }); ``` - It wires the `prepareStep` callback behind compaction, mid-turn steering, and background - injection, injects the system prompt from `chat.prompt()`, resolves the registry model, and adds - telemetry. Omitting it makes all of those silently no-op with no error. + The SDK's one carries the `prepareStep` behind compaction, mid-turn steering and background + injection, the system prompt from `chat.prompt()` or `chat.agent({ system })`, the registry-resolved + model, and telemetry. The imported one carries none of it, with no error. + `...chat.toStreamTextOptions()` does the same job by hand, and is what a `chat.headStart` route or a + custom agent has to use, since neither has a `run` argument. - **Declaring tools only on `streamText`.** Also declare them on `chat.agent({ tools })`, read them - back from `run`, and pass `chat.toStreamTextOptions({ tools })`. Otherwise each tool's + back from `run`, and pass that set as `tools`. Otherwise each tool's `toModelOutput` runs on turn 1 but is dropped when history is re-converted on later turns. - **Not forwarding `signal` for stop.** Without `abortSignal: signal`, Stop updates the UI but the diff --git a/packages/trigger-sdk/skills/trigger-authoring-chat-agent/SKILL.md b/packages/trigger-sdk/skills/trigger-authoring-chat-agent/SKILL.md index d34b722b5ec..b4ddb9771de 100644 --- a/packages/trigger-sdk/skills/trigger-authoring-chat-agent/SKILL.md +++ b/packages/trigger-sdk/skills/trigger-authoring-chat-agent/SKILL.md @@ -2,8 +2,8 @@ name: trigger-authoring-chat-agent description: > Author and run a durable AI chat agent with chat.agent from @trigger.dev/sdk/ai: the per-turn - run loop, why you MUST spread ...chat.toStreamTextOptions() first, returning a StreamTextResult - vs calling chat.pipe(), the two server actions (chat.createStartSessionAction + + run loop, why you MUST take streamText from the run argument rather than importing it from ai, + returning a StreamTextResult vs calling chat.pipe(), the two server actions (chat.createStartSessionAction + auth.createPublicToken), and wiring useChat to useTriggerChatTransport. Load this when building, modifying, or debugging a chat backend (the agent task or its lifecycle hooks) or its React transport, when declaring typed tools or custom data parts, or when migrating a plain AI SDK @@ -47,10 +47,9 @@ import { anthropic } from "@ai-sdk/anthropic"; export const myChat = chat.agent({ id: "my-chat", - run: async ({ messages, signal }) => + // `streamText` below is the SDK's, from the run argument. See "Common mistakes". + run: async ({ messages, signal, streamText }) => streamText({ - // Spread this FIRST. See "Common mistakes". - ...chat.toStreamTextOptions(), model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal, @@ -123,14 +122,16 @@ inside nested helpers, call `await chat.pipe(result)` from anywhere in the task ```ts export const agentChat = chat.agent({ id: "agent-chat", - run: async ({ messages }) => { - await runAgentLoop(messages); // don't return; pipe inside + run: async ({ messages, streamText }) => { + await runAgentLoop(messages, streamText); // don't return; pipe inside }, }); -async function runAgentLoop(messages: ModelMessage[]) { +// A loop factored out of `run` takes `streamText` as an argument, so it keeps the +// managed options. `ChatStreamText` (from `@trigger.dev/sdk/ai`) types the parameter. +// `chat.toStreamTextOptions()` is the alternative when threading it down is impractical. +async function runAgentLoop(messages: ModelMessage[], streamText: ChatStreamText) { const result = streamText({ - ...chat.toStreamTextOptions(), model: anthropic("claude-sonnet-4-5"), messages, }); @@ -138,10 +139,10 @@ async function runAgentLoop(messages: ModelMessage[]) { } ``` -### 2. Typed tools (declare on config AND spread back) +### 2. Typed tools (declare on config AND pass back) Declare tools on `chat.agent({ tools })`, read them back typed from the `run()` payload, and pass -that set to `chat.toStreamTextOptions({ tools })`. One declaration flows everywhere. +that set as `tools`. One declaration flows everywhere. ```ts import { tool, stepCountIs } from "ai"; @@ -158,11 +159,11 @@ const tools = { export const myChat = chat.agent({ id: "my-chat", tools, // so toModelOutput survives across turns - run: async ({ messages, tools, signal }) => + run: async ({ messages, tools, signal, streamText }) => streamText({ - ...chat.toStreamTextOptions({ tools }), // same set, handed back typed model: anthropic("claude-sonnet-4-5"), messages, + tools, // same set, handed back typed abortSignal: signal, stopWhen: stepCountIs(15), }), @@ -203,8 +204,8 @@ export const myChat = chat onTurnStart: async ({ uiMessages, writer }) => { writer.write({ type: "data-turn-status", data: { status: "preparing" } }); }, - run: async ({ messages, tools, signal }) => - streamText({ ...chat.toStreamTextOptions({ tools }), model, messages, abortSignal: signal }), + run: async ({ messages, tools, signal, streamText }) => + streamText({ model, messages, tools, abortSignal: signal }), }); ``` @@ -233,8 +234,8 @@ Stop is load-bearing: the `signal` passed to `run` aborts on stop or cancel. For server-side. ```ts -run: async ({ messages, signal }) => - streamText({ ...chat.toStreamTextOptions(), model, messages, abortSignal: signal, stopWhen: stepCountIs(15) }); +run: async ({ messages, signal, streamText }) => + streamText({ model, messages, abortSignal: signal, stopWhen: stepCountIs(15) }); ``` ### 6. Migrating from a plain AI SDK `streamText` route @@ -243,24 +244,29 @@ There is no API route in this model. The transport replaces the route round-trip - Delete the route handler. Move per-request auth into the two server actions from Setup step 2. - Move the `streamText` call into `run`. It already receives pre-converted `ModelMessage[]`. -- Return the `StreamTextResult` (it auto-pipes) and add `...chat.toStreamTextOptions()` first. +- Return the `StreamTextResult` (it auto-pipes) and take `streamText` from `run`'s argument, not from `ai`. - On the client, swap the `api` URL for `useTriggerChatTransport`; `useChat` stays the same shape. ## Common mistakes -- **CRITICAL: forgetting `...chat.toStreamTextOptions()`.** +- **CRITICAL: calling the `streamText` imported from `ai`.** ```ts // Wrong - compaction / steering / background injection silently no-op - return streamText({ model, messages, abortSignal: signal }); - // Correct - spread FIRST so explicit overrides win - return streamText({ ...chat.toStreamTextOptions(), model, messages, abortSignal: signal }); + import { streamText } from "ai"; + run: async ({ messages, signal }) => streamText({ model, messages, abortSignal: signal }); + // Correct - the run argument's streamText carries the managed options + run: async ({ messages, signal, streamText }) => streamText({ model, messages, abortSignal: signal }); ``` - It wires the `prepareStep` callback behind compaction, mid-turn steering, and background - injection, injects the system prompt from `chat.prompt()`, resolves the registry model, and adds - telemetry. Omitting it makes all of those silently no-op with no error. + The SDK's one carries the `prepareStep` behind compaction, mid-turn steering and background + injection, the system prompt from `chat.prompt()` or `chat.agent({ system })`, the registry-resolved + model, and telemetry. The imported one carries none of it, with no error. + `...chat.toStreamTextOptions()` does the same job by hand, and is what a `chat.headStart` route or a + custom agent has to use, since neither has a `run` argument. Spreading it and then re-setting + `tools` or `prepareStep` replaces the managed ones; the run argument's `streamText` merges `tools` + and composes `prepareStep` instead. - **Declaring tools only on `streamText`.** Also declare them on `chat.agent({ tools })`, read them - back from `run`, and pass `chat.toStreamTextOptions({ tools })`. Otherwise each tool's + back from `run`, and pass that set as `tools`. Otherwise each tool's `toModelOutput` runs on turn 1 but is dropped when history is re-converted on later turns. - **Not forwarding `signal` for stop.** Without `abortSignal: signal`, Stop updates the UI but the diff --git a/packages/trigger-sdk/skills/trigger-chat-agent-advanced/SKILL.md b/packages/trigger-sdk/skills/trigger-chat-agent-advanced/SKILL.md index a13899dfe78..cfd6cb55764 100644 --- a/packages/trigger-sdk/skills/trigger-chat-agent-advanced/SKILL.md +++ b/packages/trigger-sdk/skills/trigger-chat-agent-advanced/SKILL.md @@ -143,6 +143,12 @@ await waitUntilComplete(); 5s timeout, before `onTurnComplete`). `chat.inject(messages)` queues `ModelMessage[]` that drain at the next turn start or `prepareStep` boundary. +Two lanes, decided by role. A `role: "system"` message goes to the model's instructions, where it is +trusted like the system prompt, and applies to the next turn only. Any other role joins the +conversation and is untrusted by construction, so put checkable facts there and directives in the +system lane. The instructions lane reaches the model only through the managed `streamText` (or a +`chat.toStreamTextOptions()` spread), since that is where the SDK can set instructions. + ```ts export const myChat = chat.agent({ id: "my-chat", @@ -154,8 +160,9 @@ export const myChat = chat.agent({ })() ); }, - run: async ({ messages, signal }) => - streamText({ ...chat.toStreamTextOptions({ registry }), messages, abortSignal: signal, stopWhen: stepCountIs(15) }), + registry, + run: async ({ messages, signal, streamText }) => + streamText({ messages, abortSignal: signal, stopWhen: stepCountIs(15) }), }); ``` @@ -163,8 +170,9 @@ export const myChat = chat.agent({ `compaction.shouldCompact` decides when, `summarize` produces the summary that replaces the model messages. UI messages are preserved by default (customize via `compactUIMessages`). The `prepareStep` -that performs inner-loop compaction is auto-injected by `chat.toStreamTextOptions()`; a `prepareStep` -you pass after the spread wins. +that performs inner-loop compaction rides the managed `streamText`, which composes a `prepareStep` +you pass after it. Spreading `chat.toStreamTextOptions()` and then passing your own replaces it, +switching compaction off. ```ts compaction: { @@ -182,7 +190,14 @@ compaction: { `actionSchema` validates; `onAction` mutates via `chat.history` (`slice`, `replace`, `rollbackTo`, `remove`, `getPendingToolCalls`, `extractNewToolResults`). Actions fire `hydrateMessages` and `onAction` only, never `run()` or the turn hooks. Return a `StreamTextResult`, string, or `UIMessage` -to also emit a model response. +to also emit a model response, built with the `streamText` from `onAction`'s own argument so it +carries the agent's prompt and tools like any other turn. + +Persistence splits by model. Without `hydrateMessages` the runtime snapshots the conversation after +an action that changed it, so a rollback or a returned response survives the run ending. With +`hydrateMessages` your store is the source of truth and the runtime does not write, so mirror every +mutation yourself: a regenerate is a delete and an insert, and `chat.pipeAndCapture` hands back the +same assistant message the runtime would have captured. ```ts export const myChat = chat.agent({ @@ -195,7 +210,8 @@ export const myChat = chat.agent({ if (action.type === "undo") chat.history.slice(0, -2); if (action.type === "rollback") chat.history.rollbackTo(action.targetMessageId); }, - run: async ({ messages, signal }) => streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }), + run: async ({ messages, signal, streamText }) => + streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }), }); ``` @@ -210,18 +226,19 @@ must be **schema-only** (a module importing `ai` + `zod` only); heavy executes s ```ts import { chat } from "@trigger.dev/sdk/chat-server"; -import { streamText, stepCountIs } from "ai"; import { anthropic } from "@ai-sdk/anthropic"; import { headStartTools } from "@/lib/chat-tools/schemas"; export const chatHandler = chat.headStart({ agentId: "my-chat", - run: async ({ chat: helper }) => + // `streamText` from the run argument owns `messages`, `stopWhen` and + // `abortSignal`: the handover needs `stopWhen: stepCountIs(1)` so the agent, + // not this handler, runs step 2 onward. Passing any of them is a type error. + run: async ({ streamText }) => streamText({ - ...helper.toStreamTextOptions({ tools: headStartTools }), model: anthropic("claude-sonnet-4-6"), system: "You are helpful.", - stopWhen: stepCountIs(15), + tools: headStartTools, }), }); // Next.js: export const POST = chatHandler; Transport: headStart: "/api/chat" @@ -248,8 +265,10 @@ export const myChat = chat.agent({ ### 8. Pending messages (mid-stream user input) A message sent while a turn is streaming should NOT cancel the stream. Configure -`pendingMessages` (`shouldInject`, `prepare`, `onReceived`, `onInjected`) on the agent so the SDK's -auto-injected `prepareStep` folds them in at the next boundary. On the frontend, `usePendingMessages` +`pendingMessages` (`shouldInject`, `prepare`, `onReceived`, `onInjected`) on the agent so the managed +`streamText`'s `prepareStep` folds them in at the next boundary. An injected steering message is part +of the conversation your hooks see, so it arrives in `uiMessages` and `newUIMessages` at +`onTurnComplete` and an app persisting from there stores it without extra work. On the frontend, `usePendingMessages` returns `pending`, `steer(text)`, `queue(text)`, and `promoteToSteering(id)`; send via `transport.sendPendingMessage(chatId, uiMessage, metadata?)`. diff --git a/packages/trigger-sdk/src/imports/ai-runtime-cjs.cts b/packages/trigger-sdk/src/imports/ai-runtime-cjs.cts index 8ad336dbaeb..320e6adcdc6 100644 --- a/packages/trigger-sdk/src/imports/ai-runtime-cjs.cts +++ b/packages/trigger-sdk/src/imports/ai-runtime-cjs.cts @@ -21,6 +21,8 @@ module.exports.readUIMessageStream = ai.readUIMessageStream; // @ts-ignore module.exports.stepCountIs = ai.stepCountIs; // @ts-ignore +module.exports.streamText = ai.streamText; +// @ts-ignore module.exports.tool = ai.tool; // @ts-ignore module.exports.zodSchema = ai.zodSchema; diff --git a/packages/trigger-sdk/src/imports/ai-runtime.ts b/packages/trigger-sdk/src/imports/ai-runtime.ts index 8c713bce876..0d09b764faf 100644 --- a/packages/trigger-sdk/src/imports/ai-runtime.ts +++ b/packages/trigger-sdk/src/imports/ai-runtime.ts @@ -20,6 +20,7 @@ import { jsonSchema, readUIMessageStream, stepCountIs, + streamText, tool, zodSchema, } from "ai"; @@ -34,6 +35,7 @@ export { jsonSchema, readUIMessageStream, stepCountIs, + streamText, tool, zodSchema, }; diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 5069d406018..14b241f1629 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -43,6 +43,7 @@ import { type SessionStreamRecord, } from "@trigger.dev/core/v3"; import type { + streamText as aiStreamTextSignature, FinishReason, LanguageModelUsage, ModelMessage, @@ -68,6 +69,7 @@ import { isToolUIPart, jsonSchema, readUIMessageStream, + streamText as aiStreamText, zodSchema, } from "../imports/ai-runtime.js"; import { @@ -1652,6 +1654,16 @@ export type ChatTaskRunPayload< * Use for tags, metadata, parent run links, or any API that needs the full run record. */ ctx: TaskRunContext; + /** + * `streamText` with this agent's managed options already applied: the + * prompt from `chat.prompt.set()`, the skill tools, telemetry, and the + * `prepareStep` that delivers steering, compaction and injected context. + * + * Prefer it over importing `streamText` from `ai`. The imported one takes + * none of that unless you spread `chat.toStreamTextOptions()` yourself, and + * forgetting the spread fails silently. + */ + streamText: AiStreamTextFn; /** Token usage from the previous turn. Undefined on turn 0. */ previousTurnUsage?: LanguageModelUsage; /** Cumulative token usage across all completed turns so far. */ @@ -4625,6 +4637,12 @@ export function buildSkillTools(skills: ResolvedSkill[]): Record { * Options for {@link toStreamTextOptions}. */ export type ToStreamTextOptionsOptions = { + /** + * A base system prompt, used when `chat.prompt.set()` has not supplied one. + * `chat.agent({ system })` and a `system` passed to the managed `streamText` + * both arrive here. + */ + system?: string | SystemModelMessage; /** Additional telemetry metadata merged into `experimental_telemetry.metadata`. */ telemetry?: Record; /** @@ -4692,13 +4710,148 @@ export type SystemCacheControl = { type: "ephemeral"; ttl?: "5m" | "1h" }; * * If no prompt has been set, returns `{}` (no-op spread). */ +/** + * The AI SDK's own `streamText` signature, borrowed rather than restated. + * + * The peer range spans `ai` v5, v6 and v7, whose `streamText` options differ. + * Describing the shape here would drift against all three; `typeof` resolves to + * whichever version the user installed, so generics, overloads and tool + * inference are exactly theirs. + */ +type AiStreamTextFn = typeof aiStreamTextSignature; + +/** + * The `streamText` handed to a `chat.agent` `run()`, carrying the agent's + * managed options. + * + * Exported so a loop factored out of `run` can take it as a parameter and keep + * them: `function loop(messages: ModelMessage[], streamText: ChatStreamText)`. + */ +export type ChatStreamText = AiStreamTextFn; + +/** + * A `streamText` with the agent's managed options already applied. + * + * Handed to `run()` so the managed state cannot be missed by omission. Spreading + * `chat.toStreamTextOptions()` is still supported and equivalent; this exists + * because forgetting the spread silently drops the managed prompt, the skill + * tools, telemetry, and the `prepareStep` that delivers steering, compaction and + * conversational injection. + * + * Caller options win for everything the caller owns (model, messages, signal, + * stopWhen). The three that would otherwise clobber managed behaviour are + * merged instead of replaced: + * + * - `tools` are passed into the helper, so skill tools survive. + * - `prepareStep` is composed after the managed one, so a caller's per-step + * overrides apply on top of steering and compaction instead of disabling them. + * + * `system` may be set at the call site, on `chat.agent({ system })`, or + * through `chat.prompt.set()`, but only in one of them. Setting it in two + * places throws: no shape merges two system values on every supported + * version, and dropping one silently is the failure this seam exists to + * prevent. Injected instructions append to whichever one is in play. + */ +type ManagedStreamTextConfig = { + registry?: ToStreamTextOptionsOptions["registry"]; + system?: ToStreamTextOptionsOptions["system"]; + cacheControl?: ToStreamTextOptionsOptions["cacheControl"]; + systemProviderOptions?: ToStreamTextOptionsOptions["systemProviderOptions"]; +}; + +/** + * The caller's `streamText` options merged with the agent's managed ones. + * + * Pure, and separate from the call so it can be asserted directly: everything + * the caller did not name has to survive the merge, and the way to be sure of + * that is to look at the merged object rather than at what the model received. + */ +function buildManagedStreamTextOptions( + options: Record, + config: ManagedStreamTextConfig +): Record { + const { registry, system: agentSystem, cacheControl, systemProviderOptions } = config; + + /** + * Only the three keys that collide are intercepted. Everything else, telemetry + * included, stays in `rest` and reaches `streamText` untouched, with the + * caller's value winning because `rest` is spread after `managed`. Pulling a + * key out to "handle" it is how a caller's option gets silently dropped. + */ + const { + tools, + system: callerSystem, + prepareStep: callerPrepareStep, + ...rest + } = options as Record; + + const managed = toStreamTextOptions({ + registry, + system: (callerSystem as ToStreamTextOptionsOptions["system"]) ?? agentSystem, + cacheControl, + systemProviderOptions, + tools: tools as Record | undefined, + }); + + const promptSystem = locals.get(chatPromptKey)?.text; + const managedSystem = promptSystem || agentSystem; + if (callerSystem !== undefined && managedSystem) { + throw new Error( + "chat.agent: `system` is already set " + + (promptSystem ? "by chat.prompt.set()" : "on chat.agent({ system })") + + ", so it cannot also be passed to the `streamText` given to run(). Set it in one place, and add " + + "per-turn context with chat.inject({ role: 'system' }) rather than a second system value." + ); + } + + const managedPrepareStep = managed.prepareStep as ((arg: any) => Promise | any) | undefined; + if (typeof callerPrepareStep === "function") { + managed.prepareStep = async (arg: any) => { + const first = managedPrepareStep ? await managedPrepareStep(arg) : undefined; + const second = await callerPrepareStep({ ...arg, ...(first ?? {}) }); + return { ...(first ?? {}), ...(second ?? {}) }; + }; + } + + return { ...managed, ...rest }; +} + +/** @internal Test hook for {@link buildManagedStreamTextOptions}. */ +export const __buildManagedStreamTextOptionsForTests = buildManagedStreamTextOptions; + +function createBoundStreamText( + registry: ToStreamTextOptionsOptions["registry"] | undefined, + agentSystem: ToStreamTextOptionsOptions["system"] | undefined, + agentCacheControl: ToStreamTextOptionsOptions["cacheControl"] | undefined, + agentSystemProviderOptions: ToStreamTextOptionsOptions["systemProviderOptions"] | undefined +): AiStreamTextFn { + const bound = (options: Record = {}) => + aiStreamText( + buildManagedStreamTextOptions(options, { + registry, + system: agentSystem, + cacheControl: agentCacheControl, + systemProviderOptions: agentSystemProviderOptions, + }) as any + ); + + return bound as unknown as AiStreamTextFn; +} + function toStreamTextOptions(options?: ToStreamTextOptionsOptions): Record { const prompt = locals.get(chatPromptKey); const skills = locals.get(chatSkillsKey); const result: Record = {}; // Build the combined system prompt: stored prompt + skills preamble. - const promptText = prompt?.text ?? ""; + const baseSystem = options?.system; + const baseSystemText = + typeof baseSystem === "string" + ? baseSystem + : typeof baseSystem?.content === "string" + ? baseSystem.content + : ""; + const promptText = prompt?.text || baseSystemText; const skillsText = skills && skills.length > 0 ? buildSkillsSystemPrompt(skills) : ""; if (promptText || skillsText) { const systemText = [promptText, skillsText].filter(Boolean).join("\n\n"); @@ -5366,6 +5519,17 @@ export type ActionEvent< uiMessages: TUIM[]; /** The accumulated model messages (after hydration, if set). */ messages: ModelMessage[]; + /** + * `streamText` with the agent's managed options already applied, the same one + * `run()` receives: the prompt from `chat.prompt.set()` or + * `chat.agent({ system })`, the skill tools, the registry-resolved model and + * telemetry. + * + * Use it for a response produced from an action. A regenerate built with the + * `streamText` imported from `ai` answers without the agent's own system + * prompt, which is a behaviour difference nobody expects from a regenerate. + */ + streamText: ChatStreamText; }; /** @@ -6031,6 +6195,35 @@ export type ChatAgentOptions< * }); * ``` */ + /** + * A provider registry, so the runtime can resolve the managed prompt's model + * for the `streamText` it hands to `run()`. Only needed when you set a model + * through `chat.prompt.set()` and want the bound `streamText` to honour it. + */ + registry?: ToStreamTextOptionsOptions["registry"]; + + /** + * The agent's system prompt. Injected instructions append to it, and the + * `streamText` handed to `run()` carries it without a spread. + * + * Set it here, at the call site, or through `chat.prompt.set()` for versioned + * config, but only in one of them. + */ + system?: ToStreamTextOptionsOptions["system"]; + + /** + * Mark the system prompt for provider-side caching, the same sugar + * `chat.toStreamTextOptions({ cacheControl })` takes. See + * [prompt caching](/ai-chat/prompt-caching). + */ + cacheControl?: ToStreamTextOptionsOptions["cacheControl"]; + + /** + * Provider options for the system block, when `cacheControl` is not enough. + * Takes precedence over `cacheControl`. + */ + systemProviderOptions?: ToStreamTextOptionsOptions["systemProviderOptions"]; + compaction?: ChatAgentCompactionOptions; /** @@ -6528,6 +6721,10 @@ function chatAgent< onChatResume, exitAfterPreloadIdle = false, oomMachine, + registry: promptRegistry, + system: agentSystem, + cacheControl: agentCacheControl, + systemProviderOptions: agentSystemProviderOptions, ...restOptions } = options; @@ -7764,6 +7961,12 @@ function chatAgent< clientData, uiMessages: accumulatedUIMessages, messages: accumulatedMessages, + streamText: createBoundStreamText( + promptRegistry, + agentSystem, + agentCacheControl, + agentSystemProviderOptions + ), }); }, { @@ -8341,6 +8544,12 @@ function chatAgent< signal: combinedSignal, cancelSignal, stopSignal, + streamText: createBoundStreamText( + promptRegistry, + agentSystem, + agentCacheControl, + agentSystemProviderOptions + ), } as any); } diff --git a/packages/trigger-sdk/src/v3/chat-server.test.ts b/packages/trigger-sdk/src/v3/chat-server.test.ts index 539fe0247ad..a4e3e65a044 100644 --- a/packages/trigger-sdk/src/v3/chat-server.test.ts +++ b/packages/trigger-sdk/src/v3/chat-server.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; -import { simulateReadableStream, streamText } from "ai"; +import { simulateReadableStream, stepCountIs, streamText } from "ai"; import type { UIMessageChunk } from "ai"; import { MockLanguageModelV3 } from "ai/test"; import type { LanguageModelV3StreamPart } from "@ai-sdk/provider"; @@ -160,6 +160,97 @@ describe("chat.headStart (route handler)", () => { vi.restoreAllMocks(); }); + it("hands run() a streamText that carries the handover options without a spread", async () => { + global.fetch = vi.fn().mockImplementation(async (url: string | URL) => { + const urlStr = typeof url === "string" ? url : url.toString(); + if (urlStr.endsWith("/api/v1/sessions") || urlStr.endsWith("/api/v1/sessions/")) { + return createSessionResponse("chat-bound"); + } + if (urlStr.includes("/realtime/v1/sessions/") && urlStr.endsWith("/in/append")) { + return appendOkResponse(); + } + throw new Error(`Unexpected URL: ${urlStr}`); + }); + + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("hi back") }), + }); + + /** No `...chatHelper.toStreamTextOptions()` anywhere. */ + const handler = chat.headStart({ + agentId: "test-agent", + run: async ({ streamText: managedStreamText }) => managedStreamText({ model }), + }); + + const res = await withApiContext(() => + handler( + makeRequest({ + chatId: "chat-bound", + trigger: "submit-message", + headStartMessages: [{ id: "m1", role: "user", parts: [{ type: "text", text: "hi" }] }], + }) + ) + ); + + expect(res.status).toBe(200); + + // The handler pipes the result to session.out in the background; give it a tick. + await new Promise((r) => setTimeout(r, 200)); + + /** + * The caller passed only `model`, so `messages` reaching the provider is + * proof the managed options were applied: without them the prompt would be + * empty and `streamText` would have had nothing to send. + */ + const call = model.doStreamCalls.at(-1)!; + expect(JSON.stringify(call.prompt)).toContain("hi"); + }); + + it("refuses the handover options at the call site", async () => { + global.fetch = vi.fn().mockImplementation(async (url: string | URL) => { + const urlStr = typeof url === "string" ? url : url.toString(); + if (urlStr.endsWith("/api/v1/sessions") || urlStr.endsWith("/api/v1/sessions/")) { + return createSessionResponse("chat-owned"); + } + if (urlStr.includes("/realtime/v1/sessions/") && urlStr.endsWith("/in/append")) { + return appendOkResponse(); + } + throw new Error(`Unexpected URL: ${urlStr}`); + }); + + let thrown: unknown; + + const handler = chat.headStart({ + agentId: "test-agent", + run: async ({ streamText: managedStreamText }) => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("hi") }), + }); + try { + // `stopWhen` is what stops step 1 and hands over. Overriding it after a + // spread breaks the protocol silently; here it throws. + return managedStreamText({ model, stopWhen: stepCountIs(20) } as never); + } catch (error) { + thrown = error; + return managedStreamText({ model }); + } + }, + }); + + await withApiContext(() => + handler( + makeRequest({ + chatId: "chat-owned", + trigger: "submit-message", + headStartMessages: [{ id: "m1", role: "user", parts: [{ type: "text", text: "hi" }] }], + }) + ) + ); + + expect((thrown as Error)?.message).toContain("owns `stopWhen`"); + expect((thrown as Error)?.message).toContain("step 1"); + }); + it("creates the session with handover-prepare in basePayload and returns the session PAT in headers", async () => { const requests: CapturedRequest[] = []; global.fetch = vi.fn().mockImplementation(async (url: string | URL, init?: RequestInit) => { diff --git a/packages/trigger-sdk/src/v3/chat-server.ts b/packages/trigger-sdk/src/v3/chat-server.ts index 5e48e3b24b6..06121414c5d 100644 --- a/packages/trigger-sdk/src/v3/chat-server.ts +++ b/packages/trigger-sdk/src/v3/chat-server.ts @@ -68,8 +68,16 @@ import { convertToModelMessages, generateId as generateAssistantMessageId, stepCountIs, + streamText as aiStreamText, } from "../imports/ai-runtime.js"; -import type { FinishReason, ModelMessage, Tool, UIMessage, UIMessageChunk } from "ai"; +import type { + streamText as aiStreamTextSignature, + FinishReason, + ModelMessage, + Tool, + UIMessage, + UIMessageChunk, +} from "ai"; import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js"; // `StreamTextResult` is defined locally rather than imported from `ai`: its @@ -103,6 +111,53 @@ export type HeadStartStreamTextOptions = { abortSignal: AbortSignal; }; +/** + * The AI SDK's own `streamText` signature, borrowed rather than restated, so it + * resolves to whichever of `ai` v5/v6/v7 the caller installed. + */ +type AiStreamTextFn = typeof aiStreamTextSignature; + +/** + * The same signature minus the options the handover owns. + * + * Borrowed and narrowed rather than restated: `Omit` removes exactly the four + * keys `buildStreamTextOptions` supplies, so passing one is a compile error + * before it is a runtime throw. Taking `typeof streamText` unchanged would be + * worse than useless here, since it *requires* `messages` — the one key the + * caller must not set. + */ +type HeadStartStreamTextFn = ( + options: Omit[0], "messages" | "prompt" | "stopWhen" | "abortSignal"> +) => ReturnType; + +/** The keys `buildStreamTextOptions` owns. Overriding any of them breaks handover. */ +const HEAD_START_OWNED_OPTIONS = ["messages", "stopWhen", "abortSignal"] as const; + +function createBoundHeadStartStreamText( + build: (opts?: { tools?: Record }) => Record +): HeadStartStreamTextFn { + const bound = (options: Record = {}) => { + const { tools, ...rest } = options as Record; + + const owned = HEAD_START_OWNED_OPTIONS.filter((key) => key in rest); + if (owned.length > 0) { + throw new Error( + `chat.headStart: the \`streamText\` passed to run() owns ${owned + .map((key) => `\`${key}\``) + .join(", ")}, so it cannot be set at the call site. The handover protocol depends on ` + + "them: `messages` is the converted wire payload, `stopWhen: stepCountIs(1)` stops after " + + "step 1 so the agent run picks up tool execution, and `abortSignal` combines the request " + + "lifecycle with the idle timeout. Pass `model`, `system`, `providerOptions` and your own " + + "keys instead." + ); + } + + return aiStreamText({ ...build({ tools }), ...rest } as any); + }; + + return bound as unknown as HeadStartStreamTextFn; +} + export type HeadStartRunArgs> = { /** User messages parsed from the incoming request. */ messages: UIMessage[]; @@ -110,6 +165,19 @@ export type HeadStartRunArgs> = { signal: AbortSignal; /** Helper exposing `toStreamTextOptions(...)` and a session escape hatch. */ chat: HeadStartChatHelper; + /** + * `streamText` with the four options the handover protocol depends on already + * applied: the converted `messages`, your `tools`, `stopWhen: stepCountIs(1)` + * and the combined `abortSignal`. + * + * Prefer it over importing `streamText` from `ai`. Spreading + * `chat.toStreamTextOptions()` into the imported one is equivalent, but + * setting `messages`, `stopWhen` or `abortSignal` after the spread breaks the + * handover, and nothing catches that. Passing any of them here throws + * instead, and note that the three it owns are a type error, not just a + * runtime one. + */ + streamText: HeadStartStreamTextFn; }; export type HeadStartChatHelper> = { @@ -279,6 +347,7 @@ export const chat = { messages: session.uiMessages, signal: session.combinedSignal, chat: helper, + streamText: createBoundHeadStartStreamText(session.buildStreamTextOptions), }); return session.handle.handoverResponse(result); @@ -365,6 +434,7 @@ export const chat = { messages: session.uiMessages, signal: session.combinedSignal, chat: helper, + streamText: createBoundHeadStartStreamText(session.buildStreamTextOptions), }); } catch (err) { // The warm step never produced a result — tell the agent run to exit diff --git a/packages/trigger-sdk/test/bound-streamtext.test.ts b/packages/trigger-sdk/test/bound-streamtext.test.ts new file mode 100644 index 00000000000..8915a0f9ed7 --- /dev/null +++ b/packages/trigger-sdk/test/bound-streamtext.test.ts @@ -0,0 +1,476 @@ +// 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, __buildManagedStreamTextOptionsForTests as buildManaged } from "../src/v3/ai.js"; +import { chat as chatServer } from "../src/v3/chat-server.js"; +import { simulateReadableStream, stepCountIs, tool } 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: 5, noCache: 5, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 5, text: 5, reasoning: undefined }, + }, + }, + ], + }); +} + +function makeGate() { + let open!: () => void; + const promise = new Promise((r) => (open = r)); + return { promise, open }; +} + +/** A tool-call step, then a text step, so the steering drain has a boundary. */ +function twoStepModel(answer: string) { + let call = 0; + return new MockLanguageModelV3({ + doStream: async () => { + call++; + if (call === 1) { + return { + stream: simulateReadableStream({ + chunks: [ + { type: "tool-input-start", id: "c1", toolName: "gate" }, + { type: "tool-input-delta", id: "c1", delta: '{"q":"x"}' }, + { type: "tool-input-end", id: "c1" }, + { type: "tool-call", toolCallId: "c1", toolName: "gate", input: '{"q":"x"}' }, + { + type: "finish", + finishReason: { unified: "tool-calls", raw: "tool_calls" }, + usage: { + inputTokens: { + total: 5, + noCache: 5, + cacheRead: undefined, + cacheWrite: undefined, + }, + outputTokens: { total: 5, text: 5, reasoning: undefined }, + }, + }, + ], + chunkDelayInMs: 10, + }) as ReadableStream, + }; + } + return { stream: textStream(answer) }; + }, + }); +} + +async function waitFor(check: () => boolean, label: string, timeoutMs = 10_000) { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (check()) return; + await new Promise((r) => setTimeout(r, 20)); + } + throw new Error(`timeout waiting for ${label}`); +} + +describe("the streamText handed to run()", () => { + it("carries the managed prompt and an injection with no spread", async () => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }); + let injected = false; + + const agent = chat.agent({ + id: "bound-streamtext-managed", + 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; + chat.inject([{ role: "system", content: "SENTINEL-NO-SPREAD" }]); + }, + /** No `...chat.toStreamTextOptions()` anywhere. */ + run: async ({ messages, signal, streamText }) => + streamText({ model, messages, abortSignal: signal }), + }); + + const harness = mockChatAgent(agent, { chatId: "bound-streamtext-managed" }); + + 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)); + + const system = JSON.stringify( + model.doStreamCalls.at(-1)!.prompt.filter((m) => m.role === "system") + ); + expect(system).toContain("You are a helpful assistant."); + expect(system).toContain("SENTINEL-NO-SPREAD"); + } finally { + await harness.close(); + } + }); + + it("keeps the agent's tools when the caller passes its own", async () => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }); + + const agentTool = tool({ + description: "declared on the agent", + inputSchema: z.object({ a: z.string() }), + execute: async () => "a", + }); + const callerTool = tool({ + description: "passed at the call site", + inputSchema: z.object({ b: z.string() }), + execute: async () => "b", + }); + + const agent = chat.agent({ + id: "bound-streamtext-tools", + tools: { agentTool }, + run: async ({ messages, signal, streamText }) => + streamText({ model, messages, abortSignal: signal, tools: { callerTool } }), + }); + + const harness = mockChatAgent(agent, { chatId: "bound-streamtext-tools" }); + + try { + await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] }); + await new Promise((r) => setTimeout(r, 40)); + + const names = (model.doStreamCalls.at(-1)!.tools ?? []).map((t) => t.name).sort(); + expect(names).toContain("callerTool"); + } finally { + await harness.close(); + } + }); + + it("takes a system at the call site when nothing else set one", async () => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }); + + const agent = chat.agent({ + id: "bound-streamtext-caller-system", + run: async ({ messages, signal, streamText }) => + streamText({ model, messages, abortSignal: signal, system: "CALLER-SYSTEM" }), + }); + + const harness = mockChatAgent(agent, { chatId: "bound-streamtext-caller-system" }); + + try { + await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] }); + await new Promise((r) => setTimeout(r, 60)); + + const prompt = JSON.stringify(model.doStreamCalls.at(-1)!.prompt); + expect(prompt).toContain("CALLER-SYSTEM"); + } finally { + await harness.close(); + } + }); + + it("carries a system set on the agent, with injections appended to it", async () => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }); + let injected = false; + + const agent = chat.agent({ + id: "bound-streamtext-agent-system", + system: "AGENT-SYSTEM", + onTurnComplete: async () => { + if (injected) return; + injected = true; + chat.inject([{ role: "system", content: "APPENDED-INJECTION" }]); + }, + run: async ({ messages, signal, streamText }) => + streamText({ model, messages, abortSignal: signal }), + }); + + const harness = mockChatAgent(agent, { chatId: "bound-streamtext-agent-system" }); + + 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)); + + const system = JSON.stringify( + model.doStreamCalls.at(-1)!.prompt.filter((m) => m.role === "system") + ); + expect(system).toContain("AGENT-SYSTEM"); + expect(system).toContain("APPENDED-INJECTION"); + } finally { + await harness.close(); + } + }); + + it("refuses a call-site system when the agent already set one", async () => { + const model = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("ok") }), + }); + + let thrown: unknown; + + const agent = chat.agent({ + id: "bound-streamtext-system-conflict", + system: "AGENT-SYSTEM", + run: async ({ messages, signal, streamText }) => { + try { + return streamText({ model, messages, abortSignal: signal, system: "ALSO-MINE" }); + } catch (error) { + thrown = error; + return streamText({ model, messages, abortSignal: signal }); + } + }, + }); + + const harness = mockChatAgent(agent, { chatId: "bound-streamtext-system-conflict" }); + + try { + await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] }); + await new Promise((r) => setTimeout(r, 60)); + + expect((thrown as Error)?.message).toContain("already set"); + expect((thrown as Error)?.message).toContain("chat.agent({ system })"); + } finally { + await harness.close(); + } + }); + + it("runs a caller prepareStep without disabling the steering drain", async () => { + /** + * Spreading the helper and then passing your own `prepareStep` replaces the + * managed one, which silently turns off steering, compaction and + * conversational injection. Composed here, so both run. + */ + const toolGate = makeGate(); + let toolEntered = false; + let callerPrepareStepCalls = 0; + + const gateTool = tool({ + description: "blocks until the test opens it", + inputSchema: z.object({ q: z.string() }), + execute: async () => { + toolEntered = true; + await toolGate.promise; + return "ok"; + }, + }); + + const model = twoStepModel("ANSWER"); + + const agent = chat.agent({ + id: "bound-streamtext-preparestep", + pendingMessages: { shouldInject: () => true }, + run: async ({ messages, signal, streamText }) => + streamText({ + model, + messages, + abortSignal: signal, + tools: { gate: gateTool }, + stopWhen: stepCountIs(5), + prepareStep: async () => { + callerPrepareStepCalls++; + return {}; + }, + }), + }); + + const harness = mockChatAgent(agent, { chatId: "bound-streamtext-preparestep" }); + + try { + const first = harness.sendMessage({ + id: "u1", + role: "user", + parts: [{ type: "text", text: "what is the queue depth" }], + }); + await waitFor(() => toolEntered, "tool entered"); + + await harness.sendPendingMessage({ + id: "u2", + role: "user", + parts: [{ type: "text", text: "STEER-VIA-COMPOSED-PREPARESTEP" }], + } as never); + + toolGate.open(); + await first.catch(() => {}); + await new Promise((r) => setTimeout(r, 80)); + + // The caller's hook ran. + expect(callerPrepareStepCalls).toBeGreaterThan(0); + + // And the managed one still delivered the steer to the model. + /** + * The discriminator is WHICH call carries the steer, not whether any does. + * Composed, the managed drain injects it at the step boundary, so it lands + * in the second call of this turn and the turn ends. Replaced, the drain + * never runs, the message falls through to a turn of its own, and it shows + * up in a third call instead. Asserting "some prompt contains it" passes + * either way. + */ + const hasSteer = (i: number) => + JSON.stringify(model.doStreamCalls[i]?.prompt ?? null).includes( + "STEER-VIA-COMPOSED-PREPARESTEP" + ); + + expect(hasSteer(1)).toBe(true); + expect(model.doStreamCalls).toHaveLength(2); + } finally { + toolGate.open(); + await harness.close(); + } + }); +}); + +describe("what the managed streamText passes through", () => { + /** + * Asserted on the merged options rather than on what the model received, + * because most `streamText` options never reach `doStream` and so cannot be + * observed from the provider side. Only `tools`, `system` and `prepareStep` + * are intercepted; anything else the caller names has to survive untouched. + * Pulling a key out to "handle" it is how one gets silently dropped, which is + * what happened to `experimental_telemetry`. + */ + it("leaves every option it does not merge alone, telemetry included", () => { + const telemetry = { isEnabled: true, metadata: { requestId: "abc" } }; + const onStepFinish = () => {}; + + const merged = buildManaged( + { + model: "m", + messages: [], + experimental_telemetry: telemetry, + temperature: 0.3, + maxOutputTokens: 512, + onStepFinish, + providerOptions: { anthropic: { thinking: { type: "enabled" } } }, + }, + {} + ); + + expect(merged.experimental_telemetry).toBe(telemetry); + expect(merged.temperature).toBe(0.3); + expect(merged.maxOutputTokens).toBe(512); + expect(merged.onStepFinish).toBe(onStepFinish); + expect(merged.providerOptions).toEqual({ anthropic: { thinking: { type: "enabled" } } }); + }); + + it("lets a caller's telemetry win over the agent's", () => { + const callerTelemetry = { isEnabled: true, metadata: { source: "caller" } }; + const merged = buildManaged({ model: "m", experimental_telemetry: callerTelemetry }, {}); + expect(merged.experimental_telemetry).toBe(callerTelemetry); + }); + + it("carries the agent's system when the caller names none", () => { + const merged = buildManaged({ model: "m" }, { system: "AGENT-SYSTEM" }); + expect(JSON.stringify(merged.system)).toContain("AGENT-SYSTEM"); + }); +}); + +describe("what the types reject", () => { + /** + * Compile-time assertions, not runtime ones. `tsc` checks this file, so an + * unsatisfied `@ts-expect-error` fails the typecheck and a regression here is + * caught by CI rather than by a customer. The headStart options below are the + * ones the handover protocol owns: they throw at runtime too, but the type + * error is the guarantee worth pinning, and an autojudge leg cannot see it. + */ + it("rejects the handover-owned options at the call site", () => { + const _handler = chatServer.headStart({ + agentId: "a", + run: async ({ streamText }) => + // @ts-expect-error `stopWhen` is pinned to stepCountIs(1) by the handover + streamText({ model: {} as never, stopWhen: 1 as never }), + }); + + const _messages = chatServer.headStart({ + agentId: "a", + run: async ({ streamText }) => + // @ts-expect-error `messages` is the converted wire payload + streamText({ model: {} as never, messages: [] }), + }); + + const _signal = chatServer.headStart({ + agentId: "a", + run: async ({ streamText }) => + // @ts-expect-error `abortSignal` combines the request lifecycle and the idle timeout + streamText({ model: {} as never, abortSignal: new AbortController().signal }), + }); + + expect([_handler, _messages, _signal].every((h) => typeof h === "function")).toBe(true); + }); +}); + +describe("the streamText handed to onAction", () => { + it("carries the agent's system prompt into a regenerated answer", async () => { + /** + * A regenerate is the agent answering again, so it has to answer with the + * agent's own instructions. Built with the `streamText` imported from `ai` + * it answers with none, and nothing reports that: the reply looks fine and + * is simply produced by a differently-configured model than every other + * turn. + */ + const turnModel = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("first answer") }), + }); + const actionModel = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("regenerated answer") }), + }); + + const agent = chat.agent({ + id: "bound-streamtext-onaction", + system: "AGENT-SYSTEM-FOR-ACTIONS", + actionSchema: z.discriminatedUnion("type", [z.object({ type: z.literal("regenerate") })]), + onAction: async ({ action, messages, streamText }) => { + if (action.type !== "regenerate") return; + chat.history.slice(0, -1); + return streamText({ model: actionModel, messages }); + }, + run: async ({ messages, signal, streamText }) => + streamText({ model: turnModel, messages, abortSignal: signal }), + }); + + const harness = mockChatAgent(agent, { chatId: "bound-streamtext-onaction" }); + + try { + await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "ask" }] }); + await new Promise((r) => setTimeout(r, 40)); + + // The turn carries it, which is the baseline. + expect(JSON.stringify(turnModel.doStreamCalls.at(-1)!.prompt)).toContain( + "AGENT-SYSTEM-FOR-ACTIONS" + ); + + await harness.sendAction({ type: "regenerate" }); + await new Promise((r) => setTimeout(r, 60)); + + // And so does the action's own stream. + expect(actionModel.doStreamCalls).toHaveLength(1); + expect(JSON.stringify(actionModel.doStreamCalls[0]!.prompt)).toContain( + "AGENT-SYSTEM-FOR-ACTIONS" + ); + } finally { + await harness.close(); + } + }); +});