fix(task-execution): Route conversation wakes to the current deployment

openspec/changes/actor-identity-end-state/tasks.md

@@ -89,9 +89,9 @@ Finish #2482.
     - The Inbound message schema is strict. The Slack, web, Agent
       invocation, and plugin dispatch kinds of `input.metadata` are strict
       too. Old code rejects the id on the entry and in its metadata.
-    - The mailbox is in shared Redis. Vercel Queues sends a queue message
-      only to the deployment that sent it, but this does not stop an old
-      worker from reading a new mailbox entry.
+    - The mailbox is in shared Redis. Conversation wakes are not pinned to
+      the deployment that sent them, and an old worker can read a new
+      mailbox entry.
     - During a deploy, an old worker can still run a Turn while new code
       adds a steering message to the same Conversation. After that, the old
       worker cannot read the Conversation execution. Its lease check-in,

packages/junior/src/chat/queue/callback.ts

@@ -10,6 +10,8 @@ import type { QueueRejectReason, QueueVerifyResult } from "./sign";
 
 export interface QueueCallbackOptions<Message> {
   consumerGroup: string;
+  /** Receive messages sent with `currentDeployment`. */
+  currentDeployment?: boolean;
   /** Use `null` when durable state already owns the attempt limit. */
   maxDeliveries: number | null;
   onRejected(
@@ -89,6 +91,7 @@ export function queueCallback<Message>(options: QueueCallbackOptions<Message>) {
 
   const consumer: QueueConsumer = {
     consumerGroup: options.consumerGroup,
+    currentDeployment: options.currentDeployment,
     handler,
     retry,
     topic: options.topic,

packages/junior/src/chat/queue/provider.ts

@@ -31,6 +31,11 @@ export interface QueueSendOptions {
   idempotencyKey?: string;
   /** Drop the message if it is not delivered within this many seconds. */
   retentionSeconds?: number;
+  /**
+   * Do not pin the message to the deployment that sent it, so a newer
+   * deployment can receive it. Its consumer must set the same flag.
+   */
+  currentDeployment?: boolean;
 }
 
 /** Result of one queue send. */
@@ -48,6 +53,8 @@ export interface QueueConsumer {
     metadata: QueueMessageMetadata,
   ): QueueRetryDirective | undefined;
   topic: string;
+  /** Receive messages sent with `currentDeployment`. */
+  currentDeployment?: boolean;
   /** Wait this many seconds before a message that is in flight is sent again. */
   visibilityTimeoutSeconds?: number;
 }

packages/junior/src/chat/queue/vercel.ts

@@ -15,13 +15,21 @@ function defaultQueueClientOptions(): QueueClientOptions {
 }
 
 let client: QueueClient | undefined;
+let currentDeploymentClient: QueueClient | undefined;
 
 // Build on first use, not at import, so env is read when the app runs.
 function getVercelQueueClient(): QueueClient {
   client ??= new QueueClient(defaultQueueClientOptions());
   return client;
 }
 
+// A default client pins each message to the deployment that sent it. This
+// client is unpinned, so its messages are not tied to an old deployment.
+function getCurrentDeploymentQueueClient(): QueueClient {
+  currentDeploymentClient ??= new QueueClient({ deploymentId: null });
+  return currentDeploymentClient;
+}
+
 function visibility(consumer: QueueConsumer) {
   return consumer.visibilityTimeoutSeconds === undefined
     ? {}
@@ -33,18 +41,22 @@ export const vercelQueueProvider: QueueProvider = {
   // OpenTelemetry has no well-known `messaging.system` value for Vercel Queues.
   messagingSystem: "vercel_queue",
   send: async (topic, payload, options) => {
-    const { messageId } = await getVercelQueueClient().send(
-      topic,
-      payload,
-      options,
-    );
+    const { currentDeployment, ...sendOptions } = options ?? {};
+    const queueClient = currentDeployment
+      ? getCurrentDeploymentQueueClient()
+      : getVercelQueueClient();
+    const { messageId } = await queueClient.send(topic, payload, sendOptions);
     return messageId ? { messageId } : {};
   },
-  createCallback: (consumer) =>
-    handleCallback(consumer.handler, {
-      retry: consumer.retry,
-      ...visibility(consumer),
-    }),
+  createCallback: (consumer) => {
+    const options = { retry: consumer.retry, ...visibility(consumer) };
+    return consumer.currentDeployment
+      ? getCurrentDeploymentQueueClient().handleCallback(
+          consumer.handler,
+          options,
+        )
+      : handleCallback(consumer.handler, options);
+  },
   registerDevConsumer: (consumer) =>
     registerDevConsumer({
       client: getVercelQueueClient(),

packages/junior/src/chat/task-execution/vercel-callback.ts

@@ -181,6 +181,7 @@ function conversationWorkCallback(
 ) {
   return queueCallback<ConversationQueueMessage>({
     consumerGroup: CONVERSATION_WORK_DEV_CONSUMER_GROUP,
+    currentDeployment: true,
     // Conversation work stores and limits failed attempts in durable state.
     maxDeliveries: null,
     onRejected: logConversationQueueMessageRejected,

packages/junior/src/chat/task-execution/vercel-queue.ts

@@ -67,6 +67,9 @@ export function createVercelConversationWorkQueue(
             retentionSeconds:
               options.retentionSeconds ??
               CONVERSATION_WORK_QUEUE_RETENTION_SECONDS,
+            // Work continues on the new version after a deploy. The payload
+            // holds only ids, and durable state already spans versions.
+            currentDeployment: true,
           });
           if (!result.messageId) {
             return {};

packages/junior/tests/component/vercel-queue-dev.test.ts

@@ -88,6 +88,7 @@ const originalQueueTopic = process.env.JUNIOR_CONVERSATION_WORK_QUEUE_TOPIC;
 const originalConversationWorkEnabled =
   process.env.JUNIOR_CONVERSATION_WORK_ENABLED;
 const originalJuniorSecret = process.env.JUNIOR_SECRET;
+const originalDeploymentId = process.env.VERCEL_DEPLOYMENT_ID;
 
 afterEach(async () => {
   const { disconnectStateAdapter } = await import("@/chat/state/adapter");
@@ -109,6 +110,11 @@ afterEach(async () => {
   } else {
     process.env.JUNIOR_SECRET = originalJuniorSecret;
   }
+  if (originalDeploymentId === undefined) {
+    delete process.env.VERCEL_DEPLOYMENT_ID;
+  } else {
+    process.env.VERCEL_DEPLOYMENT_ID = originalDeploymentId;
+  }
   vi.doUnmock("@vercel/queue");
   vi.doUnmock("@/chat/plugins/task-runner");
   vi.doUnmock("@/chat/sandbox/snapshot/job-runner");
@@ -312,6 +318,7 @@ describe("plugin task Vercel queue integration", () => {
     }));
 
     process.env.JUNIOR_SECRET = "plugin-task-secret";
+    process.env.VERCEL_DEPLOYMENT_ID = "dpl_sender";
 
     const { PLUGIN_TASK_QUEUE_TOPIC, pluginTaskId, sendVercelPluginTask } =
       await import("@/chat/plugins/task-queue");
@@ -339,6 +346,41 @@ describe("plugin task Vercel queue integration", () => {
         retentionSeconds: 3600,
       },
     );
+    // Plugin tasks stay on the deployment that sent them.
+    expect(QueueClient).toHaveBeenCalledExactlyOnceWith({});
+  });
+
+  it("sends conversation wakes unpinned from the sending deployment", async () => {
+    const send = vi.fn(async () => ({ messageId: "msg_wake" }));
+    const QueueClient = vi.fn(function QueueClientMock() {
+      return { send };
+    });
+
+    vi.doMock("@vercel/queue", () => ({
+      QueueClient,
+      handleCallback: vi.fn(),
+      registerDevConsumer: vi.fn(),
+    }));
+
+    process.env.JUNIOR_SECRET = "conversation-work-secret";
+    process.env.VERCEL_DEPLOYMENT_ID = "dpl_sender";
+
+    const { createVercelConversationWorkQueue } =
+      await import("@/chat/task-execution/vercel-queue");
+
+    await expect(
+      createVercelConversationWorkQueue({ topic: "work" }).send({
+        schemaVersion: 2,
+        conversationId: "slack:C123:1712345.0001",
+      }),
+    ).resolves.toEqual({ messageId: "msg_wake" });
+
+    expect(QueueClient).toHaveBeenCalledExactlyOnceWith({ deploymentId: null });
+    expect(send).toHaveBeenCalledWith("work", expect.any(Object), {
+      delaySeconds: undefined,
+      idempotencyKey: undefined,
+      retentionSeconds: 3600,
+    });
   });
 });
 
@@ -386,9 +428,12 @@ describe("registerVercelConversationWorkDevConsumer", () => {
     const routeHandler = vi.fn();
     const handleCallback = vi.fn(() => routeHandler);
 
+    const QueueClient = vi.fn(function QueueClientMock() {
+      return { handleCallback };
+    });
     vi.doMock("@vercel/queue", () => ({
-      QueueClient: vi.fn(),
-      handleCallback,
+      QueueClient,
+      handleCallback: vi.fn(),
       registerDevConsumer: vi.fn(),
     }));
 
@@ -405,6 +450,8 @@ describe("registerVercelConversationWorkDevConsumer", () => {
         visibilityTimeoutSeconds: 45,
       }),
     ).toBe(routeHandler);
+    // The consumer must ack unpinned messages with an unpinned client.
+    expect(QueueClient).toHaveBeenCalledExactlyOnceWith({ deploymentId: null });
 
     type TestQueueMetadata = {
       consumerGroup: string;
@@ -514,9 +561,12 @@ describe("registerVercelConversationWorkDevConsumer", () => {
     const routeHandler = vi.fn();
     const handleCallback = vi.fn(() => routeHandler);
 
+    const QueueClient = vi.fn(function QueueClientMock() {
+      return { handleCallback };
+    });
     vi.doMock("@vercel/queue", () => ({
-      QueueClient: vi.fn(),
-      handleCallback,
+      QueueClient,
+      handleCallback: vi.fn(),
       registerDevConsumer: vi.fn(),
     }));
 

packages/junior/tests/integration/conversation-queue.test.ts

@@ -35,6 +35,7 @@ describe("conversation queue transport", () => {
         delaySeconds: 16,
         idempotencyKey: "idem-1",
         retentionSeconds: 3_600,
+        currentDeployment: true,
       },
     );
   });