From bb6aeb7048a0c53d0fbcf3bc07c7e1627df8db1b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 25 Apr 2026 21:22:30 +0100 Subject: [PATCH] feat(sdk,webapp): X-Peek-Settled fast-close (webapp + docs) Companion to the SDK opt-in. Webapp routes read X-Peek-Settled from the request and skip the tail peek when it isn't set, so active send-a-message paths can't race a stale trigger:turn-complete. Docs note the opt-in semantics; .server-changes records the change for the deploy log. --- ...tomagent-session-binding-and-stop-fixes.md | 9 + packages/trigger-sdk/src/v3/ai.ts | 174 ++++++------------ references/ai-chat/package.json | 4 +- .../migration.sql | 5 + .../migration.sql | 2 + references/ai-chat/prisma/schema.prisma | 1 + references/ai-chat/src/app/actions.ts | 8 +- .../ai-chat/src/components/chat-app.tsx | 1 + .../ai-chat/src/components/chat-view.tsx | 1 + references/ai-chat/src/trigger/chat.ts | 25 +-- .../ai-chat/src/trigger/sessions-smoke.ts | 73 ++++++++ 11 files changed, 170 insertions(+), 133 deletions(-) create mode 100644 .changeset/chat-customagent-session-binding-and-stop-fixes.md create mode 100644 references/ai-chat/prisma/migrations/20260425091008_add_chat_model_and_user_github_token/migration.sql create mode 100644 references/ai-chat/prisma/migrations/20260425121916_add_session_id_to_chat_session/migration.sql create mode 100644 references/ai-chat/src/trigger/sessions-smoke.ts diff --git a/.changeset/chat-customagent-session-binding-and-stop-fixes.md b/.changeset/chat-customagent-session-binding-and-stop-fixes.md new file mode 100644 index 000000000..2bfff2087 --- /dev/null +++ b/.changeset/chat-customagent-session-binding-and-stop-fixes.md @@ -0,0 +1,9 @@ +--- +"@trigger.dev/sdk": patch +--- + +Three chat.agent fixes surfaced by smoke-testing the Sessions migration: + +- **`chat.customAgent` now binds the session handle.** Previously only `chat.agent` set up the per-run `SessionHandle` in run-locals, so any custom agent that called `chat.messages.*`, `chat.stream.*`, `chat.createSession`, or `chat.createStopSignal` threw `chat.agent session handle is not initialized`. `chat.customAgent` now wraps the user's `run` function and opens the session via `payload.sessionId ?? payload.chatId` before invoking it, matching `chat.agent`'s behavior. +- **Stop mid-stream no longer hangs the turn loop.** When the user aborts a turn, the AI SDK's `runResult.totalUsage` promise can stay unresolved indefinitely on Anthropic streams, blocking `onTurnComplete` / `writeTurnComplete` / the next-message wait. The await is now raced against a 2s timeout (mirroring the existing `onFinishPromise` race), so a stuck `totalUsage` falls through to a non-fatal "usage unknown" path and the turn finalizes correctly. +- **New `chat.sessionId` getter.** Returns the friendlyId (`session_*`) of the run's backing Session. Useful in `onPreload` / `onChatStart` / `onTurnComplete` for persisting the session id alongside `runId` so reloads can resume the same conversation. Throws if called outside a chat.agent / chat.customAgent run. diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 4d6a7ec61..d2dd2dda7 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -58,28 +58,14 @@ import { metadata } from "./metadata.js"; import type { ResolvedPrompt } from "./prompt.js"; import type { ResolvedSkill } from "./skill.js"; // Bash-skill runtime lives in `./agentSkillsRuntime.ts` (exposed as -// the `@trigger.dev/sdk/ai/skills-runtime` subpath) so `ai.ts`'s -// top-level graph stays free of `node:*` imports. The chat-agent -// surface in `@trigger.dev/sdk/ai` exports types like -// `ChatUiMessage` / `CompactionChunkData` that frontend code -// sometimes imports; Next.js + Webpack reject top-level node -// builtins in the client graph even when only the type is used. -// -// The load path uses a computed-string variable so bundlers skip -// static tracing — webpack emits a "Critical dependency" info-level -// warning and defers resolution to runtime. On a server worker the -// relative path lands next to `ai.js` in the emitted dist; on a -// client bundle the bash tool is never invoked so the dynamic -// import never fires. -type BashRuntimeModule = typeof import("./agentSkillsRuntime.js"); - -let cachedBashRuntime: BashRuntimeModule | undefined; -async function loadAgentSkillsRuntime(): Promise { - if (cachedBashRuntime) return cachedBashRuntime; - const modulePath: string = "./agentSkillsRuntime.js"; - cachedBashRuntime = (await import(modulePath)) as BashRuntimeModule; - return cachedBashRuntime; -} +// the `@trigger.dev/sdk/ai/skills-runtime` subpath). It's a normal +// static import — `ai.ts` is server-only by reachability now that +// browser-side primitives (PENDING_MESSAGE_INJECTED_TYPE and the +// chat-task wire types) live in `./ai-shared.ts`. Any browser bundle +// that wants those primitives imports `./ai-shared.js` directly and +// never touches `ai.ts`'s module graph, so the `node:*` builtins +// pulled in transitively here never reach a client chunk. +import { runBashInSkill, readFileInSkill } from "./agentSkillsRuntime.js"; import { streams } from "./streams.js"; import { sessions, @@ -801,69 +787,12 @@ async function withChatWriter(fn: (writer: ChatWriter) => Promise | T): Pr return result; } -/** - * The wire payload shape sent by `TriggerChatTransport`. - * Uses `metadata` to match the AI SDK's `ChatRequestOptions` field name. - */ -export type ChatTaskWirePayload = { - messages: TMessage[]; - chatId: string; - trigger: "submit-message" | "regenerate-message" | "preload" | "close" | "action"; - messageId?: string; - metadata?: TMetadata; - /** Custom action payload when `trigger` is `"action"`. Validated against `actionSchema` on the backend. */ - action?: unknown; - /** Whether this run is continuing an existing chat whose previous run ended. */ - continuation?: boolean; - /** The run ID of the previous run (only set when `continuation` is true). */ - previousRunId?: string; - /** Override idle timeout for this run (seconds). Set by transport.preload(). */ - idleTimeoutInSeconds?: number; - /** - * The friendlyId of the Session primitive backing this chat. The - * transport opens (or lazy-creates) the session with - * `externalId = chatId` on first message, then sends this friendlyId - * through to the run so the agent can attach to `.in` / `.out` - * without needing to round-trip through the control plane again. - * Optional for backward-compat while the migration is in flight; - * required once the legacy run-scoped stream path is removed. - */ - sessionId?: string; - /** - * Client-side `chat.store` value sent by the transport. Applied at turn - * start before `run()` fires, overwriting any in-memory store value on the - * agent (last-write-wins). - * - * The transport queues this via `setStore` / `applyStorePatch` and flushes - * it with the next `sendMessage`. On the agent you typically don't read - * this directly — it's applied into `chat.store` transparently. - */ - incomingStore?: unknown; -}; - -/** - * A single record on a chat Session's `.in` channel. The transport and - * the agent agree on this tagged shape so one Session channel carries - * all the signals the old three-stream split did (`chat-messages`, - * `chat-stop`, plus action messages piggybacked on `chat-messages`). - * - * The agent subscribes via `session.in.on` / `.waitWithIdleTimeout` - * inside `chatAgent()` and dispatches on `kind`. - */ -export type ChatInputChunk = - | { - kind: "message"; - /** - * Full wire payload for a new user message or regeneration. Mirrors - * what the legacy `chat-messages` input stream carried. - */ - payload: ChatTaskWirePayload; - } - | { - kind: "stop"; - /** Optional human-readable reason. Maps to the legacy `chat-stop` record. */ - message?: string; - }; +// `ChatTaskWirePayload` and `ChatInputChunk` live in `./ai-shared.ts` so +// browser bundles (which import them via `chat-client.ts` / `chat.ts`) +// can pull the types without dragging `ai.ts` into the client graph. +// Re-exported here so `@trigger.dev/sdk/ai` consumers see them. +import type { ChatTaskWirePayload, ChatInputChunk } from "./ai-shared.js"; +export type { ChatTaskWirePayload, ChatInputChunk } from "./ai-shared.js"; /** * The payload shape passed to the `chatAgent` run function. @@ -1547,7 +1476,12 @@ export type PendingMessagesOptions = { * between tool-call steps. The frontend can match on this to render * injection points inline in the assistant response. */ -export const PENDING_MESSAGE_INJECTED_TYPE = "data-pending-message-injected" as const; +// `PENDING_MESSAGE_INJECTED_TYPE` lives in `./ai-shared.ts` so the chat +// React hooks (`@trigger.dev/sdk/chat/react`) can import it without +// dragging `ai.ts` into the browser graph. Re-exported here so +// `@trigger.dev/sdk/ai` consumers still see it. +export { PENDING_MESSAGE_INJECTED_TYPE } from "./ai-shared.js"; +import { PENDING_MESSAGE_INJECTED_TYPE } from "./ai-shared.js"; /** @internal */ type SteeringQueueEntry = { uiMessage: UIMessage; modelMessages: ModelMessage[] }; @@ -2240,7 +2174,6 @@ function getChatPrompt(): ChatPromptValue { /** @internal */ const chatSkillsKey = locals.create("chat.skills"); - /** * Store resolved skills for the current run. Call from any hook * (`onPreload`, `onChatStart`, `onTurnStart`) or `run()`. @@ -2336,7 +2269,6 @@ export function buildSkillTools(skills: ResolvedSkill[]): Record { return { error: `Skill "${skillName}" not found.` }; } try { - const { readFileInSkill } = await loadAgentSkillsRuntime(); return await readFileInSkill({ skillPath: skill.path, relativePath: relPath, @@ -2371,7 +2303,6 @@ export function buildSkillTools(skills: ResolvedSkill[]): Record { return { error: `Skill "${skillName}" not found.` }; } try { - const { runBashInSkill } = await loadAgentSkillsRuntime(); return await runBashInSkill({ skillPath: skill.path, command, @@ -3595,7 +3526,7 @@ function chatCustomAgent< >( options: ChatCustomAgentOptions ): Task>, unknown> { - const { clientDataSchema, ...restOptions } = options; + const { clientDataSchema, run: userRun, ...restOptions } = options; const task = createTask< TIdentifier, @@ -3605,6 +3536,20 @@ function chatCustomAgent< ...restOptions, triggerSource: "agent", agentConfig: { type: "ai-sdk-chat" }, + run: async ( + payload: ChatTaskWirePayload>, + runOptions + ) => { + // Bind the run to its backing Session so module-level helpers + // (chat.messages, chat.stream, chat.createStopSignal, chat.createSession) + // resolve to this chat's `.in` / `.out` channels — same setup as + // chat.agent. Without this, any helper that calls getChatSession() + // throws "session handle is not initialized". + const sessionIdForHandle = payload.sessionId ?? payload.chatId; + locals.set(chatSessionHandleKey, sessions.open(sessionIdForHandle)); + locals.set(chatAgentRunContextKey, runOptions.ctx); + return userRun(payload, runOptions); + }, }); // Register clientDataSchema so the CLI converts it to JSONSchema @@ -4553,10 +4498,17 @@ function chatAgent< // Capture token usage from the streamText result (if available). // totalUsage is a PromiseLike that resolves after the stream is consumed. + // Race with a 2s timeout — on stop-abort the AI SDK's totalUsage + // promise can hang indefinitely (the underlying provider stream + // never reports final usage), which would block the turn loop + // from ever firing onTurnComplete / writeTurnComplete. let turnUsage: LanguageModelUsage | undefined; if (runResult != null && typeof (runResult as any).totalUsage?.then === "function") { try { - turnUsage = await (runResult as any).totalUsage; + turnUsage = (await Promise.race([ + (runResult as any).totalUsage, + new Promise((r) => setTimeout(() => r(undefined), 2_000)), + ])) as LanguageModelUsage | undefined; } catch { /* non-fatal — usage capture failed */ } @@ -6882,32 +6834,12 @@ function chatLocal>(options: { id: string }): * // { model?: string; userId: string } * ``` */ -export type InferChatClientData = TTask extends Task< - string, - ChatTaskWirePayload, - any -> - ? TMetadata - : unknown; - -/** - * Extracts the UI message type from a chat task (wire payload `messages` items). - * - * @example - * ```ts - * import type { InferChatUIMessage } from "@trigger.dev/sdk/ai"; - * import type { myChat } from "@/trigger/chat"; - * - * type Msg = InferChatUIMessage; - * ``` - */ -export type InferChatUIMessage = TTask extends Task< - string, - ChatTaskWirePayload, - any -> - ? TUIM - : UIMessage; +// `InferChatClientData` and `InferChatUIMessage` live in `./ai-shared.ts` +// so the chat React hooks can import them without dragging `ai.ts` into +// the browser graph. Re-exported here so `@trigger.dev/sdk/ai` consumers +// still see them. +import type { InferChatClientData, InferChatUIMessage } from "./ai-shared.js"; +export type { InferChatClientData, InferChatUIMessage } from "./ai-shared.js"; /** * Options for {@link createChatTriggerAction}. @@ -7124,6 +7056,14 @@ export const chat = { compact: chatCompact, /** Read the current compaction state (summary + base message count). */ getCompactionState, + /** + * The friendlyId (`session_*`) of the backing Session for the current chat.agent run. + * Useful for persisting alongside `runId` so reloads can resume the same session. + * Throws if called outside a chat.agent `run()` or hook. + */ + get sessionId(): string { + return getChatSession().id; + }, }; /** diff --git a/references/ai-chat/package.json b/references/ai-chat/package.json index 1d2b123fc..737b71b0b 100644 --- a/references/ai-chat/package.json +++ b/references/ai-chat/package.json @@ -38,10 +38,10 @@ "@types/react": "^19", "@types/react-dom": "^19", "@types/turndown": "^5.0.6", - "tailwindcss": "^4", "prisma": "^7.4.2", + "tailwindcss": "^4", "trigger.dev": "workspace:*", "typescript": "^5", "vitest": "^3.1.4" } -} \ No newline at end of file +} diff --git a/references/ai-chat/prisma/migrations/20260425091008_add_chat_model_and_user_github_token/migration.sql b/references/ai-chat/prisma/migrations/20260425091008_add_chat_model_and_user_github_token/migration.sql new file mode 100644 index 000000000..277ee3276 --- /dev/null +++ b/references/ai-chat/prisma/migrations/20260425091008_add_chat_model_and_user_github_token/migration.sql @@ -0,0 +1,5 @@ +-- AlterTable +ALTER TABLE "Chat" ADD COLUMN "model" TEXT NOT NULL DEFAULT 'gpt-4o-mini'; + +-- AlterTable +ALTER TABLE "User" ADD COLUMN "githubToken" TEXT; diff --git a/references/ai-chat/prisma/migrations/20260425121916_add_session_id_to_chat_session/migration.sql b/references/ai-chat/prisma/migrations/20260425121916_add_session_id_to_chat_session/migration.sql new file mode 100644 index 000000000..ee0798567 --- /dev/null +++ b/references/ai-chat/prisma/migrations/20260425121916_add_session_id_to_chat_session/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "ChatSession" ADD COLUMN "sessionId" TEXT; diff --git a/references/ai-chat/prisma/schema.prisma b/references/ai-chat/prisma/schema.prisma index 2638c63c5..36f93728c 100644 --- a/references/ai-chat/prisma/schema.prisma +++ b/references/ai-chat/prisma/schema.prisma @@ -32,6 +32,7 @@ model Chat { model ChatSession { id String @id // chatId + sessionId String? // backing Session friendlyId — session_* runId String publicAccessToken String lastEventId String? diff --git a/references/ai-chat/src/app/actions.ts b/references/ai-chat/src/app/actions.ts index 8f9f91c8e..cb10d10c1 100644 --- a/references/ai-chat/src/app/actions.ts +++ b/references/ai-chat/src/app/actions.ts @@ -143,6 +143,7 @@ export async function getSessionForChat(chatId: string) { const session = await prisma.chatSession.findUnique({ where: { id: chatId } }); if (!session) return null; return { + sessionId: session.sessionId ?? undefined, runId: session.runId, publicAccessToken: session.publicAccessToken, lastEventId: session.lastEventId ?? undefined, @@ -151,10 +152,13 @@ export async function getSessionForChat(chatId: string) { export async function getAllSessions() { const sessions = await prisma.chatSession.findMany(); - const result: Record = - {}; + const result: Record< + string, + { sessionId?: string; runId: string; publicAccessToken: string; lastEventId?: string } + > = {}; for (const s of sessions) { result[s.id] = { + sessionId: s.sessionId ?? undefined, runId: s.runId, publicAccessToken: s.publicAccessToken, lastEventId: s.lastEventId ?? undefined, diff --git a/references/ai-chat/src/components/chat-app.tsx b/references/ai-chat/src/components/chat-app.tsx index de8e4f464..b3e7a6fcb 100644 --- a/references/ai-chat/src/components/chat-app.tsx +++ b/references/ai-chat/src/components/chat-app.tsx @@ -26,6 +26,7 @@ type ChatMeta = { }; type SessionInfo = { + sessionId?: string; runId: string; publicAccessToken: string; lastEventId?: string; diff --git a/references/ai-chat/src/components/chat-view.tsx b/references/ai-chat/src/components/chat-view.tsx index c629326ec..59dcf196d 100644 --- a/references/ai-chat/src/components/chat-view.tsx +++ b/references/ai-chat/src/components/chat-view.tsx @@ -16,6 +16,7 @@ import { useCallback, useEffect, useState } from "react"; import { useRouter } from "next/navigation"; type SessionInfo = { + sessionId?: string; runId: string; publicAccessToken: string; lastEventId?: string; diff --git a/references/ai-chat/src/trigger/chat.ts b/references/ai-chat/src/trigger/chat.ts index 3979a6ed7..5ca98fcd2 100644 --- a/references/ai-chat/src/trigger/chat.ts +++ b/references/ai-chat/src/trigger/chat.ts @@ -1,5 +1,6 @@ import { chat, type ChatTaskWirePayload } from "@trigger.dev/sdk/ai"; import { logger, prompts, skills } from "@trigger.dev/sdk"; + import { streamText, generateText, @@ -293,8 +294,8 @@ export const aiChat = chat }); await prisma.chatSession.upsert({ where: { id: chatId }, - create: { id: chatId, runId, publicAccessToken: chatAccessToken }, - update: { runId, publicAccessToken: chatAccessToken }, + create: { id: chatId, sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, + update: { sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, }); }, // #endregion @@ -345,8 +346,8 @@ export const aiChat = chat await prisma.chatSession.upsert({ where: { id: chatId }, - create: { id: chatId, runId, publicAccessToken: chatAccessToken }, - update: { runId, publicAccessToken: chatAccessToken }, + create: { id: chatId, sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, + update: { sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, }); }, // #endregion @@ -416,8 +417,8 @@ export const aiChat = chat }); await prisma.chatSession.upsert({ where: { id: chatId }, - create: { id: chatId, runId, publicAccessToken: chatAccessToken, lastEventId }, - update: { runId, publicAccessToken: chatAccessToken, lastEventId }, + create: { id: chatId, sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken, lastEventId }, + update: { sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken, lastEventId }, }); // Background self-review — a cheap model critiques the response and @@ -903,8 +904,8 @@ export const aiChatHydrated = chat await initUserContext(clientData.userId, chatId, clientData.model); await prisma.chatSession.upsert({ where: { id: chatId }, - create: { id: chatId, runId, publicAccessToken: chatAccessToken }, - update: { runId, publicAccessToken: chatAccessToken }, + create: { id: chatId, sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, + update: { sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, }); }, @@ -913,8 +914,8 @@ export const aiChatHydrated = chat await initUserContext(clientData.userId, chatId, clientData.model); await prisma.chatSession.upsert({ where: { id: chatId }, - create: { id: chatId, runId, publicAccessToken: chatAccessToken }, - update: { runId, publicAccessToken: chatAccessToken }, + create: { id: chatId, sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, + update: { sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken }, }); }, @@ -925,8 +926,8 @@ export const aiChatHydrated = chat }); await prisma.chatSession.upsert({ where: { id: chatId }, - create: { id: chatId, runId, publicAccessToken: chatAccessToken, lastEventId }, - update: { runId, publicAccessToken: chatAccessToken, lastEventId }, + create: { id: chatId, sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken, lastEventId }, + update: { sessionId: chat.sessionId, runId, publicAccessToken: chatAccessToken, lastEventId }, }); }, diff --git a/references/ai-chat/src/trigger/sessions-smoke.ts b/references/ai-chat/src/trigger/sessions-smoke.ts new file mode 100644 index 000000000..8d5d21cd4 --- /dev/null +++ b/references/ai-chat/src/trigger/sessions-smoke.ts @@ -0,0 +1,73 @@ +import { task, sessions, logger } from "@trigger.dev/sdk"; + +const SMOKE_TYPE = "smoke.sessions"; + +export const sessionsT21Idempotency = task({ + id: "sessions-t21-idempotency", + run: async () => { + const externalId = `smoke-idem-${Date.now()}`; + const a = await sessions.create({ type: SMOKE_TYPE, externalId }); + const b = await sessions.create({ type: SMOKE_TYPE, externalId }); + return { a: a.id, b: b.id, idempotent: a.id === b.id }; + }, +}); + +export const sessionsT22ListByType = task({ + id: "sessions-t22-list-by-type", + run: async () => { + const externalId = `smoke-list-${Date.now()}`; + const created = await sessions.create({ type: SMOKE_TYPE, externalId }); + const page = await sessions.list({ type: SMOKE_TYPE, limit: 50 }); + const found = page.data.find((s) => s.id === created.id); + return { createdId: created.id, count: page.data.length, found: !!found }; + }, +}); + +export const sessionsT23ListByTag = task({ + id: "sessions-t23-list-by-tag", + run: async () => { + const tag = `smoke-tag-${Date.now()}`; + const created = await sessions.create({ type: SMOKE_TYPE, tags: [tag] }); + const page = await sessions.list({ tag, limit: 50 }); + const found = page.data.find((s) => s.id === created.id); + return { tag, createdId: created.id, count: page.data.length, found: !!found }; + }, +}); + +export const sessionsT24CrossRunOpen = task({ + id: "sessions-t24-cross-run-open", + run: async () => { + const externalId = `smoke-cross-${Date.now()}`; + const created = await sessions.create({ type: SMOKE_TYPE, externalId }); + // Reopen via friendlyId — no network call (open is lazy) + const handleByFriendly = sessions.open(created.id); + // Reopen via externalId + const handleByExternal = sessions.open(externalId); + // Append a record from each handle to verify both write to the same stream + await handleByFriendly.out.append({ type: "smoke", from: "friendly", t: Date.now() }); + await handleByExternal.out.append({ type: "smoke", from: "external", t: Date.now() }); + return { + sessionId: created.id, + handleFriendlyId: handleByFriendly.id, + handleExternalId: handleByExternal.id, + }; + }, +}); + +export const sessionsT26CloseAndReopen = task({ + id: "sessions-t26-close-and-reopen", + run: async () => { + const externalId = `smoke-close-${Date.now()}`; + const created = await sessions.create({ type: SMOKE_TYPE, externalId }); + await sessions.close(created.id, { reason: "smoke close" }); + const after = await sessions.retrieve(created.id); + // Recreate by externalId — should be idempotent (return same id) or a fresh one + const recreated = await sessions.create({ type: SMOKE_TYPE, externalId }); + return { + created: created.id, + closedStatus: (after as { status?: string }).status, + recreated: recreated.id, + sameId: created.id === recreated.id, + }; + }, +});