diff --git a/.changeset/agentchat-forward-trigger-config.md b/.changeset/agentchat-forward-trigger-config.md new file mode 100644 index 00000000000..d0cb1bc69cb --- /dev/null +++ b/.changeset/agentchat-forward-trigger-config.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Fixed `AgentChat` silently ignoring `maxDuration`, `region` and `lockToVersion` when they were set on its `triggerConfig`. They are now applied to the session's runs. diff --git a/.changeset/chat-agent-version-skew-protection.md b/.changeset/chat-agent-version-skew-protection.md new file mode 100644 index 00000000000..c8c15da3d4c --- /dev/null +++ b/.changeset/chat-agent-version-skew-protection.md @@ -0,0 +1,15 @@ +--- +"@trigger.dev/core": patch +"@trigger.dev/sdk": patch +--- + +Chat sessions now stay on the deployment that matched the app build that started them, so a conversation keeps talking to the agent version its release shipped with across every turn, idle suspend and recovery. The id is resolved wherever you start the session, exactly as it is for `trigger()`, so a chat picks up whatever your app already sends when it triggers a task, with no chat-specific setup. If your app sends no id, nothing changes: chats run on the current version as they do today. + +```ts +// Opt a single chat out of pinning: +export const startChatSession = chat.createStartSessionAction("my-chat", { + triggerConfig: { externalDeploymentId: null }, +}); +``` + +Messages sent while a chat waits on a deployment that is still building are stored and answered once it lands, and the transport emits a `run-pending-version` event so your UI can say so. `chat.requestUpgrade()` now clears the session's pin so the handoff can reach a new version, and accepts `{ externalDeploymentId }` to move to a specific one. diff --git a/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts b/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts index c6e5a90f667..ce3fbeb9f6d 100644 --- a/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts +++ b/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts @@ -117,6 +117,7 @@ const { action, loader } = createActionApiRoute( callingRunId: callingRun.id, environment: authentication.environment, reason, + externalDeploymentId: body.externalDeploymentId, }); // Read-after-write: the swap just triggered (or claimed) the diff --git a/apps/webapp/app/routes/api.v1.sessions.ts b/apps/webapp/app/routes/api.v1.sessions.ts index c7cf2b27a0f..c5372a9f844 100644 --- a/apps/webapp/app/routes/api.v1.sessions.ts +++ b/apps/webapp/app/routes/api.v1.sessions.ts @@ -246,6 +246,7 @@ const { action } = createActionApiRoute( runId: run.friendlyId, publicAccessToken, isCached, + pendingVersion: ensureResult.pendingVersion, }; return json(responseBody, { diff --git a/apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts b/apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts index d4dd1d9f19f..5723e1e9c54 100644 --- a/apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts +++ b/apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts @@ -121,7 +121,7 @@ const { action, loader } = createActionApiRoute( // durable and the next append will retry the ensure step. Don't // surface the error to the caller; the SSE tail just won't deliver // it until a run boots. - const [ensureError] = await tryCatch( + const [ensureError, ensureResult] = await tryCatch( ensureRunForSession({ session, environment: authentication.environment, @@ -236,7 +236,14 @@ const { action, loader } = createActionApiRoute( } // `seq` lets the client correlate this send to the turn that consumes it. - return json({ ok: true, seq: appendSeq }, { status: 200 }); + return json( + { + ok: true, + seq: appendSeq, + ...(ensureResult?.pendingVersion ? { pendingVersion: true } : {}), + }, + { status: 200 } + ); } ); diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts index a1989a9ef7a..215d850176d 100644 --- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts +++ b/apps/webapp/app/services/realtime/sessionRunManager.server.ts @@ -1,4 +1,4 @@ -import type { Session, TaskRunStatus } from "@trigger.dev/database"; +import type { Prisma, Session, TaskRunStatus } from "@trigger.dev/database"; import { SessionTriggerConfig as SessionTriggerConfigZod } from "@trigger.dev/core/v3"; import type { z } from "zod"; import { prisma, $replica } from "~/db.server"; @@ -76,6 +76,8 @@ export type EnsureRunResult = { runId: string; /** True if this call triggered a fresh run; false if it reused an alive existing one. */ triggered: boolean; + /** The run is parked waiting for a deployment carrying the session's external deployment id. */ + pendingVersion: boolean; }; /** @@ -123,7 +125,11 @@ export async function ensureRunForSession( ); } if (probe && !isFinalRunStatus(probe.status)) { - return { runId: session.currentRunId, triggered: false }; + return { + runId: session.currentRunId, + triggered: false, + pendingVersion: isPendingVersionStatus(probe.status), + }; } // Either the row vanished on the writer too (probe null) or its status // is final. Either way the prior run isn't going to consume new @@ -210,7 +216,11 @@ export async function ensureRunForSession( }); }); - return { runId: triggered.id, triggered: true }; + return { + runId: triggered.id, + triggered: true, + pendingVersion: isPendingVersionStatus(triggered.status), + }; } // 4. Lost the race. Cancel our triggered run; reuse the winner's. @@ -255,7 +265,11 @@ export async function ensureRunForSession( prisma ); if (probe && !isFinalRunStatus(probe.status)) { - return { runId: fresh.currentRunId, triggered: false }; + return { + runId: fresh.currentRunId, + triggered: false, + pendingVersion: isPendingVersionStatus(probe.status), + }; } } @@ -272,6 +286,24 @@ export async function ensureRunForSession( }); } +/** Both version pins are forwarded; `TriggerTaskService` decides which governs. */ +export function buildSessionRunOptions(config: SessionTriggerConfig) { + return { + ...(config.machine ? { machine: config.machine as never } : {}), + ...(config.queue ? { queue: { name: config.queue } } : {}), + ...(config.tags ? { tags: config.tags } : {}), + ...(config.maxAttempts !== undefined ? { maxAttempts: config.maxAttempts } : {}), + ...(config.maxDuration !== undefined ? { maxDuration: config.maxDuration } : {}), + ...(config.lockToVersion ? { lockToVersion: config.lockToVersion } : {}), + ...(config.externalDeploymentId ? { externalDeploymentId: config.externalDeploymentId } : {}), + ...(config.region ? { region: config.region } : {}), + }; +} + +function isPendingVersionStatus(status: TaskRunStatus): boolean { + return status === "PENDING_VERSION"; +} + /** * Trigger a single run for a session. Builds `TriggerTaskRequestBody` * by shallow-merging `payloadOverrides` over `config.basePayload` and @@ -288,7 +320,7 @@ async function triggerSessionRun(params: { config: SessionTriggerConfig; environment: AuthenticatedEnvironment; payloadOverrides?: Record; -}): Promise<{ id: string; friendlyId: string }> { +}): Promise<{ id: string; friendlyId: string; status: TaskRunStatus }> { const { session, config, environment, payloadOverrides } = params; const payload = { @@ -302,15 +334,7 @@ async function triggerSessionRun(params: { const body = { payload, context: {}, - options: { - ...(config.machine ? { machine: config.machine as never } : {}), - ...(config.queue ? { queue: { name: config.queue } } : {}), - ...(config.tags ? { tags: config.tags } : {}), - ...(config.maxAttempts !== undefined ? { maxAttempts: config.maxAttempts } : {}), - ...(config.maxDuration !== undefined ? { maxDuration: config.maxDuration } : {}), - ...(config.lockToVersion ? { lockToVersion: config.lockToVersion } : {}), - ...(config.region ? { region: config.region } : {}), - }, + options: buildSessionRunOptions(config), }; const service = new TriggerTaskService(); @@ -329,7 +353,11 @@ async function triggerSessionRun(params: { ); } - return { id: result.run.id, friendlyId: result.run.friendlyId }; + return { + id: result.run.id, + friendlyId: result.run.friendlyId, + status: result.run.status, + }; } type SwapSessionRunParams = { @@ -359,6 +387,8 @@ type SwapSessionRunParams = { environment: AuthenticatedEnvironment; reason: EnsureRunReason; payloadOverrides?: Record; + /** Only read when `reason` is `"upgrade"`: a string re-pins the session, absent clears the pin. */ + externalDeploymentId?: string | null; }; export type SwapSessionRunResult = { @@ -371,6 +401,8 @@ export type SwapSessionRunResult = { * next run. */ swapped: boolean; + /** See {@link EnsureRunResult.pendingVersion}. */ + pendingVersion: boolean; }; /** @@ -413,7 +445,15 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise { }); // Observable: the writer re-probe recovered the live run → reuse it, do NOT trigger a second run. - expect(result).toEqual({ runId, triggered: false }); + expect(result).toEqual({ runId, triggered: false, pendingVersion: false }); expect(triggerState.calls).toHaveLength(0); // The replica WAS consulted first (and, frozen, missed) — proving the recovery is the writer // re-probe, not a lucky replica hit. @@ -423,7 +423,7 @@ describe("realtime-svc — replica-lag guards", () => { }); // Observable 1: the swap COMPLETED — the replica miss did not fail it. - expect(result).toEqual({ runId: newRunId, swapped: true }); + expect(result).toEqual({ runId: newRunId, swapped: true, pendingVersion: false }); // Observable 2: resolveRunFriendlyId missed on the replica and degraded to the cuid, so the // previousRunId forwarded to the triggered run is the calling run's cuid (documented fallback). diff --git a/apps/webapp/test/sessionRunManagerExternalDeploymentId.test.ts b/apps/webapp/test/sessionRunManagerExternalDeploymentId.test.ts new file mode 100644 index 00000000000..b6473ecb6ba --- /dev/null +++ b/apps/webapp/test/sessionRunManagerExternalDeploymentId.test.ts @@ -0,0 +1,505 @@ +// Version-skew pinning for chat sessions, driven through the real `ensureRunForSession` / +// `swapSessionRun` against a real Postgres: the stored pin is forwarded to every run the session +// schedules, and `reason: "upgrade"` drops it (or replaces it) AND persists that on the row. + +import { postgresTest } from "@internal/testcontainers"; +import type { PrismaClient } from "@trigger.dev/database"; +import { describe, expect, vi } from "vitest"; + +vi.setConfig({ testTimeout: 60_000, hookTimeout: 60_000 }); + +const primaryHolder = vi.hoisted(() => ({ client: undefined as any })); +const storeHolder = vi.hoisted(() => ({ store: undefined as any })); + +vi.mock("~/services/realtime/v1StreamsGlobal.server", () => ({ + determineRealtimeStreamsVersion: () => "v2", +})); + +const triggerState = vi.hoisted(() => ({ + calls: [] as Array<{ taskIdentifier: string; body: any; options: any }>, + result: { run: { id: "", friendlyId: "", status: "PENDING" } } as { + run: { id: string; friendlyId: string; status: string }; + }, +})); + +vi.mock("~/db.server", () => { + const lazyProxy = (holder: { client: any }, label: string) => + new Proxy( + {}, + { + get(_t, prop) { + if (!holder.client) throw new Error(`${label} not set for this test`); + const value = holder.client[prop]; + if (value !== null && typeof value === "object") { + return new Proxy(value, { get: (_d, method) => holder.client[prop][method] }); + } + return value; + }, + } + ); + return { + prisma: lazyProxy(primaryHolder, "primaryHolder.client"), + $replica: lazyProxy(primaryHolder, "primaryHolder.client"), + }; +}); + +vi.mock("~/v3/runStore.server", () => ({ + runStore: new Proxy( + {}, + { + get(_t, prop) { + const store = storeHolder.store as Record; + if (!store) throw new Error("test bug: storeHolder.store not set before caller ran"); + const value = store[prop]; + return typeof value === "function" + ? (value as (...a: unknown[]) => unknown).bind(store) + : value; + }, + } + ), +})); + +vi.mock("~/services/logger.server", () => ({ + logger: { warn: vi.fn(), info: vi.fn(), error: vi.fn(), debug: vi.fn() }, +})); + +vi.mock("~/v3/services/triggerTask.server", () => ({ + TriggerTaskService: class { + async call(taskIdentifier: string, _environment: any, body: any, options: any) { + triggerState.calls.push({ taskIdentifier, body, options }); + return triggerState.result; + } + }, +})); + +vi.mock("~/v3/services/cancelTaskRun.server", () => ({ + CancelTaskRunService: class { + async call() {} + }, +})); + +import { PostgresRunStore } from "@internal/run-store"; +import type { CreateRunInput } from "@internal/run-store"; +import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; +import { ensureRunForSession, swapSessionRun } from "~/services/realtime/sessionRunManager.server"; + +let seq = 0; + +const cuidRunId = (suffix: string) => `run_${suffix.padEnd(24, "x").slice(0, 24)}`; + +async function seedTenant(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "PRODUCTION", + slug: "prod", + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_prod_${suffix}`, + pkApiKey: `pk_prod_${suffix}`, + shortcode: `short_${suffix}`, + }, + }); + return { organization, project, environment }; +} + +function buildCreateRunInput(p: { + runId: string; + friendlyId: string; + organizationId: string; + projectId: string; + runtimeEnvironmentId: string; + status?: CreateRunInput["data"]["status"]; +}): CreateRunInput { + return { + data: { + id: p.runId, + engine: "V2", + status: p.status ?? "PENDING", + friendlyId: p.friendlyId, + runtimeEnvironmentId: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + organizationId: p.organizationId, + projectId: p.projectId, + taskIdentifier: "my-chat", + payload: '{"hello":"world"}', + payloadType: "application/json", + context: { foo: "bar" }, + traceContext: { trace: "ctx" }, + traceId: "trace_1", + spanId: "span_1", + runTags: [], + queue: "task/my-chat", + isTest: false, + taskEventStore: "taskEvent", + depth: 0, + createdAt: new Date("2024-01-01T00:00:00.000Z"), + }, + snapshot: { + engine: "V2", + executionStatus: "RUN_CREATED", + description: "Run was created", + runStatus: p.status ?? "PENDING", + environmentId: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + projectId: p.projectId, + organizationId: p.organizationId, + }, + }; +} + +function environmentFor(seed: Awaited>) { + return { + id: seed.environment.id, + organization: { streamBasinName: null }, + } as unknown as AuthenticatedEnvironment; +} + +async function setup(prisma: PrismaClient, suffix: string) { + const seed = await seedTenant(prisma, suffix); + const writerStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + primaryHolder.client = prisma; + storeHolder.store = writerStore; + triggerState.calls.length = 0; + return { seed, writerStore }; +} + +/** The trigger options the caller built from the session's config on the last recorded call. */ +function optionsOnLastTrigger(): Record { + return triggerState.calls.at(-1)?.body?.options ?? {}; +} + +function pinOnLastTrigger(): unknown { + return optionsOnLastTrigger().externalDeploymentId; +} + +describe("session runs — external deployment id", () => { + postgresTest("an initial run forwards the session's stored pin", async ({ prisma }) => { + const suffix = `pin_initial_${seq++}`; + const { seed } = await setup(prisma as unknown as PrismaClient, suffix); + triggerState.result = { + run: { id: cuidRunId(`i${seq}`), friendlyId: `run_${suffix}`, status: "PENDING" }, + }; + + const session = await prisma.session.create({ + data: { + friendlyId: `session_${suffix}`, + type: "chat.agent", + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + environmentType: "PRODUCTION", + organizationId: seed.organization.id, + taskIdentifier: "my-chat", + triggerConfig: { basePayload: {}, externalDeploymentId: "commit-abc" }, + currentRunVersion: 0, + }, + }); + + const result = await ensureRunForSession({ + session, + environment: environmentFor(seed), + reason: "initial", + }); + + expect(result.triggered).toBe(true); + expect(pinOnLastTrigger()).toBe("commit-abc"); + }); + + postgresTest( + "a continuation re-applies the stored pin — the conversation stays on its deployment", + async ({ prisma }) => { + const suffix = `pin_cont_${seq++}`; + const { seed, writerStore } = await setup(prisma as unknown as PrismaClient, suffix); + + // A dead prior run, so `ensureRunForSession` takes the continuation branch. + const deadRunId = cuidRunId(`d${seq}`); + await writerStore.createRun( + buildCreateRunInput({ + runId: deadRunId, + friendlyId: `run_${suffix}_dead`, + organizationId: seed.organization.id, + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + status: "COMPLETED_SUCCESSFULLY", + }) + ); + triggerState.result = { + run: { id: cuidRunId(`c${seq}`), friendlyId: `run_${suffix}_new`, status: "PENDING" }, + }; + + const session = await prisma.session.create({ + data: { + friendlyId: `session_${suffix}`, + type: "chat.agent", + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + environmentType: "PRODUCTION", + organizationId: seed.organization.id, + taskIdentifier: "my-chat", + triggerConfig: { basePayload: {}, externalDeploymentId: "commit-abc" }, + currentRunId: deadRunId, + currentRunVersion: 0, + }, + }); + + const result = await ensureRunForSession({ + session, + environment: environmentFor(seed), + reason: "continuation", + }); + + expect(result.triggered).toBe(true); + expect(pinOnLastTrigger()).toBe("commit-abc"); + } + ); + + postgresTest("reports pendingVersion when the triggered run parks", async ({ prisma }) => { + const suffix = `pin_parked_${seq++}`; + const { seed } = await setup(prisma as unknown as PrismaClient, suffix); + triggerState.result = { + run: { + id: cuidRunId(`p${seq}`), + friendlyId: `run_${suffix}`, + status: "PENDING_VERSION", + }, + }; + + const session = await prisma.session.create({ + data: { + friendlyId: `session_${suffix}`, + type: "chat.agent", + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + environmentType: "PRODUCTION", + organizationId: seed.organization.id, + taskIdentifier: "my-chat", + triggerConfig: { basePayload: {}, externalDeploymentId: "not-deployed-yet" }, + currentRunVersion: 0, + }, + }); + + const result = await ensureRunForSession({ + session, + environment: environmentFor(seed), + reason: "initial", + }); + + expect(result.pendingVersion).toBe(true); + }); + + postgresTest( + "reuses a parked run rather than triggering a second one — appends queue instead", + async ({ prisma }) => { + const suffix = `pin_reuse_${seq++}`; + const { seed, writerStore } = await setup(prisma as unknown as PrismaClient, suffix); + + // PENDING_VERSION is non-final, so the probe must treat the parked run as alive. + const parkedRunId = cuidRunId(`r${seq}`); + await writerStore.createRun( + buildCreateRunInput({ + runId: parkedRunId, + friendlyId: `run_${suffix}_parked`, + organizationId: seed.organization.id, + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + status: "PENDING_VERSION", + }) + ); + + const session = await prisma.session.create({ + data: { + friendlyId: `session_${suffix}`, + type: "chat.agent", + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + environmentType: "PRODUCTION", + organizationId: seed.organization.id, + taskIdentifier: "my-chat", + triggerConfig: { basePayload: {}, externalDeploymentId: "not-deployed-yet" }, + currentRunId: parkedRunId, + currentRunVersion: 0, + }, + }); + + const result = await ensureRunForSession({ + session, + environment: environmentFor(seed), + reason: "continuation", + }); + + expect(result).toEqual({ runId: parkedRunId, triggered: false, pendingVersion: true }); + expect(triggerState.calls).toHaveLength(0); + } + ); + + postgresTest( + "an upgrade drops the pin and persists that, so the next continuation cannot bounce back", + async ({ prisma }) => { + const suffix = `pin_upgrade_${seq++}`; + const { seed, writerStore } = await setup(prisma as unknown as PrismaClient, suffix); + + const callingRunId = cuidRunId(`u${seq}`); + await writerStore.createRun( + buildCreateRunInput({ + runId: callingRunId, + friendlyId: `run_${suffix}_calling`, + organizationId: seed.organization.id, + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + }) + ); + triggerState.result = { + run: { id: cuidRunId(`u2${seq}`), friendlyId: `run_${suffix}_new`, status: "PENDING" }, + }; + + const session = await prisma.session.create({ + data: { + friendlyId: `session_${suffix}`, + type: "chat.agent", + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + environmentType: "PRODUCTION", + organizationId: seed.organization.id, + taskIdentifier: "my-chat", + triggerConfig: { + basePayload: {}, + externalDeploymentId: "commit-old", + lockToVersion: "20260807.1", + }, + currentRunId: callingRunId, + currentRunVersion: 0, + }, + }); + + const result = await swapSessionRun({ + session, + callingRunId, + environment: environmentFor(seed), + reason: "upgrade", + }); + + expect(result.swapped).toBe(true); + expect(pinOnLastTrigger()).toBeUndefined(); + // `lockToVersion` is an explicit customer pin with different intent — never cleared. + expect(optionsOnLastTrigger().lockToVersion).toBe("20260807.1"); + + const stored = await prisma.session.findFirstOrThrow({ where: { id: session.id } }); + expect(stored.triggerConfig).toMatchObject({ lockToVersion: "20260807.1" }); + expect(stored.triggerConfig).not.toHaveProperty("externalDeploymentId"); + } + ); + + postgresTest( + "an upgrade with an explicit target re-pins and persists the new id", + async ({ prisma }) => { + const suffix = `pin_repin_${seq++}`; + const { seed, writerStore } = await setup(prisma as unknown as PrismaClient, suffix); + + const callingRunId = cuidRunId(`t${seq}`); + await writerStore.createRun( + buildCreateRunInput({ + runId: callingRunId, + friendlyId: `run_${suffix}_calling`, + organizationId: seed.organization.id, + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + }) + ); + triggerState.result = { + run: { id: cuidRunId(`t2${seq}`), friendlyId: `run_${suffix}_new`, status: "PENDING" }, + }; + + const session = await prisma.session.create({ + data: { + friendlyId: `session_${suffix}`, + type: "chat.agent", + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + environmentType: "PRODUCTION", + organizationId: seed.organization.id, + taskIdentifier: "my-chat", + triggerConfig: { basePayload: {}, externalDeploymentId: "commit-old" }, + currentRunId: callingRunId, + currentRunVersion: 0, + }, + }); + + const result = await swapSessionRun({ + session, + callingRunId, + environment: environmentFor(seed), + reason: "upgrade", + externalDeploymentId: "commit-new", + }); + + expect(result.swapped).toBe(true); + expect(pinOnLastTrigger()).toBe("commit-new"); + + const stored = await prisma.session.findFirstOrThrow({ where: { id: session.id } }); + expect(stored.triggerConfig).toMatchObject({ externalDeploymentId: "commit-new" }); + } + ); + + postgresTest("a preempted upgrade leaves the stored config untouched", async ({ prisma }) => { + const suffix = `pin_preempt_${seq++}`; + const { seed, writerStore } = await setup(prisma as unknown as PrismaClient, suffix); + + const callingRunId = cuidRunId(`x${seq}`); + const winnerRunId = cuidRunId(`w${seq}`); + for (const [runId, name] of [ + [callingRunId, "calling"], + [winnerRunId, "winner"], + ] as const) { + await writerStore.createRun( + buildCreateRunInput({ + runId, + friendlyId: `run_${suffix}_${name}`, + organizationId: seed.organization.id, + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + }) + ); + } + triggerState.result = { + run: { id: cuidRunId(`x2${seq}`), friendlyId: `run_${suffix}_new`, status: "PENDING" }, + }; + + // `currentRunId` is already the winner, so the claim (keyed on callingRunId) cannot match. + const session = await prisma.session.create({ + data: { + friendlyId: `session_${suffix}`, + type: "chat.agent", + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + environmentType: "PRODUCTION", + organizationId: seed.organization.id, + taskIdentifier: "my-chat", + triggerConfig: { basePayload: {}, externalDeploymentId: "commit-old" }, + currentRunId: winnerRunId, + currentRunVersion: 0, + }, + }); + + const result = await swapSessionRun({ + session: { ...session, currentRunId: callingRunId }, + callingRunId, + environment: environmentFor(seed), + reason: "upgrade", + }); + + expect(result.swapped).toBe(false); + + const stored = await prisma.session.findFirstOrThrow({ where: { id: session.id } }); + expect(stored.triggerConfig).toMatchObject({ externalDeploymentId: "commit-old" }); + }); +}); diff --git a/apps/webapp/test/sessionRunTriggerOptions.test.ts b/apps/webapp/test/sessionRunTriggerOptions.test.ts new file mode 100644 index 00000000000..581f4647744 --- /dev/null +++ b/apps/webapp/test/sessionRunTriggerOptions.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; +import { buildSessionRunOptions } from "~/services/realtime/sessionRunManager.server"; + +const baseConfig = { basePayload: {} }; + +describe("buildSessionRunOptions", () => { + it("forwards the session's external deployment id", () => { + const options = buildSessionRunOptions({ ...baseConfig, externalDeploymentId: "commit-abc" }); + + expect(options.externalDeploymentId).toBe("commit-abc"); + }); + + it("forwards both pins so the trigger path can apply precedence", () => { + const options = buildSessionRunOptions({ + ...baseConfig, + lockToVersion: "20260807.1", + externalDeploymentId: "commit-abc", + }); + + expect(options.lockToVersion).toBe("20260807.1"); + expect(options.externalDeploymentId).toBe("commit-abc"); + }); + + it("omits the id when the session has no pin", () => { + expect(buildSessionRunOptions(baseConfig)).not.toHaveProperty("externalDeploymentId"); + }); + + it("still maps the rest of the config", () => { + const options = buildSessionRunOptions({ + ...baseConfig, + machine: "small-1x", + queue: "my-queue", + tags: ["chat:abc"], + maxAttempts: 3, + maxDuration: 600, + region: "us-east-1", + externalDeploymentId: "commit-abc", + }); + + expect(options).toMatchObject({ + machine: "small-1x", + queue: { name: "my-queue" }, + tags: ["chat:abc"], + maxAttempts: 3, + maxDuration: 600, + region: "us-east-1", + externalDeploymentId: "commit-abc", + }); + }); +}); diff --git a/docs/ai-chat/client-protocol.mdx b/docs/ai-chat/client-protocol.mdx index d039b39366a..857b09025f9 100644 --- a/docs/ai-chat/client-protocol.mdx +++ b/docs/ai-chat/client-protocol.mdx @@ -210,6 +210,7 @@ Pick `"preload"` when the UI has rendered but the user hasn't typed (warms the a | `triggerConfig.maxAttempts` | `number` | Per-run retry cap (1–10). | | `triggerConfig.maxDuration` | `number` | Per-run wall-clock cap, seconds. | | `triggerConfig.lockToVersion` | `string` | Pin every run to a specific worker version. | +| `triggerConfig.externalDeploymentId` | `string \| null` | Pin every run to the deployment carrying this [external deployment id](/deployment/version-skew-protection#chat-sessions). Discovered from the environment when omitted; `null` opts the chat out. | | `triggerConfig.region` | `string` | Region preference. | | `triggerConfig.idleTimeoutInSeconds` | `number` | Surfaced to the agent through the wire payload (1–3600). | diff --git a/docs/ai-chat/patterns/version-upgrades.mdx b/docs/ai-chat/patterns/version-upgrades.mdx index 830f673e9a3..c61efb4cade 100644 --- a/docs/ai-chat/patterns/version-upgrades.mdx +++ b/docs/ai-chat/patterns/version-upgrades.mdx @@ -8,6 +8,14 @@ Chat agent runs are pinned to the worker version they started on. When you deplo `chat.requestUpgrade()` lets the agent opt out of the current run so the transport triggers a new one on the latest version. + + If you use [version skew protection](/deployment/version-skew-protection#chat-sessions), most of + this page is done for you: sessions pin to the deployment matching the app build that started + them, so the agent and the frontend move together without a hand-maintained version. Read this + page for the cases skew protection doesn't cover — an agent that wants to leave its pin + mid-conversation, or a session that was never pinned. + + ## How it works When `chat.requestUpgrade()` is called in `onTurnStart` or `onValidateMessages`: @@ -19,6 +27,25 @@ When `chat.requestUpgrade()` is called in `onTurnStart` or `onValidateMessages`: The new run lives on the **same Session** as the old one. `chatId` is the durable identity; only the underlying `currentRunId` rotates. The audit log records the new run with `reason: "upgrade"`. +### What "the latest deployment" means + +The handoff clears the session's [external deployment id](/deployment/version-skew-protection#chat-sessions) so the new run can land on the current version — re-applying the pin the agent just rejected would make the upgrade impossible. The cleared pin is persisted on the session, so the next continuation doesn't fall back to it either. + +To move to a specific deployment rather than to whatever is current, name it: + +```ts +chat.requestUpgrade({ externalDeploymentId: clientData.commitSha }); +``` + +That is usually what you want when the client told you which build it is on: it upgrades to the version the client expects instead of merely to the newest one. + + + `lockToVersion` is a different thing and is **never** cleared. A session started with an explicit + `lockToVersion` re-applies it on every run including upgrade handoffs, so `chat.requestUpgrade()` + cannot escape it — the new run lands on the same version the old one did. Use the external + deployment id if you want a pin an agent can opt out of. + + When called from inside `run()` or `chat.defer()`, the current turn completes normally first and the run exits afterward. The next message triggers the continuation on the same session. ```mermaid @@ -151,6 +178,12 @@ export const myChat = chat This upgrades on **every** deploy, not just breaking changes. Good for fast-moving projects where you always want the latest code. + + With [version skew protection](/deployment/version-skew-protection#chat-sessions) on, you don't + need this recipe: the session is already pinned to the deployment that matches the app build that + started it, and a client on a new build re-pins the session when it starts. + + ## Other agent types - **`chat.agent()`** and **`chat.createSession()`** — use `chat.requestUpgrade()` as shown above @@ -162,6 +195,7 @@ This upgrades on **every** deploy, not just breaking changes. Good for fast-movi ## See also +- [Version skew protection](/deployment/version-skew-protection#chat-sessions) — pin a session to the deployment matching the app build that started it - [Lifecycle hooks](/ai-chat/lifecycle-hooks) — where `onTurnStart` and `onChatResume` fit in the turn cycle - [Recovery boot](/ai-chat/patterns/recovery-boot) — the sibling hook for mid-stream interruptions (does NOT fire on `requestUpgrade`) - [Database persistence](/ai-chat/patterns/database-persistence) — how continuations interact with session state diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx index da08f9a0473..d21bec858cd 100644 --- a/docs/ai-chat/reference.mdx +++ b/docs/ai-chat/reference.mdx @@ -490,7 +490,7 @@ Options for [`chat.headStart()`](/ai-chat/fast-starts#head-start), the warm-serv | `agentId` | `string` | required | The `chat.agent` / `chat.customAgent` id to hand off to | | `run` | `(args: HeadStartRunArgs) => Promise` | required | First-turn callback. Call `streamText` and spread `chat.toStreamTextOptions({ tools })` | | `idleTimeoutInSeconds` | `number` | `60` | How long the agent waits for the handover signal | -| `triggerConfig` | `Partial` | `undefined` | Run options (tags, queue, machine, maxAttempts, maxDuration, region, lockToVersion) for the auto-triggered handover-prepare run. The `chat:{chatId}` tag is prepended automatically | +| `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. diff --git a/docs/ai-chat/sessions.mdx b/docs/ai-chat/sessions.mdx index 041fd3d99ee..9232dc13fd9 100644 --- a/docs/ai-chat/sessions.mdx +++ b/docs/ai-chat/sessions.mdx @@ -111,7 +111,7 @@ const { id, runId, publicAccessToken, isCached } = await sessions.start({ | `type` | `string` | Free-form discriminator. `chat.agent` uses `"chat.agent"`. | | `externalId` | `string?` | Your stable identity. Cannot start with `session_` (reserved). | | `taskIdentifier` | `string` | Task this session triggers runs against. | -| `triggerConfig` | `SessionTriggerConfig` | Trigger options applied to every run: `tags`, `queue`, `machine`, `maxAttempts`, `idleTimeoutInSeconds`, `basePayload`. | +| `triggerConfig` | `SessionTriggerConfig` | Trigger options applied to every run: `tags`, `queue`, `machine`, `maxAttempts`, `maxDuration`, `region`, `idleTimeoutInSeconds`, `basePayload`, and the version pins `lockToVersion` / [`externalDeploymentId`](/deployment/version-skew-protection#chat-sessions). | | `tags` | `string[]?` | Up to 10 tags on the Session row (separate from `triggerConfig.tags`). | | `metadata` | `Record?` | Arbitrary JSON. | | `expiresAt` | `Date?` | Hard retention deadline. | diff --git a/docs/deployment/version-skew-protection.mdx b/docs/deployment/version-skew-protection.mdx index 2ecd783a5ef..3a9dc07c329 100644 --- a/docs/deployment/version-skew-protection.mdx +++ b/docs/deployment/version-skew-protection.mdx @@ -276,6 +276,66 @@ An empty or whitespace-only value counts as "not supplied" rather than an error, Batch triggers carry the id too. `batchTrigger` resolves it per item exactly as `trigger` does, and it survives the asynchronous materialisation of batch items — so a large batch triggered during a deploy waits and releases item by item, each pinned to the deployment its calling code came from. +## Chat sessions + +[Chat agents](/ai-chat/overview) are covered by the same mechanism, with one difference: the id belongs to the **session**, not to a single trigger. It is resolved wherever you start the session — your server action, your route handler, `sessions.start()` — using the same order of precedence as a task trigger, and stored on the session. Every run that session goes on to schedule carries it: the first run, each continuation after an idle suspend, and each recovery after a crash. + +That is what you want for a conversation. A chat started by one release of your app keeps talking to the agent build that release shipped with, however many turns and however many runs that takes. + +Chats need the same two halves as tasks, and no more: a deployment carrying an id, and an app that sends the same one (explicitly, through `TRIGGER_EXTERNAL_DEPLOYMENT_ID`, or through [automatic discovery](#automatic-discovery)). There is nothing chat-specific to switch on, so an app already pinning its task runs gets pinned chats with no code change. + +```ts +// app/actions.ts +"use server"; +import { chat } from "@trigger.dev/sdk/ai"; +import type { myChat } from "@/trigger/chat"; + +// No chat-specific setup: the id is discovered per call, exactly as it is for `trigger()`. +export const startChatSession = chat.createStartSessionAction("my-chat"); +``` + +Three things follow from the pin living on the session: + +- **Starting the session again refreshes it.** `sessions.start()` is idempotent on `chatId` and rewrites the stored config, so when your transport calls `startSession` after a redeploy, the *next* run picks up the new id. The turn already in flight finishes on the code it started on. +- **There is one pin per `chatId`.** If the same conversation is open in two tabs on two different releases of your app, whichever called `startSession` most recently sets the pin for both. +- **A parked chat is waiting, not broken.** A run pinned to a deployment that hasn't landed parks, and every message sent meanwhile is stored durably and delivered once the deployment arrives. Nothing is lost — but nothing answers either, so tell the user. Pass `pendingVersion` through your `startSession` callback and the transport emits a `run-pending-version` event: + +```tsx +const transport = useTriggerChatTransport({ + task: "my-chat", + accessToken: ({ chatId }) => mintChatAccessToken(chatId), + startSession: ({ chatId, clientData }) => startChatSession({ chatId, clientData }), + onEvent: (event) => { + if (event.type === "run-pending-version") setDeploying(true); + if (event.type === "first-chunk") setDeploying(false); + }, +}); +``` + +The event repeats on every message sent while the chat is parked, so a notice driven off it stays accurate. + +### Opting a chat out + +Pass `null` and that chat is never pinned, whatever the environment says: + +```ts +export const startChatSession = chat.createStartSessionAction("my-chat", { + triggerConfig: { externalDeploymentId: null }, +}); +``` + +Use this for a conversation that should always run on the current version — a long-lived support thread, say — while the rest of your chats stay pinned. To turn pinning off everywhere instead, set `TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION` to `0` and don't set `TRIGGER_EXTERNAL_DEPLOYMENT_ID`. + +### Escaping the pin from inside the agent + +[`chat.requestUpgrade()`](/ai-chat/patterns/version-upgrades) clears the session's external deployment id as part of the handoff, so the new run is free to land on the current version. Pass a target to move to a specific deployment instead: + +```ts +chat.requestUpgrade({ externalDeploymentId: clientData.commitSha }); +``` + +Either way the change is persisted on the session, so the next continuation doesn't fall back to the id the agent just rejected. `lockToVersion` is a separate, explicit pin and is never cleared — `requestUpgrade()` cannot escape it. + ## Waiting and expiry When a run arrives with an id that isn't deployed yet, it doesn't fail — it **waits**. This is the ordinary case, not an edge case: your app frequently goes live a few seconds before your task build finishes. diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index a90430953d4..6c155951486 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -254,6 +254,11 @@ export type IdempotencyKeyOptionsSchema = z.infer String(value)); const ExternalDeploymentId = z.preprocess((value) => { + // `null` is the opt-out sentinel callers write; treat it as absent, not a validation error. + if (value === null) { + return undefined; + } + if (typeof value !== "string") { return value; } @@ -1848,6 +1853,11 @@ export const SessionTriggerConfig = z.object({ maxDuration: z.number().int().positive().optional(), /** Pin every run to a specific worker version. Forwarded to `TaskRunOptions.lockToVersion`. */ lockToVersion: z.string().optional(), + /** + * Pin every run the session schedules to the deployment carrying this id, refreshed by each + * `sessions.start`. Independent of `lockToVersion`, which wins. + */ + externalDeploymentId: ExternalDeploymentId, /** Region to schedule runs in. Forwarded to `TaskRunOptions.region`. */ region: z.string().optional(), /** Convenience field surfaced to chat.agent via the wire payload. */ @@ -1923,6 +1933,11 @@ export const CreatedSessionResponseBody = SessionItem.extend({ publicAccessToken: z.string(), /** True if the session existed already (idempotent upsert), false if newly created. */ isCached: z.boolean(), + /** + * The session's live run is parked waiting for a deployment carrying its external deployment + * id. Messages sent meanwhile are durable. Optional, so older servers read as `false`. + */ + pendingVersion: z.boolean().optional(), }); export type CreatedSessionResponseBody = z.infer; @@ -1941,6 +1956,11 @@ export const EndAndContinueSessionRequestBody = z.object({ callingRunId: z.string(), /** Free-form label for the SessionRun audit row. e.g. `"upgrade"`. */ reason: z.string().max(64), + /** + * Re-pin the session to this id instead of clearing the pin. Only read when `reason` is + * `"upgrade"`. `lockToVersion` is never cleared. + */ + externalDeploymentId: ExternalDeploymentId, }); export type EndAndContinueSessionRequestBody = z.infer; diff --git a/packages/core/src/v3/schemas/sessionExternalDeploymentId.test.ts b/packages/core/src/v3/schemas/sessionExternalDeploymentId.test.ts new file mode 100644 index 00000000000..1fd77e0ddef --- /dev/null +++ b/packages/core/src/v3/schemas/sessionExternalDeploymentId.test.ts @@ -0,0 +1,101 @@ +import { describe, expect, it } from "vitest"; +import { EndAndContinueSessionRequestBody, SessionTriggerConfig } from "./api.js"; + +const baseConfig = { basePayload: {} }; + +describe("SessionTriggerConfig.externalDeploymentId", () => { + it("accepts an id alongside lockToVersion — the trigger path decides which governs", () => { + const result = SessionTriggerConfig.safeParse({ + ...baseConfig, + lockToVersion: "20260807.1", + externalDeploymentId: "commit-abc", + }); + + expect(result.success).toBe(true); + if (result.success) { + expect(result.data.lockToVersion).toBe("20260807.1"); + expect(result.data.externalDeploymentId).toBe("commit-abc"); + } + }); + + it("imposes no format", () => { + for (const id of ["a1b2c3", "v1.2.3", "release/2026-08-07", "refs/heads/main", "1"]) { + const result = SessionTriggerConfig.safeParse({ ...baseConfig, externalDeploymentId: id }); + expect(result.success, `expected ${id} to be accepted`).toBe(true); + } + }); + + it("trims surrounding whitespace", () => { + const result = SessionTriggerConfig.safeParse({ + ...baseConfig, + externalDeploymentId: " commit-abc ", + }); + + expect(result.success).toBe(true); + if (result.success) expect(result.data.externalDeploymentId).toBe("commit-abc"); + }); + + it("treats null as the opt-out rather than a 400", () => { + const result = SessionTriggerConfig.safeParse({ ...baseConfig, externalDeploymentId: null }); + + expect(result.success).toBe(true); + if (result.success) expect(result.data.externalDeploymentId).toBeUndefined(); + }); + + it.each(["", " "])("treats %j as absent rather than a 400", (value) => { + const result = SessionTriggerConfig.safeParse({ ...baseConfig, externalDeploymentId: value }); + + expect(result.success).toBe(true); + if (result.success) expect(result.data.externalDeploymentId).toBeUndefined(); + }); + + it("is optional", () => { + const result = SessionTriggerConfig.safeParse(baseConfig); + + expect(result.success).toBe(true); + if (result.success) expect(result.data.externalDeploymentId).toBeUndefined(); + }); + + it("accepts 128 characters and rejects 129", () => { + expect( + SessionTriggerConfig.safeParse({ ...baseConfig, externalDeploymentId: "a".repeat(128) }) + .success + ).toBe(true); + expect( + SessionTriggerConfig.safeParse({ ...baseConfig, externalDeploymentId: "a".repeat(129) }) + .success + ).toBe(false); + }); + + it("measures length after trimming", () => { + const result = SessionTriggerConfig.safeParse({ + ...baseConfig, + externalDeploymentId: ` ${"a".repeat(128)} `, + }); + + expect(result.success).toBe(true); + }); +}); + +describe("EndAndContinueSessionRequestBody.externalDeploymentId", () => { + it("normalizes an upgrade re-pin", () => { + const result = EndAndContinueSessionRequestBody.safeParse({ + callingRunId: "run_123", + reason: "upgrade", + externalDeploymentId: " commit-abc ", + }); + + expect(result.success).toBe(true); + if (result.success) expect(result.data.externalDeploymentId).toBe("commit-abc"); + }); + + it("is optional — an upgrade without one clears the pin", () => { + const result = EndAndContinueSessionRequestBody.safeParse({ + callingRunId: "run_123", + reason: "upgrade", + }); + + expect(result.success).toBe(true); + if (result.success) expect(result.data.externalDeploymentId).toBeUndefined(); + }); +}); diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index c241930323b..1f508af3e30 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -24,7 +24,6 @@ import { type RealtimeDefinedInputStream, type RealtimeDefinedStream, resourceCatalog, - type SessionTriggerConfig, SemanticInternalAttributes, SESSION_IN_EVENT_ID_HEADER, sessionStreams, @@ -101,11 +100,13 @@ type ToolCallOptions = { // pulled in transitively here never reach a client chunk. import { readFileInSkill, runBashInSkill } from "./agentSkillsRuntime.js"; import { ensureAiSdkTelemetry } from "./aiAutoTelemetry.js"; +import { withResolvedExternalDeploymentId } from "./externalDeploymentId.js"; import { type SessionHandle, type SessionPipeStreamOptions, sessions, type SessionSubscribeOptions, + type SessionTriggerConfigInput, } from "./sessions.js"; import { createTask } from "./shared.js"; import { markChatAgentRunForStreamsWarning } from "./streams.js"; @@ -2559,6 +2560,10 @@ const chatResolvedToolsKey = locals.create("chat.resolvedTools"); /** @internal Flag set by `chat.requestUpgrade()` to exit the loop after the current turn. */ const chatUpgradeRequestedKey = locals.create("chat.upgradeRequested"); +/** @internal Target for the upgrade handoff, set by `chat.requestUpgrade({ externalDeploymentId })`. */ +const chatUpgradeExternalDeploymentIdKey = locals.create( + "chat.upgradeExternalDeploymentId" +); /** * @internal Flag set by `chat.endRun()` to exit the loop after the current @@ -8701,8 +8706,12 @@ function isStopped(): boolean { * }); * ``` */ -function requestUpgrade(): void { +function requestUpgrade(options?: { externalDeploymentId?: string }): void { locals.set(chatUpgradeRequestedKey, true); + + // Without a target the handoff clears the session's pin; with one it re-pins to that deployment. + const target = options?.externalDeploymentId?.trim(); + if (target) locals.set(chatUpgradeExternalDeploymentIdKey, target); } /** @@ -10303,7 +10312,7 @@ export type CreateChatStartSessionActionOptions = { * Default trigger config used when starting a new session for a chat. * Per-call `params.triggerConfig` shallow-merges on top. */ - triggerConfig?: Partial; + triggerConfig?: Partial; /** * Override the Trigger.dev API base URL. String applies to both * `/api/v1/sessions` and `/api/v1/auth/jwt/claims`; function picks per @@ -10347,7 +10356,7 @@ export type ChatStartSessionParams = { * `chat.agent`: anything beyond `chatId`/`messages`/`trigger`/`metadata`, * which the runtime injects automatically). */ - triggerConfig?: Partial; + triggerConfig?: Partial; /** * Opaque session-level metadata stored on the Session row. Separate from * the per-turn `clientData` above. Use this when you want to attach @@ -10370,6 +10379,11 @@ export type ChatStartSessionResult = { runId: string; /** Session friendlyId — informational. */ sessionId: string; + /** + * The session's run is parked waiting for its deployment. Messages sent meanwhile are durable + * and delivered once it lands; surface this so the wait reads as a deploy in progress. + */ + pendingVersion?: boolean; }; /** @@ -10449,7 +10463,14 @@ function createChatStartSessionAction( const idleTimeoutInSeconds = params.triggerConfig?.idleTimeoutInSeconds ?? options?.triggerConfig?.idleTimeoutInSeconds; - const triggerConfig: SessionTriggerConfig = { + // Only `undefined` means "not supplied": a per-call `null` (opt out) has to beat a pinning + // action default, which neither truthiness nor `??` would allow. + const externalDeploymentId = + params.triggerConfig?.externalDeploymentId !== undefined + ? params.triggerConfig.externalDeploymentId + : options?.triggerConfig?.externalDeploymentId; + + const triggerConfig: SessionTriggerConfigInput = { basePayload: { messages: [], trigger: "preload", @@ -10476,6 +10497,7 @@ function createChatStartSessionAction( params.triggerConfig?.lockToVersion ?? options?.triggerConfig?.lockToVersion, } : {}), + ...(externalDeploymentId !== undefined ? { externalDeploymentId } : {}), ...(idleTimeoutInSeconds !== undefined ? { idleTimeoutInSeconds } : {}), }; @@ -10491,7 +10513,12 @@ function createChatStartSessionAction( const fetchOverride = options?.fetch; const hasOverride = baseURLOption !== undefined || fetchOverride !== undefined; - const created: { id: string; runId: string; publicAccessToken: string } = hasOverride + const created: { + id: string; + runId: string; + publicAccessToken: string; + pendingVersion?: boolean; + } = hasOverride ? await callSessionsCreateWithOverride({ chatId: params.chatId, body: startBody, @@ -10526,6 +10553,7 @@ function createChatStartSessionAction( publicAccessToken, runId: created.runId, sessionId: created.id, + ...(created.pendingVersion ? { pendingVersion: true } : {}), }; }; } @@ -10562,12 +10590,17 @@ async function callSessionsCreateWithOverride(args: { type: "chat.agent"; externalId: string; taskIdentifier: string; - triggerConfig: SessionTriggerConfig; + triggerConfig: SessionTriggerConfigInput; metadata?: Record; }; baseURLOption: string | ChatStartSessionBaseURLResolver | undefined; fetchOverride: ChatStartSessionFetchOverride | undefined; -}): Promise<{ id: string; runId: string; publicAccessToken: string }> { +}): Promise<{ + id: string; + runId: string; + publicAccessToken: string; + pendingVersion?: boolean; +}> { const accessToken = apiClientManager.accessToken; if (!accessToken) { throw new Error( @@ -10579,7 +10612,8 @@ async function callSessionsCreateWithOverride(args: { const init: RequestInit = { method: "POST", headers: overrideRequestHeaders(accessToken), - body: JSON.stringify(args.body), + // This path bypasses `sessions.start`, so it resolves the pin itself. + body: JSON.stringify(withResolvedExternalDeploymentId(args.body)), }; const response = args.fetchOverride ? await args.fetchOverride(url, init, ctx) @@ -10588,7 +10622,12 @@ async function callSessionsCreateWithOverride(args: { const text = await response.text().catch(() => ""); throw new Error(`sessions.start failed: ${response.status} ${text}`); } - const json = (await response.json()) as { id: string; runId: string; publicAccessToken: string }; + const json = (await response.json()) as { + id: string; + runId: string; + publicAccessToken: string; + pendingVersion?: boolean; + }; return json; } @@ -10885,9 +10924,11 @@ async function writeUpgradeRequiredChunk(): Promise { if (chatId && callingRunId) { const apiClient = apiClientManager.clientOrThrow(); try { + const externalDeploymentId = locals.get(chatUpgradeExternalDeploymentIdKey); await apiClient.endAndContinueSession(chatId, { callingRunId, reason: "upgrade", + ...(externalDeploymentId ? { externalDeploymentId } : {}), }); } catch (error) { // Non-fatal: the next `.in/append` re-triggers via the probe. diff --git a/packages/trigger-sdk/src/v3/chat-client.ts b/packages/trigger-sdk/src/v3/chat-client.ts index 35cfd0b6af9..9367207c8ae 100644 --- a/packages/trigger-sdk/src/v3/chat-client.ts +++ b/packages/trigger-sdk/src/v3/chat-client.ts @@ -16,7 +16,7 @@ * ``` */ -import type { SessionTriggerConfig, Task } from "@trigger.dev/core/v3"; +import type { Task } from "@trigger.dev/core/v3"; import type { ModelMessage, UIMessage, UIMessageChunk } from "ai"; // `readUIMessageStream` is a runtime value — via the ESM/CJS shim so the CJS // build can `require` ESM-only `ai@7` (see ../imports/ai-runtime.ts). @@ -29,7 +29,7 @@ import { } from "@trigger.dev/core/v3"; import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js"; import { slimSubmitMessageForWire } from "./ai-shared.js"; -import { sessions } from "./sessions.js"; +import { sessions, type SessionTriggerConfigInput } from "./sessions.js"; // ─── Type inference ──────────────────────────────────────────────── @@ -105,7 +105,7 @@ export type AgentChatOptions = { * Default trigger config used when starting a new session for this * chat. Folded into `sessions.start({...triggerConfig})` body. */ - triggerConfig?: SessionTriggerConfig; + triggerConfig?: SessionTriggerConfigInput; /** * Override the Trigger.dev API base URL for the chat's `.in/append` and * `.out` SSE endpoints. String form applies to both; pass a function to @@ -306,7 +306,7 @@ export class AgentChat { private readonly chatId: string; private readonly streamTimeoutSeconds: number; private readonly clientData: Record | undefined; - private readonly triggerConfigDefault: SessionTriggerConfig | undefined; + private readonly triggerConfigDefault: SessionTriggerConfigInput | undefined; private readonly onTriggered: AgentChatOptions["onTriggered"]; private readonly onTurnComplete: AgentChatOptions["onTurnComplete"]; private readonly baseURLResolver: AgentChatBaseURLResolver; @@ -656,7 +656,7 @@ export class AgentChat { const idleTimeoutInSeconds = options?.idleTimeoutInSeconds ?? this.triggerConfigDefault?.idleTimeoutInSeconds; - const triggerConfig: SessionTriggerConfig = { + const triggerConfig: SessionTriggerConfigInput = { basePayload: { // `trigger: "preload"` mirrors the browser-mediated // `chat.createStartSessionAction` shape so the agent runtime fires @@ -675,6 +675,17 @@ export class AgentChat { ...(this.triggerConfigDefault?.maxAttempts !== undefined ? { maxAttempts: this.triggerConfigDefault.maxAttempts } : {}), + ...(this.triggerConfigDefault?.maxDuration !== undefined + ? { maxDuration: this.triggerConfigDefault.maxDuration } + : {}), + ...(this.triggerConfigDefault?.region ? { region: this.triggerConfigDefault.region } : {}), + ...(this.triggerConfigDefault?.lockToVersion + ? { lockToVersion: this.triggerConfigDefault.lockToVersion } + : {}), + // Not truthiness: `null` opts out and must reach the resolver in `sessions.start`. + ...(this.triggerConfigDefault?.externalDeploymentId !== undefined + ? { externalDeploymentId: this.triggerConfigDefault.externalDeploymentId } + : {}), ...(idleTimeoutInSeconds !== undefined ? { idleTimeoutInSeconds } : {}), }; diff --git a/packages/trigger-sdk/src/v3/chat-server.ts b/packages/trigger-sdk/src/v3/chat-server.ts index 5e48e3b24b6..98bf7b99854 100644 --- a/packages/trigger-sdk/src/v3/chat-server.ts +++ b/packages/trigger-sdk/src/v3/chat-server.ts @@ -60,7 +60,6 @@ import { TRIGGER_CONTROL_SUBTYPE, apiClientManager, type ApiClientConfiguration, - type SessionTriggerConfig, } from "@trigger.dev/core/v3"; // Runtime VALUES via the ESM/CJS shim so the CJS build can `require` ESM-only // `ai@7` (see ../imports/ai-runtime.ts). @@ -71,6 +70,8 @@ import { } from "../imports/ai-runtime.js"; import type { FinishReason, ModelMessage, Tool, UIMessage, UIMessageChunk } from "ai"; import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js"; +import { withResolvedExternalDeploymentId } from "./externalDeploymentId.js"; +import type { SessionTriggerConfigInput } from "./sessions.js"; // `StreamTextResult` is defined locally rather than imported from `ai`: its // generic arity diverged (v6 `StreamTextResult`, v7 @@ -200,7 +201,7 @@ export type HeadStartHandlerOptions> = { * tags, queue, machine, etc. Mirrors `chat.createStartSessionAction`. * The `chat:{chatId}` tag is prepended automatically. */ - triggerConfig?: Partial; + triggerConfig?: Partial; /** * API client config (base URL + access token) for creating the session * and triggering the agent run. When set, the handler runs under this @@ -224,7 +225,7 @@ export type StartHeadStartOptions> = { /** Seconds the agent run waits for the handover signal before exiting. Default 60. */ idleTimeoutInSeconds?: number; /** Run options for the auto-triggered `handover-prepare` run (tags, queue, machine, …). */ - triggerConfig?: Partial; + triggerConfig?: Partial; /** API client config for session creation + trigger when the agent lives in another project/env. */ apiClient?: ApiClientConfiguration; /** Metadata merged into the run's wire payload (auth tokens, context, …). Never sent to the browser. */ @@ -401,7 +402,7 @@ export const chat = { req: Request; agentId: string; idleTimeoutInSeconds?: number; - triggerConfig?: Partial; + triggerConfig?: Partial; }): Promise { return (async () => { const session = await openHandoverSession({ @@ -505,7 +506,7 @@ async function openHandoverSession(opts: { wirePayload: ChatTaskWirePayload; agentId: string; idleTimeoutInSeconds?: number; - triggerConfig?: Partial; + triggerConfig?: Partial; /** Request-lifecycle signal on the HTTP path; omitted on the detached path. */ requestSignal?: AbortSignal; }): Promise { @@ -532,7 +533,7 @@ async function openHandoverSession(opts: { const userTags = opts.triggerConfig?.tags ?? []; const tags = [`chat:${chatId}`, ...userTags].slice(0, 5); - const triggerConfig: SessionTriggerConfig = { + const triggerConfig: SessionTriggerConfigInput = { basePayload: { ...(opts.triggerConfig?.basePayload ?? {}), ...wirePayload, @@ -553,6 +554,10 @@ async function openHandoverSession(opts: { ...(opts.triggerConfig?.lockToVersion ? { lockToVersion: opts.triggerConfig.lockToVersion } : {}), + // Not truthiness: `null` opts this chat out of pinning and must reach the resolver. + ...(opts.triggerConfig?.externalDeploymentId !== undefined + ? { externalDeploymentId: opts.triggerConfig.externalDeploymentId } + : {}), idleTimeoutInSeconds, }; @@ -569,12 +574,15 @@ async function openHandoverSession(opts: { // run to be there to consume it. The added latency (~one round trip // to the control plane) is bounded; the agent's compute boot still // overlaps with LLM TTFB. - const created = await apiClient.createSession({ - type: "chat.agent", - externalId: chatId, - taskIdentifier: opts.agentId, - triggerConfig, - }); + // Bypasses `sessions.start`, so it resolves the pin itself. + const created = await apiClient.createSession( + withResolvedExternalDeploymentId({ + type: "chat.agent", + externalId: chatId, + taskIdentifier: opts.agentId, + triggerConfig, + }) + ); const sessionPublicAccessToken = created.publicAccessToken; // Combined abort signal: request lifecycle OR an internal timeout diff --git a/packages/trigger-sdk/src/v3/chat.test.ts b/packages/trigger-sdk/src/v3/chat.test.ts index 452c8090b70..d0dbd61e2a3 100644 --- a/packages/trigger-sdk/src/v3/chat.test.ts +++ b/packages/trigger-sdk/src/v3/chat.test.ts @@ -508,6 +508,111 @@ describe("TriggerChatTransport", () => { }); }); + describe("run-pending-version", () => { + it("emits on send when the append response says the run is parked", async () => { + global.fetch = vi.fn().mockImplementation(async (url: string | URL) => { + const urlStr = typeof url === "string" ? url : url.toString(); + if (isSessionStreamAppendUrl(urlStr)) { + return new Response(JSON.stringify({ ok: true, seq: 1, pendingVersion: true }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse(); + throw new Error(`Unexpected URL: ${urlStr}`); + }); + + const events: ChatTransportEvent[] = []; + const transport = new TriggerChatTransport({ + task: "my-chat-task", + accessToken: () => "pat", + onEvent: (e) => events.push(e), + sessions: { "chat-parked": { publicAccessToken: "p" } }, + }); + + const stream = await transport.sendMessages({ + trigger: "submit-message", + chatId: "chat-parked", + messageId: "m1", + messages: [createUserMessage("Hello")], + abortSignal: undefined, + }); + await drainChunks(stream); + + expect(events).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "run-pending-version", + chatId: "chat-parked", + source: "send", + }), + ]) + ); + }); + + it("stays quiet on an ordinary append response", async () => { + global.fetch = vi.fn().mockImplementation(async (url: string | URL) => { + const urlStr = typeof url === "string" ? url : url.toString(); + if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse(); + if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse(); + throw new Error(`Unexpected URL: ${urlStr}`); + }); + + const events: ChatTransportEvent[] = []; + const transport = new TriggerChatTransport({ + task: "my-chat-task", + accessToken: () => "pat", + onEvent: (e) => events.push(e), + sessions: { "chat-normal": { publicAccessToken: "p" } }, + }); + + const stream = await transport.sendMessages({ + trigger: "submit-message", + chatId: "chat-normal", + messageId: "m1", + messages: [createUserMessage("Hello")], + abortSignal: undefined, + }); + await drainChunks(stream); + + expect(events.some((e) => e.type === "run-pending-version")).toBe(false); + }); + + it("emits on start when startSession reports a parked run", async () => { + const events: ChatTransportEvent[] = []; + const transport = new TriggerChatTransport({ + task: "my-chat-task", + accessToken: () => "should-not-be-called", + onEvent: (e) => events.push(e), + startSession: vi.fn().mockResolvedValue({ publicAccessToken: "pat", pendingVersion: true }), + }); + + await transport.start("chat-start-parked"); + + expect(events).toEqual([ + expect.objectContaining({ + type: "run-pending-version", + chatId: "chat-start-parked", + source: "start", + }), + ]); + }); + + it("stays quiet when startSession reports nothing", async () => { + const events: ChatTransportEvent[] = []; + const transport = new TriggerChatTransport({ + task: "my-chat-task", + accessToken: () => "should-not-be-called", + onEvent: (e) => events.push(e), + startSession: vi.fn().mockResolvedValue({ publicAccessToken: "pat" }), + }); + + await transport.start("chat-start-normal"); + + expect(events.some((e) => e.type === "run-pending-version")).toBe(false); + }); + }); + describe("sendMessages", () => { it("posts the user message to .in/append and streams chunks from .out", async () => { const requests: Array<{ url: string; init?: RequestInit }> = []; diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts index a7c7125575e..59421922682 100644 --- a/packages/trigger-sdk/src/v3/chat.ts +++ b/packages/trigger-sdk/src/v3/chat.ts @@ -225,6 +225,18 @@ export type ChatTransportSendSource = * stream fails unrecoverably. */ export type ChatTransportEvent = + | { + /** + * The chat's run is parked waiting for its deployment (version skew protection). Messages + * sent meanwhile are durable and drain when it lands, but nothing answers until then. + * Re-emitted on every send while parked; clear the notice on `first-chunk`. + */ + type: "run-pending-version"; + chatId: string; + timestamp: number; + /** Whether we learned this from starting the session or from sending a message. */ + source: "start" | "send"; + } | { type: "message-sent"; chatId: string; @@ -416,6 +428,11 @@ export type StartSessionParams = { export type StartSessionResult = { /** Session-scoped PAT — `read:sessions:{chatId} + write:sessions:{chatId}`. */ publicAccessToken: string; + /** + * Pass through `pendingVersion` from `chat.createStartSessionAction` (or `POST + * /api/v1/sessions`) and the transport emits `run-pending-version`. + */ + pendingVersion?: boolean; }; /** @@ -1491,12 +1508,21 @@ export class TriggerChatTransport implements ChatTransport { ); } - const { publicAccessToken } = await this.resolveStartSession({ + const { publicAccessToken, pendingVersion } = await this.resolveStartSession({ taskId: this.taskId, chatId, clientData: (this.defaultMetadata ?? {}) as Record, }); + if (pendingVersion) { + this.emitEvent({ + type: "run-pending-version", + chatId, + timestamp: Date.now(), + source: "start", + }); + } + const state: ChatSessionState = { publicAccessToken, isStreaming: false, @@ -1553,7 +1579,19 @@ export class TriggerChatTransport implements ChatTransport { } // The appended record's `.in` seq, for correlating the response stream to // this send. Omitted by older webapps / a lost idempotency claim. - const data = (await response.json().catch(() => undefined)) as { seq?: unknown } | undefined; + const data = (await response.json().catch(() => undefined)) as + | { seq?: unknown; pendingVersion?: unknown } + | undefined; + + if (data?.pendingVersion === true) { + this.emitEvent({ + type: "run-pending-version", + chatId, + timestamp: Date.now(), + source: "send", + }); + } + return typeof data?.seq === "number" ? data.seq : undefined; } @@ -1609,11 +1647,19 @@ export class TriggerChatTransport implements ChatTransport { "TriggerChatTransport: session not found and no `startSession` configured to recreate it. The stored session state for this chat may be stale (e.g. created in a different environment) — provide `startSession` or clear the stored session so a fresh one can be created." ); } - const { publicAccessToken } = await this.resolveStartSession({ + const { publicAccessToken, pendingVersion } = await this.resolveStartSession({ taskId: this.taskId, chatId, clientData: (this.defaultMetadata ?? {}) as Record, }); + if (pendingVersion) { + this.emitEvent({ + type: "run-pending-version", + chatId, + timestamp: Date.now(), + source: "start", + }); + } state.publicAccessToken = publicAccessToken; state.lastEventId = undefined; state.isStreaming = false; diff --git a/packages/trigger-sdk/src/v3/externalDeploymentId.ts b/packages/trigger-sdk/src/v3/externalDeploymentId.ts new file mode 100644 index 00000000000..5f41a23b5eb --- /dev/null +++ b/packages/trigger-sdk/src/v3/externalDeploymentId.ts @@ -0,0 +1,53 @@ +import { + apiClientManager, + getEnvVar, + resolveExternalDeploymentId, + sdkScope, + type SessionTriggerConfig, +} from "@trigger.dev/core/v3"; + +/** Reads an env var unless the scope opted out of ambient context (`inheritContext: false`). */ +export function scopedEnvVar(name: string): string | undefined { + const scope = sdkScope.getStore(); + if (scope && !scope.inheritContext) return undefined; + return getEnvVar(name); +} + +/** + * Precedence: an explicit value, the client config, `TRIGGER_EXTERNAL_DEPLOYMENT_ID`, then + * platform/CI commit vars when automatic skew protection is on. `null` is an explicit opt-out. + */ +export function resolveTriggerExternalDeploymentId(explicit?: string | null): string | undefined { + if (explicit === null) return undefined; + + return resolveExternalDeploymentId({ + explicit, + clientConfig: apiClientManager.externalDeploymentId, + read: scopedEnvVar, + }); +} + +/** A session trigger config as callers write it: the pin is optional, and `null` opts out. */ +type TriggerConfigInput = Omit & { + externalDeploymentId?: string | null; +}; + +/** + * Fill in `triggerConfig.externalDeploymentId` on an outgoing session-create body, at each point + * one leaves the SDK. Discovery has to happen here rather than server-side: the commit SHA lives + * in the calling application's runtime, so the caller's build is what selects the agent version. + */ +export function withResolvedExternalDeploymentId< + TBody extends { triggerConfig: TriggerConfigInput }, +>(body: TBody): TBody & { triggerConfig: SessionTriggerConfig } { + const resolved = resolveTriggerExternalDeploymentId(body.triggerConfig.externalDeploymentId); + const { externalDeploymentId: _omit, ...rest } = body.triggerConfig; + + return { + ...body, + triggerConfig: { + ...rest, + ...(resolved ? { externalDeploymentId: resolved } : {}), + }, + }; +} diff --git a/packages/trigger-sdk/src/v3/sessionExternalDeploymentId.test.ts b/packages/trigger-sdk/src/v3/sessionExternalDeploymentId.test.ts new file mode 100644 index 00000000000..74573a3f7b5 --- /dev/null +++ b/packages/trigger-sdk/src/v3/sessionExternalDeploymentId.test.ts @@ -0,0 +1,201 @@ +import { apiClientManager, type CreatedSessionResponseBody } from "@trigger.dev/core/v3"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { chat } from "./ai.js"; +import { + __setSessionOpenImplForTests, + __setSessionStartImplForTests, + SessionHandle, + sessions, +} from "./sessions.js"; + +const ENV_KEYS = [ + "TRIGGER_EXTERNAL_DEPLOYMENT_ID", + "TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION", + "VERCEL_GIT_COMMIT_SHA", + "GITHUB_SHA", +] as const; + +let capturedPin: string | undefined; + +function installStartFixture() { + __setSessionStartImplForTests(async (body): Promise => { + capturedPin = body.triggerConfig.externalDeploymentId; + return { + id: "session_fixture", + externalId: body.externalId ?? null, + type: body.type, + taskIdentifier: body.taskIdentifier, + triggerConfig: body.triggerConfig, + currentRunId: "run_fixture", + tags: body.triggerConfig.tags ?? [], + metadata: body.metadata ?? null, + closedAt: null, + closedReason: null, + expiresAt: null, + createdAt: new Date(), + updatedAt: new Date(), + runId: "run_fixture", + publicAccessToken: "tr_pat_fixture", + isCached: false, + }; + }); + __setSessionOpenImplForTests(() => new SessionHandle("session_fixture")); +} + +function setGlobalConfig(externalDeploymentId?: string) { + apiClientManager.setGlobalAPIClientConfiguration({ + baseURL: "https://example.invalid", + accessToken: "tr_test_secret", + ...(externalDeploymentId ? { externalDeploymentId } : {}), + }); +} + +async function startRaw(triggerConfig: Parameters[0]["triggerConfig"]) { + await sessions.start({ + type: "chat.agent", + externalId: "chat-1", + taskIdentifier: "fake-chat", + triggerConfig, + }); + return capturedPin; +} + +beforeEach(() => { + for (const key of ENV_KEYS) delete process.env[key]; + setGlobalConfig(); + installStartFixture(); +}); + +afterEach(() => { + __setSessionStartImplForTests(undefined); + __setSessionOpenImplForTests(undefined); + capturedPin = undefined; + for (const key of ENV_KEYS) delete process.env[key]; +}); + +describe("sessions.start — external deployment id discovery", () => { + it("sends nothing when there is nothing to discover", async () => { + expect(await startRaw({ basePayload: {} })).toBeUndefined(); + }); + + it("discovers TRIGGER_EXTERNAL_DEPLOYMENT_ID", async () => { + process.env.TRIGGER_EXTERNAL_DEPLOYMENT_ID = "from-env"; + + expect(await startRaw({ basePayload: {} })).toBe("from-env"); + }); + + it("prefers an explicit id over the environment", async () => { + process.env.TRIGGER_EXTERNAL_DEPLOYMENT_ID = "from-env"; + + expect(await startRaw({ basePayload: {}, externalDeploymentId: "explicit" })).toBe("explicit"); + }); + + it("prefers the client config over the environment", async () => { + process.env.TRIGGER_EXTERNAL_DEPLOYMENT_ID = "from-env"; + setGlobalConfig("from-client-config"); + + expect(await startRaw({ basePayload: {} })).toBe("from-client-config"); + }); + + it("ignores platform commit vars unless automatic protection is on", async () => { + process.env.VERCEL_GIT_COMMIT_SHA = "commit-sha"; + + expect(await startRaw({ basePayload: {} })).toBeUndefined(); + }); + + it("discovers the platform commit var when automatic protection is on", async () => { + process.env.TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION = "1"; + process.env.VERCEL_GIT_COMMIT_SHA = "commit-sha"; + + expect(await startRaw({ basePayload: {} })).toBe("commit-sha"); + }); + + it("treats null as an opt-out, outranking every source", async () => { + process.env.TRIGGER_EXTERNAL_DEPLOYMENT_ID = "from-env"; + process.env.TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION = "1"; + process.env.VERCEL_GIT_COMMIT_SHA = "commit-sha"; + setGlobalConfig("from-client-config"); + + expect(await startRaw({ basePayload: {}, externalDeploymentId: null })).toBeUndefined(); + }); + + it("normalizes a whitespace-only id to absent", async () => { + expect(await startRaw({ basePayload: {}, externalDeploymentId: " " })).toBeUndefined(); + }); + + it("skips an over-long id rather than sending it", async () => { + expect( + await startRaw({ basePayload: {}, externalDeploymentId: "a".repeat(129) }) + ).toBeUndefined(); + }); +}); + +describe("chat.createStartSessionAction — external deployment id", () => { + it("discovers from the environment", async () => { + process.env.TRIGGER_EXTERNAL_DEPLOYMENT_ID = "from-env"; + const start = chat.createStartSessionAction("fake-chat"); + + await start({ chatId: "chat-1" }); + + expect(capturedPin).toBe("from-env"); + }); + + it("prefers the per-call id over the action default", async () => { + const start = chat.createStartSessionAction("fake-chat", { + triggerConfig: { externalDeploymentId: "action-default" }, + }); + + await start({ chatId: "chat-1", triggerConfig: { externalDeploymentId: "per-call" } }); + + expect(capturedPin).toBe("per-call"); + }); + + it("prefers the action default over the environment", async () => { + process.env.TRIGGER_EXTERNAL_DEPLOYMENT_ID = "from-env"; + const start = chat.createStartSessionAction("fake-chat", { + triggerConfig: { externalDeploymentId: "action-default" }, + }); + + await start({ chatId: "chat-1" }); + + expect(capturedPin).toBe("action-default"); + }); + + it("honours a per-call null opt-out over a pinning action default", async () => { + const start = chat.createStartSessionAction("fake-chat", { + triggerConfig: { externalDeploymentId: "action-default" }, + }); + + await start({ chatId: "chat-1", triggerConfig: { externalDeploymentId: null } }); + + expect(capturedPin).toBeUndefined(); + }); + + it("still resolves inside an apiClient-scoped action", async () => { + // `apiClient` re-enters via `runWithConfig`, which inherits context — so env discovery + // must survive the scope rather than being suppressed like it is for `new TriggerClient`. + process.env.TRIGGER_EXTERNAL_DEPLOYMENT_ID = "from-env"; + const start = chat.createStartSessionAction("fake-chat", { + apiClient: { baseURL: "https://scoped.invalid", accessToken: "tr_scoped_secret" }, + }); + + await start({ chatId: "chat-1" }); + + expect(capturedPin).toBe("from-env"); + }); + + it("carries an id set on the scoped apiClient config", async () => { + const start = chat.createStartSessionAction("fake-chat", { + apiClient: { + baseURL: "https://scoped.invalid", + accessToken: "tr_scoped_secret", + externalDeploymentId: "from-scoped-config", + }, + }); + + await start({ chatId: "chat-1" }); + + expect(capturedPin).toBe("from-scoped-config"); + }); +}); diff --git a/packages/trigger-sdk/src/v3/sessions.ts b/packages/trigger-sdk/src/v3/sessions.ts index 8a01f8293c4..9bafc25cfb3 100644 --- a/packages/trigger-sdk/src/v3/sessions.ts +++ b/packages/trigger-sdk/src/v3/sessions.ts @@ -17,6 +17,7 @@ import type { PipeStreamOptions, PipeStreamResult, RetrieveSessionResponseBody, + SessionTriggerConfig, StreamWriteResult, UpdateSessionRequestBody, WriterStreamOptions, @@ -39,6 +40,7 @@ import { writeSessionControlRecord, } from "@trigger.dev/core/v3"; import { conditionallyImportAndParsePacket } from "@trigger.dev/core/v3/utils/ioSerialization"; +import { withResolvedExternalDeploymentId } from "./externalDeploymentId.js"; import { tracer } from "./tracer.js"; export type { @@ -51,6 +53,19 @@ export type { UpdateSessionRequestBody, }; +/** + * `SessionTriggerConfig` as callers supply it. `externalDeploymentId` is normally discovered from + * the environment, so passing it is optional; pass `null` to opt this chat out of version pinning. + */ +export type SessionTriggerConfigInput = Omit & { + externalDeploymentId?: string | null; +}; + +/** {@link CreateSessionRequestBody} with the caller-facing trigger config. */ +export type CreateSessionInput = Omit & { + triggerConfig: SessionTriggerConfigInput; +}; + export const sessions = { start: startSession, retrieve: retrieveSession, @@ -97,9 +112,12 @@ export function __setSessionStartImplForTests(impl: SessionStartImpl | undefined * Two browser tabs of the same chat converge to one session. */ function startSession( - body: CreateSessionRequestBody, + input: CreateSessionInput, requestOptions?: ApiRequestOptions ): ApiPromise { + // Resolved before the test hook so fixtures observe the body that would go on the wire. + const body = withResolvedExternalDeploymentId(input); + if (sessionStartImpl) { const result = sessionStartImpl(body); return Promise.resolve(result) as ApiPromise; diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts index 69b51f0ed3c..2a022dff3c0 100644 --- a/packages/trigger-sdk/src/v3/shared.ts +++ b/packages/trigger-sdk/src/v3/shared.ts @@ -17,18 +17,15 @@ import { createErrorTaskError, defaultRetryOptions, flattenIdempotencyKey, - getEnvVar, getIdempotencyKeyOptions, getSchemaParseFn, lifecycleHooks, makeIdempotencyKey, packetRequiresOffloading, - resolveExternalDeploymentId, parsePacket, RateLimitError, resourceCatalog, runtime, - sdkScope, SemanticInternalAttributes, stringifyIO, SubtaskUnwrapError, @@ -91,6 +88,7 @@ import { type TriggerApiRequestOptions, type TriggerOptions, } from "@trigger.dev/core/v3"; +import { resolveTriggerExternalDeploymentId, scopedEnvVar } from "./externalDeploymentId.js"; import { tracer } from "./tracer.js"; export type { @@ -120,20 +118,6 @@ export { SubtaskUnwrapError, TaskRunPromise }; export type Context = TaskRunContext; -function scopedEnvVar(name: string): string | undefined { - const scope = sdkScope.getStore(); - if (scope && !scope.inheritContext) return undefined; - return getEnvVar(name); -} - -function resolveTriggerExternalDeploymentId(explicit?: string): string | undefined { - return resolveExternalDeploymentId({ - explicit, - clientConfig: apiClientManager.externalDeploymentId, - read: scopedEnvVar, - }); -} - export function queue(options: QueueOptions): Queue { resourceCatalog.registerQueueMetadata(options);