feat(core,sdk,webapp): version skew protection for chat.agent sessions

A chat session now carries an external deployment id, so every run it
schedules lands on the deployment that matched the app build that
started the conversation: the first run, each continuation after an idle
suspend, and each recovery after a crash.

Public SDK surface:

- `triggerConfig.externalDeploymentId`, on every entry point that starts
  a session (`sessions.start`, `chat.createStartSessionAction`,
  `chat.headStart`, `chat.handover`, `AgentChat`). Normally omitted: it
  is discovered wherever the session is started, with the same
  precedence `trigger()` uses. `null` opts one chat out.
- `pendingVersion` on the session-create and `.in/append` responses, on
  `ChatStartSessionResult` and on `StartSessionResult`, plus a new
  `run-pending-version` transport event, so a chat waiting on a
  deployment that is still building can say so instead of appearing to
  stall.
- `chat.requestUpgrade({ externalDeploymentId })`. Called without a
  target it now clears the session's pin, which is what makes upgrading
  away from a pinned version possible at all, and the cleared pin is
  persisted so the next continuation cannot bounce back.
- `SessionTriggerConfigInput` and `CreateSessionInput`, the caller-facing
  forms of the trigger config and the create body.

Parking is the right failure mode here, and an improvement on
`lockToVersion`, which throws on session create and is swallowed on the
append path, leaving the chat hung with no run at all. `PENDING_VERSION`
is non-final, so a parked run is reused rather than re-triggered and
appended messages stay durable until the deployment lands.

`lockToVersion` is deliberately untouched: it still wins where both are
set, and `requestUpgrade()` still cannot escape it.

Author: 0ski

.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 discovered 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. There is no chat-specific setup.
+
+```ts
+// Opt a single chat out of pinning:
+export const startChatSession = chat.createStartSessionAction<typeof myChat>("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.

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

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<CreatedSessionResponseBody>(responseBody, {

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 }
+    );
   }
 );
 

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<string, unknown>;
-}): 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<string, unknown>;
+  /** 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<Swap
     trigger: undefined,
   };
 
-  const config = SessionTriggerConfigSchema.parse(session.triggerConfig);
+  const storedConfig = SessionTriggerConfigSchema.parse(session.triggerConfig);
+
+  // The upgrade's pin is persisted in the claim below, not applied to this run alone: the next
+  // continuation re-reads the stored config. `lockToVersion` is deliberately untouched.
+  const config =
+    reason === "upgrade"
+      ? { ...storedConfig, externalDeploymentId: params.externalDeploymentId ?? undefined }
+      : storedConfig;
+
   const triggered = await triggerSessionRun({
     session,
     config,
@@ -430,6 +470,7 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise<Swap
     data: {
       currentRunId: triggered.id,
       currentRunVersion: { increment: 1 },
+      ...(reason === "upgrade" ? { triggerConfig: config as Prisma.InputJsonValue } : {}),
     },
   });
 
@@ -446,7 +487,11 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise<Swap
           error,
         });
       });
-    return { runId: triggered.id, swapped: true };
+    return {
+      runId: triggered.id,
+      swapped: true,
+      pendingVersion: isPendingVersionStatus(triggered.status),
+    };
   }
 
   // Lost the race — someone else already swapped to a new run. Cancel
@@ -480,9 +525,16 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise<Swap
     );
   }
 
+  const winner = await runStore.findRun(
+    { id: fresh.currentRunId },
+    { select: { status: true } },
+    prisma
+  );
+
   return {
     runId: fresh.currentRunId,
     swapped: false,
+    pendingVersion: winner ? isPendingVersionStatus(winner.status) : false,
   };
 }
 

apps/webapp/test/realtimeServices.replicaLag.test.ts

@@ -348,7 +348,7 @@ describe("realtime-svc — replica-lag guards", () => {
       });
 
       // 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).