diff --git a/openspec/changes/actor-identity-end-state/tasks.md b/openspec/changes/actor-identity-end-state/tasks.md index 7944117c84..931e7730c0 100644 --- a/openspec/changes/actor-identity-end-state/tasks.md +++ b/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, diff --git a/packages/junior/src/chat/queue/callback.ts b/packages/junior/src/chat/queue/callback.ts index 31bc11e50e..ec79555414 100644 --- a/packages/junior/src/chat/queue/callback.ts +++ b/packages/junior/src/chat/queue/callback.ts @@ -10,6 +10,8 @@ import type { QueueRejectReason, QueueVerifyResult } from "./sign"; export interface QueueCallbackOptions { 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(options: QueueCallbackOptions) { const consumer: QueueConsumer = { consumerGroup: options.consumerGroup, + currentDeployment: options.currentDeployment, handler, retry, topic: options.topic, diff --git a/packages/junior/src/chat/queue/provider.ts b/packages/junior/src/chat/queue/provider.ts index e32ee3ed47..30df2944d5 100644 --- a/packages/junior/src/chat/queue/provider.ts +++ b/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; } diff --git a/packages/junior/src/chat/queue/vercel.ts b/packages/junior/src/chat/queue/vercel.ts index b1a710c78f..7ff6f0f002 100644 --- a/packages/junior/src/chat/queue/vercel.ts +++ b/packages/junior/src/chat/queue/vercel.ts @@ -15,6 +15,7 @@ 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 { @@ -22,6 +23,13 @@ function getVercelQueueClient(): QueueClient { 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(), diff --git a/packages/junior/src/chat/task-execution/vercel-callback.ts b/packages/junior/src/chat/task-execution/vercel-callback.ts index f6a1e77594..9ef1eb9353 100644 --- a/packages/junior/src/chat/task-execution/vercel-callback.ts +++ b/packages/junior/src/chat/task-execution/vercel-callback.ts @@ -181,6 +181,7 @@ function conversationWorkCallback( ) { return queueCallback({ consumerGroup: CONVERSATION_WORK_DEV_CONSUMER_GROUP, + currentDeployment: true, // Conversation work stores and limits failed attempts in durable state. maxDeliveries: null, onRejected: logConversationQueueMessageRejected, diff --git a/packages/junior/src/chat/task-execution/vercel-queue.ts b/packages/junior/src/chat/task-execution/vercel-queue.ts index 2d70dc3b96..ef52972b98 100644 --- a/packages/junior/src/chat/task-execution/vercel-queue.ts +++ b/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 {}; diff --git a/packages/junior/tests/component/vercel-queue-dev.test.ts b/packages/junior/tests/component/vercel-queue-dev.test.ts index 840bdf7dc4..8e68645cff 100644 --- a/packages/junior/tests/component/vercel-queue-dev.test.ts +++ b/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(), })); diff --git a/packages/junior/tests/integration/conversation-queue.test.ts b/packages/junior/tests/integration/conversation-queue.test.ts index 410cde3b36..0d48f2011c 100644 --- a/packages/junior/tests/integration/conversation-queue.test.ts +++ b/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, }, ); });