diff --git a/.changeset/headstart-pending-version.md b/.changeset/headstart-pending-version.md new file mode 100644 index 00000000000..1d8aa090686 --- /dev/null +++ b/.changeset/headstart-pending-version.md @@ -0,0 +1,12 @@ +--- +"@trigger.dev/sdk": patch +--- + +Head Start now tells you when the agent run it handed over to is waiting on a deployment that is still building. Step 1 always streams from your warm process, so the wait only affects step 2, and it used to be invisible: the transport now emits `run-pending-version` with `source: "head-start"`, `chat.startHeadStart` returns `pendingVersion`, and the `chat.handover` session handle exposes it too. + +```tsx +onEvent: (event) => { + if (event.type === "run-pending-version") setDeploying(true); + if (event.type === "first-chunk") setDeploying(false); +}, +``` diff --git a/docs/ai-chat/fast-starts.mdx b/docs/ai-chat/fast-starts.mdx index 61d9c468f25..5d8392c5ffd 100644 --- a/docs/ai-chat/fast-starts.mdx +++ b/docs/ai-chat/fast-starts.mdx @@ -735,11 +735,13 @@ chat.startHeadStart({ triggerConfig?: Partial, // tags, queue, machine, … apiClient?: ApiClientConfiguration, // when the agent lives in another project/env metadata?: Record, // merged into the run payload; never sent to the browser -}): Promise<{ chatId: string; completion: Promise }> +}): Promise<{ chatId: string; pendingVersion: boolean; completion: Promise }> ``` `completion` resolves once the head start finishes; `await` it or hand it to `waitUntil`. It rejects if the warm step or the dispatch fails. +`pendingVersion` is `true` when the agent run is parked waiting for the deployment carrying the session's [external deployment id](/deployment/version-skew-protection#chat-sessions). Step 1 still runs in your process and still reaches the browser, so pass the flag to the destination page if you want it to say a deploy is in progress rather than appear to stall on step 2. + ### Limitations - **First turn only.** Step 2+ and turn 2+ run on the trigger side. There's no per-turn "head start every turn" mode — the win comes from amortizing agent boot across the LLM call once. diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx index d21bec858cd..50a9f8435d1 100644 --- a/docs/ai-chat/reference.mdx +++ b/docs/ai-chat/reference.mdx @@ -492,7 +492,7 @@ Options for [`chat.headStart()`](/ai-chat/fast-starts#head-start), the warm-serv | `idleTimeoutInSeconds` | `number` | `60` | How long the agent waits for the handover signal | | `triggerConfig` | `Partial` | `undefined` | Run options (tags, queue, machine, maxAttempts, maxDuration, region, lockToVersion, externalDeploymentId) for the auto-triggered handover-prepare run. The `chat:{chatId}` tag is prepended automatically | -`chat.headStart(options)` returns the handler `(req: Request) => Promise`. The `run` callback receives `HeadStartRunArgs`: `{ messages: UIMessage[], signal: AbortSignal, chat: HeadStartChatHelper }`, where the helper exposes `chat.toStreamTextOptions({ tools })` and a `chat.session` escape hatch. See [Head Start](/ai-chat/fast-starts#head-start) for the full guide. +`chat.headStart(options)` returns the handler `(req: Request) => Promise`. The `run` callback receives `HeadStartRunArgs`: `{ messages: UIMessage[], signal: AbortSignal, chat: HeadStartChatHelper }`, where the helper exposes `chat.toStreamTextOptions({ tools })` and a `chat.session` escape hatch (whose `pendingVersion` says whether the agent run is parked waiting for its deployment). See [Head Start](/ai-chat/fast-starts#head-start) for the full guide. ## chat namespace @@ -643,6 +643,7 @@ The `onEvent` callback receives a `ChatTransportEvent` (exported from `@trigger. | --- | --- | --- | | `message-sent` | `messageId?`, `source`, `durationMs`, `partId?`, `bodyBytes?` | A send was durably acknowledged — a 2xx from the session input stream append (or the `headStart` POST), after any internal token-refresh retries. This means the message is durably written to the stream the agent consumes from, not merely "request accepted". `partId` is the append's idempotency key, also stored on the server-side record. | | `message-send-failed` | `messageId?`, `source`, `error`, `status?`, `durationMs`, `partId?`, `bodyBytes?` | A send definitively failed after internal retries. Fires in addition to `useChat`'s `onError`. | +| `run-pending-version` | `source` | The chat's run is parked waiting for the deployment carrying its external deployment id ([version skew protection](/deployment/version-skew-protection#chat-sessions)). Everything already sent is durable and answered once the deployment lands. `source` is `"start"` (learned while starting the session), `"send"` (from a message append, re-emitted on every send while parked) or `"head-start"` (from the `headStart` POST, where step 1 still streams from your server and only step 2 waits). | | `stream-connected` | `resumed`, `lastEventId?`, `messageId?` | The SSE subscription to the session's output stream started delivering. `resumed: true` when reconnecting from a stored cursor (page reload) rather than following a fresh send. `lastEventId` is the cursor it connected from. | | `first-chunk` | `chunkType?`, `lastEventId?`, `messageId?`, `sinceSendMs?` | The first response chunk of a turn arrived. `sinceSendMs` is the delta from the last turn-producing send — time to first token without any bookkeeping. | | `turn-completed` | `lastEventId?`, `sessionInEventId?`, `messageId?`, `sinceSendMs?` | The agent's turn-complete control record arrived — the "finished answering" signal. `sinceSendMs` is the full turn latency; `sessionInEventId` is the agent's committed input-stream cursor. | diff --git a/docs/deployment/version-skew-protection.mdx b/docs/deployment/version-skew-protection.mdx index 3a9dc07c329..a12659a3416 100644 --- a/docs/deployment/version-skew-protection.mdx +++ b/docs/deployment/version-skew-protection.mdx @@ -314,6 +314,8 @@ const transport = useTriggerChatTransport({ The event repeats on every message sent while the chat is parked, so a notice driven off it stays accurate. +[Head Start](/ai-chat/fast-starts#head-start) softens this considerably: turn 1 runs in your own warm process, so a parked deployment costs nothing until step 2. The handover signal is durable, so the agent picks the turn up where it left off once the deployment lands. The transport emits `run-pending-version` with `source: "head-start"` for that case, and `chat.startHeadStart` returns `pendingVersion` for the detached flow. + ### Opting a chat out Pass `null` and that chat is never pinned, whatever the environment says: diff --git a/packages/trigger-sdk/src/v3/chat-server.test.ts b/packages/trigger-sdk/src/v3/chat-server.test.ts index 539fe0247ad..d50a4e04365 100644 --- a/packages/trigger-sdk/src/v3/chat-server.test.ts +++ b/packages/trigger-sdk/src/v3/chat-server.test.ts @@ -89,7 +89,7 @@ function makeRequest(body: unknown): Request { const SESSION_PAT = "tr_session_pat_for_handover"; -function createSessionResponse(externalId: string): Response { +function createSessionResponse(externalId: string, opts?: { pendingVersion?: boolean }): Response { return new Response( JSON.stringify({ id: "session_test", @@ -111,6 +111,7 @@ function createSessionResponse(externalId: string): Response { createdAt: new Date(0).toISOString(), updatedAt: new Date(0).toISOString(), isCached: false, + ...(opts?.pendingVersion ? { pendingVersion: true } : {}), }), { status: 200, @@ -160,6 +161,57 @@ describe("chat.headStart (route handler)", () => { vi.restoreAllMocks(); }); + it("reports a parked agent run in the response headers", 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-parked", { pendingVersion: true }); + } + if (urlStr.includes("/realtime/v1/sessions/") && urlStr.endsWith("/in/append")) { + return appendOkResponse(); + } + // Stitched response subscribes to `.out` after handover. + if (/\/realtime\/v1\/sessions\/[^/]+\/out$/.test(urlStr)) { + return new Response( + new ReadableStream({ + start(c) { + c.close(); + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } } + ); + } + throw new Error(`Unexpected URL: ${urlStr}`); + }); + + const handler = chat.headStart({ + agentId: "test-agent", + run: async ({ chat: chatHelper }) => + streamText({ + ...chatHelper.toStreamTextOptions(), + model: new MockLanguageModelV3({ + doStream: async () => ({ stream: textStream("step 1 while parked") }), + }), + }), + }); + + const res = await withApiContext(() => + handler( + makeRequest({ + chatId: "chat-parked", + trigger: "submit-message", + headStartMessages: [{ id: "m1", role: "user", parts: [{ type: "text", text: "hi" }] }], + }) + ) + ); + + // Step 1 still streams from this process even though nothing can answer step 2 yet. + expect(res.status).toBe(200); + expect(res.headers.get("X-Trigger-Chat-Pending-Version")).toBe("1"); + const chunks = await readSSEBodyToChunks(res); + expect(chunks.length).toBeGreaterThan(0); + }); + 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) => { @@ -200,6 +252,8 @@ describe("chat.headStart (route handler)", () => { expect(res.headers.get("X-Trigger-Chat-Id")).toBe("chat-1"); expect(res.headers.get("X-Trigger-Chat-Access-Token")).toBe(SESSION_PAT); expect(res.headers.get("Content-Type")).toMatch(/text\/event-stream/); + // Not parked, so the header is absent rather than "0". + expect(res.headers.get("X-Trigger-Chat-Pending-Version")).toBeNull(); const sessionCreate = requests.find( (r) => r.url.endsWith("/api/v1/sessions") || r.url.endsWith("/api/v1/sessions/") diff --git a/packages/trigger-sdk/src/v3/chat-server.ts b/packages/trigger-sdk/src/v3/chat-server.ts index 98bf7b99854..0154dbc3f11 100644 --- a/packages/trigger-sdk/src/v3/chat-server.ts +++ b/packages/trigger-sdk/src/v3/chat-server.ts @@ -141,6 +141,11 @@ export type HeadStartChatHelper> = { export type HeadStartSession = { readonly chatId: string; + /** + * The agent run is parked waiting for a deployment carrying the session's external deployment + * id. Step 1 still streams from this process; step 2 lands once the deployment does. + */ + readonly pendingVersion: boolean; /** * Tees a UIMessage stream into `session.out` for durability/resume, * fire-and-forget. Returns a passthrough that the caller can use as @@ -235,6 +240,8 @@ export type StartHeadStartOptions> = { export type StartHeadStartResult = { /** The chat id you passed in — echoed for convenience. */ chatId: string; + /** See {@link HeadStartSession.pendingVersion}. */ + pendingVersion: boolean; /** * Resolves once step 1 has drained to `session.out` and the handover is * dispatched. Hand to `waitUntil` / `after` on serverless; ignore it on a @@ -389,7 +396,7 @@ export const chat = { // returned promise still surfaces the error. completion.catch(() => {}); - return { chatId: opts.chatId, completion }; + return { chatId: opts.chatId, pendingVersion: session.handle.pendingVersion, completion }; }, /** @@ -584,6 +591,7 @@ async function openHandoverSession(opts: { }) ); const sessionPublicAccessToken = created.publicAccessToken; + const pendingVersion = created.pendingVersion === true; // Combined abort signal: request lifecycle OR an internal timeout // mirroring the agent's idle wait so a hung handler doesn't sit @@ -967,12 +975,15 @@ async function openHandoverSession(opts: { // without going back through the handler. "X-Trigger-Chat-Id": chatId, "X-Trigger-Chat-Access-Token": sessionPublicAccessToken, + // Only sent when parked, so an unpinned chat's headers are unchanged. + ...(pendingVersion ? { "X-Trigger-Chat-Pending-Version": "1" } : {}), }, }); }; const handle: HeadStartSession = { chatId, + pendingVersion, tee, handoverWhenDone, handoverResponse, diff --git a/packages/trigger-sdk/src/v3/chat.test.ts b/packages/trigger-sdk/src/v3/chat.test.ts index d0dbd61e2a3..bbe02eff5ae 100644 --- a/packages/trigger-sdk/src/v3/chat.test.ts +++ b/packages/trigger-sdk/src/v3/chat.test.ts @@ -1871,6 +1871,7 @@ describe("TriggerChatTransport", () => { chatId: string; accessToken: string; chunks: UIMessageChunk[]; + pendingVersion?: boolean; }): Response { return new Response(handoverSseBody(args.chunks), { status: 200, @@ -1878,10 +1879,48 @@ describe("TriggerChatTransport", () => { "content-type": "text/event-stream", "X-Trigger-Chat-Id": args.chatId, "X-Trigger-Chat-Access-Token": args.accessToken, + ...(args.pendingVersion ? { "X-Trigger-Chat-Pending-Version": "1" } : {}), }, }); } + it("emits run-pending-version when the handover endpoint reports a parked run", async () => { + global.fetch = vi.fn().mockImplementation(async (url: string | URL) => { + const urlStr = typeof url === "string" ? url : url.toString(); + if (urlStr === "https://my-app.example/api/chat") { + return handoverResponse({ + chatId: "chat-handover-parked", + accessToken: "handover-pat-parked", + chunks: sampleChunks, + pendingVersion: true, + }); + } + throw new Error(`Unexpected URL: ${urlStr}`); + }); + + const events: ChatTransportEvent[] = []; + const transport = new TriggerChatTransport({ + task: "my-chat-task", + accessToken: () => "pat", + headStart: "https://my-app.example/api/chat", + onEvent: (event) => events.push(event), + }); + + const stream = await transport.sendMessages({ + trigger: "submit-message", + chatId: "chat-handover-parked", + messageId: "m1", + messages: [createUserMessage("hello")], + abortSignal: undefined, + }); + // Step 1 still arrives from the warm server. + expect(await drainChunks(stream)).toEqual(sampleChunks); + + const parked = events.filter((e) => e.type === "run-pending-version"); + expect(parked).toHaveLength(1); + expect(parked[0]).toMatchObject({ chatId: "chat-handover-parked", source: "head-start" }); + }); + it("first-turn POSTs the wire payload to endpoint when no session exists", async () => { const requests: Array<{ url: string; init?: RequestInit }> = []; global.fetch = vi.fn().mockImplementation(async (url: string | URL, init?: RequestInit) => { diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts index 59421922682..f2db6653649 100644 --- a/packages/trigger-sdk/src/v3/chat.ts +++ b/packages/trigger-sdk/src/v3/chat.ts @@ -234,8 +234,8 @@ export type ChatTransportEvent = type: "run-pending-version"; chatId: string; timestamp: number; - /** Whether we learned this from starting the session or from sending a message. */ - source: "start" | "send"; + /** Whether we learned this from starting the session, a send, or the `headStart` POST. */ + source: "start" | "send" | "head-start"; } | { type: "message-sent"; @@ -983,6 +983,17 @@ export class TriggerChatTransport implements ChatTransport { this.sessions.set(chatId, state); this.notifySessionChange(chatId, state); + // Step 1 streams from the warm server either way; this says the agent run that owes step 2 + // is parked on an undeployed external deployment id. + if (response.headers.get("X-Trigger-Chat-Pending-Version") === "1") { + this.emitEvent({ + type: "run-pending-version", + chatId, + timestamp: Date.now(), + source: "head-start", + }); + } + // Filter the parsed UIMessage stream: // - Drop control chunks (`trigger:turn-complete`, // `trigger:session-state`) before they reach AI SDK — they