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

.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.

.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<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).