diff --git a/.changeset/chat-actions-no-turn.md b/.changeset/chat-actions-no-turn.md new file mode 100644 index 000000000..a01134415 --- /dev/null +++ b/.changeset/chat-actions-no-turn.md @@ -0,0 +1,33 @@ +--- +"@trigger.dev/sdk": minor +--- + +`chat.agent` actions are no longer treated as turns. They fire `hydrateMessages` and `onAction` only — no `onTurnStart` / `prepareMessages` / `onBeforeTurnComplete` / `onTurnComplete`, no `run()`, no turn-counter increment. The trace span is named `chat action` instead of `chat turn N`. + +`onAction` can now return a `StreamTextResult`, `string`, or `UIMessage` to produce a model response from the action; returning `void` (the previous and now default) is side-effect-only. + +**Migration**: if you previously had `run()` branching on `payload.trigger === "action"`, return your `streamText(...)` from `onAction` instead. If you persisted in `onTurnComplete`, do that work inside `onAction`. For any other state-only action, just remove your skip-the-model workaround — the default is now correct. + +```ts +// before +onAction: async ({ action }) => { + if (action.type === "regenerate") { + chat.store.set({ skipModelCall: false }); + chat.history.slice(0, -1); + } +}, +run: async ({ messages, signal }) => { + if (chat.store.get()?.skipModelCall) return; + return streamText({ model, messages, abortSignal: signal }); +}, + +// after +onAction: async ({ action, messages, signal }) => { + if (action.type === "regenerate") { + chat.history.slice(0, -1); + return streamText({ model, messages, abortSignal: signal }); + } +}, +run: async ({ messages, signal }) => + streamText({ model, messages, abortSignal: signal }), +``` diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index fb9514ac9..88f3a6afe 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -2679,6 +2679,17 @@ function isUIMessageStreamable(value: unknown): value is UIMessageStreamable { ); } +let warnedMissingOnAction = false; +function warnMissingOnActionOnce() { + if (warnedMissingOnAction) return; + warnedMissingOnAction = true; + console.warn( + "[chat.agent] Received an action but no `onAction` handler is configured. " + + "The action is being ignored. Define `onAction` (and optionally `actionSchema`) on " + + "your agent to handle it." + ); +} + function isAsyncIterable(value: unknown): value is AsyncIterable { return typeof value === "object" && value !== null && Symbol.asyncIterator in value; } @@ -3174,9 +3185,14 @@ export type ChatAgentOptions< /** * Called when the frontend sends a custom action via `transport.sendAction()`. * - * Fires after message hydration (if set) but before `onTurnStart` and `run()`. - * Use `chat.history.*` to modify the conversation state — the LLM will respond - * to the modified state. + * Actions are not turns. They fire `hydrateMessages` (if configured) and + * `onAction` only — no `onTurnStart` / `prepareMessages` / + * `onBeforeTurnComplete` / `onTurnComplete`, no `run()`. Use + * `chat.history.*` inside `onAction` to mutate state. + * + * To produce a model response from an action, return a + * `StreamTextResult` (auto-piped), `string`, or `UIMessage`. Returning + * `void` or nothing is the side-effect-only default. */ onAction?: ( event: ActionEvent< @@ -3184,7 +3200,7 @@ export type ChatAgentOptions< inferSchemaOut, TUIMessage > - ) => Promise | void; + ) => Promise | unknown; /** * The run function for the chat task. @@ -4072,13 +4088,20 @@ function chatAgent< ) as inferSchemaOut; const lastUserMessage = extractLastUserMessageText(uiMessages); + // Actions are not turns. They use a different span name + // and don't carry a turn.number. Branched on at `isAction`. + const isAction = currentWirePayload.trigger === "action"; + const spanName = isAction ? "chat action" : `chat turn ${turn + 1}`; + const turnAttributes: Attributes = { - "turn.number": turn + 1, + ...(isAction ? {} : { "turn.number": turn + 1 }), "gen_ai.conversation.id": currentWirePayload.chatId, "gen_ai.operation.name": "chat", "chat.trigger": currentWirePayload.trigger, - [SemanticInternalAttributes.STYLE_ICON]: "tabler-message-chatbot", - [SemanticInternalAttributes.ENTITY_TYPE]: "chat-turn", + [SemanticInternalAttributes.STYLE_ICON]: isAction + ? "tabler-bolt" + : "tabler-message-chatbot", + [SemanticInternalAttributes.ENTITY_TYPE]: isAction ? "chat-action" : "chat-turn", }; if (lastUserMessage) { @@ -4102,7 +4125,7 @@ function chatAgent< } const turnResult = await tracer.startActiveSpan( - `chat turn ${turn + 1}`, + spanName, async (turnSpan) => { // (errors are caught by the outer try/catch which writes an error chunk) locals.set(chatPipeCountKey, 0); @@ -4268,10 +4291,17 @@ function chatAgent< const turnNewUIMessages: TUIMessage[] = []; // ── Action handling ────────────────────────────────────── - // Actions arrive via the same chat-messages stream but with - // trigger === "action". They wake the agent, modify state - // (via onAction + chat.history), then fall through to run(). - if (currentWirePayload.trigger === "action") { + // Actions arrive on the same input stream but with + // trigger === "action". They are NOT turns — only + // `hydrateMessages` and `onAction` fire. No turn lifecycle + // hooks (`onTurnStart` / `prepareMessages` / + // `onBeforeTurnComplete` / `onTurnComplete`) and no + // `run()` invocation. To produce a model response from + // an action, return a `StreamTextResult` (auto-piped), + // string, or UIMessage from `onAction`. Turn counter + // does not advance. + let actionStreamResult: unknown = undefined; + if (isAction) { // Parse and validate the action payload const parsedAction = parseAction ? await parseAction(currentWirePayload.action) @@ -4298,7 +4328,6 @@ function chatAgent< [SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart", [SemanticInternalAttributes.COLLAPSED]: true, "chat.id": currentWirePayload.chatId, - "chat.turn": turn + 1, "chat.trigger": "action", }, } @@ -4308,12 +4337,13 @@ function chatAgent< locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages); } - // Fire onAction — handler uses chat.history.* to modify state + // Fire onAction — handler may mutate state via + // `chat.history.*` and / or return a model response. if (onAction) { - await tracer.startActiveSpan( + actionStreamResult = await tracer.startActiveSpan( "onAction()", async () => { - await onAction({ + return await onAction({ action: parsedAction as any, chatId: currentWirePayload.chatId, turn, @@ -4327,10 +4357,10 @@ function chatAgent< [SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart", [SemanticInternalAttributes.COLLAPSED]: true, "chat.id": currentWirePayload.chatId, - "chat.turn": turn + 1, - "chat.action": typeof parsedAction === "object" && parsedAction !== null - ? JSON.stringify(parsedAction) - : String(parsedAction), + "chat.action": + typeof parsedAction === "object" && parsedAction !== null + ? JSON.stringify(parsedAction) + : String(parsedAction), }, } ); @@ -4343,6 +4373,8 @@ function chatAgent< accumulatedMessages = await toModelMessages(actionOverride); locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages); } + } else { + warnMissingOnActionOnce(); } } @@ -4537,6 +4569,54 @@ function chatAgent< } // end if (trigger !== "action") + // ── Action result handling ────────────────────────────── + // For action turns, skip the turn machinery entirely. + // If `onAction` returned a stream / string / UIMessage, + // pipe it as the response. Either way, emit + // `trigger:turn-complete` and then fall through to the + // wait-for-next-message logic (shared with message turns). + // The turn counter is decremented so the next iteration + // sees the same `turn` value — actions don't count. + if (isAction) { + msgSub.off(); + + if ( + (locals.get(chatPipeCountKey) ?? 0) === 0 && + isUIMessageStreamable(actionStreamResult) + ) { + try { + const resolvedOptions = resolveUIMessageStreamOptions(); + const uiStream = ( + actionStreamResult as UIMessageStreamable + ).toUIMessageStream({ + ...resolvedOptions, + generateMessageId: + resolvedOptions.generateMessageId ?? generateMessageId, + }); + await pipeChat(uiStream, { + signal: combinedSignal, + spanName: "stream response", + }); + } catch (error) { + if ( + error instanceof Error && + error.name === "AbortError" && + runSignal.aborted + ) { + return "exit"; + } + throw error; + } + } + + await writeTurnCompleteChunk(currentWirePayload.chatId); + + // Don't consume a turn iteration — actions aren't turns. + turn--; + } + + if (!isAction) { + // Mint a scoped public access token once per turn, reused for // onChatStart, onTurnStart, onTurnComplete, and the turn-complete chunk. const currentRunId = ctx.run.id; @@ -5235,6 +5315,8 @@ function chatAgent< ); } + } // end if (!isAction) + // NOTE: We intentionally do NOT await deferred work from onTurnComplete here. // Promises deferred in onTurnComplete (e.g. background self-review via // chat.defer + chat.inject) run during the idle wait. If they complete diff --git a/packages/trigger-sdk/test/mockChatAgent.test.ts b/packages/trigger-sdk/test/mockChatAgent.test.ts index adcbce939..ac72e76ec 100644 --- a/packages/trigger-sdk/test/mockChatAgent.test.ts +++ b/packages/trigger-sdk/test/mockChatAgent.test.ts @@ -210,6 +210,174 @@ describe("mockChatAgent", () => { } }); + it("actions returning void do not fire turn hooks or call run()", async () => { + const onChatStart = vi.fn(); + const onTurnStart = vi.fn(); + const onBeforeTurnComplete = vi.fn(); + const onTurnComplete = vi.fn(); + const onAction = vi.fn(); + const runSpy = vi.fn(); + const model = new MockLanguageModelV3({ + doStream: async () => { + runSpy(); + return { stream: textStream("nope") }; + }, + }); + + const { z } = await import("zod"); + const agent = chat.agent({ + id: "mockChatAgent.actions.void", + actionSchema: z.object({ type: z.literal("undo") }), + onChatStart, + onTurnStart, + onBeforeTurnComplete, + onTurnComplete, + onAction: async (...args) => { + onAction(...args); + // void → side-effect only + }, + run: async ({ messages, signal }) => { + return streamText({ model, messages, abortSignal: signal }); + }, + }); + + const harness = mockChatAgent(agent, { chatId: "test-void-action" }); + try { + // Bootstrap with a message so the message-turn hooks fire once. + await harness.sendMessage(userMessage("hi")); + // sendMessage resolves on `trigger:turn-complete`, but onTurnComplete + // fires as a separate microtask after — let it settle before snapshotting. + await new Promise((r) => setTimeout(r, 50)); + + // Snapshot call counts after the bootstrap — we'll assert these + // don't change for the action below. + const baselineRun = runSpy.mock.calls.length; + const baselineChatStart = onChatStart.mock.calls.length; + const baselineTurnStart = onTurnStart.mock.calls.length; + const baselineBeforeComplete = onBeforeTurnComplete.mock.calls.length; + const baselineComplete = onTurnComplete.mock.calls.length; + + const actionTurn = await harness.sendAction({ type: "undo" }); + await new Promise((r) => setTimeout(r, 50)); + + // onAction fired exactly once; no turn hooks fired; run() / LLM did not. + expect(onAction).toHaveBeenCalledTimes(1); + expect(runSpy.mock.calls.length).toBe(baselineRun); + expect(onChatStart.mock.calls.length).toBe(baselineChatStart); + expect(onTurnStart.mock.calls.length).toBe(baselineTurnStart); + expect(onBeforeTurnComplete.mock.calls.length).toBe(baselineBeforeComplete); + expect(onTurnComplete.mock.calls.length).toBe(baselineComplete); + + // Stream still terminates cleanly with trigger:turn-complete so + // the frontend's useChat transitions back to ready. + const sawTurnComplete = actionTurn.rawChunks.some( + (c) => + typeof c === "object" && + c !== null && + (c as { type?: string }).type === "trigger:turn-complete" + ); + expect(sawTurnComplete).toBe(true); + } finally { + await harness.close(); + } + }); + + it("actions returning a stream pipe the response without firing turn hooks", async () => { + const onTurnStart = vi.fn(); + const onTurnComplete = vi.fn(); + const actionModel = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("regenerated") }), + }); + const turnModel = new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("normal-response") }), + }); + + const { z } = await import("zod"); + const agent = chat.agent({ + id: "mockChatAgent.actions.stream", + actionSchema: z.object({ type: z.literal("regenerate") }), + onTurnStart, + onTurnComplete, + onAction: async ({ messages }) => { + return streamText({ model: actionModel, messages }); + }, + run: async ({ messages, signal }) => { + return streamText({ model: turnModel, messages, abortSignal: signal }); + }, + }); + + const harness = mockChatAgent(agent, { chatId: "test-stream-action" }); + try { + await harness.sendMessage(userMessage("hi")); + await new Promise((r) => setTimeout(r, 50)); + const baselineTurnStart = onTurnStart.mock.calls.length; + const baselineTurnComplete = onTurnComplete.mock.calls.length; + + const actionTurn = await harness.sendAction({ type: "regenerate" }); + await new Promise((r) => setTimeout(r, 50)); + + // No turn hooks fired during the action. + expect(onTurnStart.mock.calls.length).toBe(baselineTurnStart); + expect(onTurnComplete.mock.calls.length).toBe(baselineTurnComplete); + + // Action's streamText output landed on the response. + const text = actionTurn.chunks + .filter((c) => c.type === "text-delta") + .map((c) => (c as { delta: string }).delta) + .join(""); + expect(text).toBe("regenerated"); + } finally { + await harness.close(); + } + }); + + it("warns once and emits turn-complete when an action arrives without onAction", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + const runSpy = vi.fn(); + const model = new MockLanguageModelV3({ + doStream: async () => { + runSpy(); + return { stream: textStream("nope") }; + }, + }); + + const { z } = await import("zod"); + const agent = chat.agent({ + id: "mockChatAgent.actions.no-handler", + actionSchema: z.object({ type: z.literal("undo") }), + run: async ({ messages, signal }) => { + return streamText({ model, messages, abortSignal: signal }); + }, + }); + + const harness = mockChatAgent(agent, { chatId: "test-no-handler" }); + try { + await harness.sendMessage(userMessage("hi")); + const baselineRun = runSpy.mock.calls.length; + + const actionTurn = await harness.sendAction({ type: "undo" }); + + // No additional model call; console.warn fired with our marker text. + expect(runSpy.mock.calls.length).toBe(baselineRun); + expect( + warnSpy.mock.calls.some((args) => + (args[0] as string).includes("no `onAction` handler") + ) + ).toBe(true); + + const sawTurnComplete = actionTurn.rawChunks.some( + (c) => + typeof c === "object" && + c !== null && + (c as { type?: string }).type === "trigger:turn-complete" + ); + expect(sawTurnComplete).toBe(true); + } finally { + await harness.close(); + warnSpy.mockRestore(); + } + }); + it("passes clientData through to run() and hooks", async () => { const model = new MockLanguageModelV3({ doStream: async () => ({ stream: textStream("ok") }),