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

Chat sessions can now be pinned to a deployment, so a conversation keeps
talking to the agent version its release shipped with, across every
turn, idle suspend and recovery. The deployment id is resolved wherever
you start the session, exactly as it is for `trigger()`, so there is no
chat-specific setup — pass `triggerConfig: { externalDeploymentId: null
}` to opt a chat out. A pinned session also follows its pin on its own:
when your app redeploys and re-pins it, the agent hands the conversation
over at the next turn boundary instead of the new pin only applying to
the next run. Set `versionSkew: "hold"` on an agent that should stay
put. If a handover lands on a deployment that is still building, the
chat reports a pending version rather than going quiet, and messages
sent meanwhile are kept. Also fixes `AgentChat` ignoring `maxDuration`,
`region` and `lockToVersion` set on its `triggerConfig`, and a chat
agent answering the same message twice after a conversation moved to a
new run.

Mono-RevId: 1c6e161b3e4f6c3e81a50a7f24d08a43b0ba0057

Author: 0ski

.changeset/chat-agent-deployment-pinning.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Chat sessions can now be pinned to a deployment, so a conversation keeps talking to the agent version its release shipped with, and follows the pin on its own when your app redeploys. Opt out with `triggerConfig: { externalDeploymentId: null }` or `versionSkew: "hold"`. Also fixes `AgentChat` ignoring `maxDuration`, `region` and `lockToVersion`, and a restored `AgentChat` session never picking up a new deployment id.

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
@@ -133,6 +134,7 @@ const { action, loader } = createActionApiRoute(
       const responseBody: EndAndContinueSessionResponseBody = {
         runId: run?.friendlyId ?? result.runId,
         swapped: result.swapped,
+        pendingVersion: result.pendingVersion,
       };
       return json<EndAndContinueSessionResponseBody>(responseBody);
     } catch (error) {

apps/webapp/app/routes/api.v1.sessions.ts

@@ -257,6 +257,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

@@ -131,7 +131,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,
@@ -213,7 +213,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
@@ -215,7 +221,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.
@@ -268,7 +278,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),
+      };
     }
   }
 
@@ -285,6 +299,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
@@ -301,7 +333,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 = {
@@ -315,15 +347,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();
@@ -342,7 +366,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 = {
@@ -372,6 +400,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 = {
@@ -384,6 +414,8 @@ export type SwapSessionRunResult = {
    * next run.
    */
   swapped: boolean;
+  /** See {@link EnsureRunResult.pendingVersion}. */
+  pendingVersion: boolean;
 };
 
 /**
@@ -426,7 +458,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,
@@ -447,6 +487,7 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise<Swap
     data: {
       currentRunId: triggered.id,
       currentRunVersion: { increment: 1 },
+      ...(reason === "upgrade" ? { triggerConfig: config as Prisma.InputJsonValue } : {}),
     },
   });
 
@@ -463,7 +504,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
@@ -504,9 +549,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).