From c2b8ccef9aaaada511fee23a60ab9132d4dd780d Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 07:29:00 +0000 Subject: [PATCH 01/10] fix(runtime): Keep resource events out of steering Co-Authored-By: David Cramer --- .../src/chat/task-execution/slack-work.ts | 24 ++++--- .../slack-conversation-work.test.ts | 72 +++++++++++++++++++ 2 files changed, 88 insertions(+), 8 deletions(-) diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index 186617f44..12a98374d 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -845,21 +845,29 @@ export function createSlackConversationWorker( ) => Promise, ): Promise => { await context.attempt.drain(async (pendingRecords) => { - const messages = pendingRecords.map((record) => { + const messages = pendingRecords.flatMap((record) => { const metadata = parseSlackMetadata(record.input.metadata); if (!metadata) { throw new Error( "Conversation mailbox record is not Slack metadata", ); } + // Provider notifications are follow-up work, not user steering. + // Leave them pending so the next worker slice handles them as + // their own subscribed turn after the active answer completes. + if (record.source === "resource_event") { + return []; + } const message = restoreMessage({ adapter, record }); - return { - activeRequest: - metadata.route === "mention" || - isSlackAssistantThreadUserMessage(message), - inboundMessageId: record.inboundMessageId, - message, - }; + return [ + { + activeRequest: + metadata.route === "mention" || + isSlackAssistantThreadUserMessage(message), + inboundMessageId: record.inboundMessageId, + message, + }, + ]; }); return await accept(messages); }); diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index 389b2e09e..63255ea63 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -939,6 +939,78 @@ describe("Slack conversation work execution", () => { }); }); + it("leaves resource events queued as follow-up work during an active turn", async () => { + const queue = createConversationWorkQueueTestAdapter(); + const state = getStateAdapter(); + await state.connect(); + const slackAdapter = createSlackAdapterFixture(); + const ingressServices = { + getSlackAdapter: () => slackAdapter, + queue, + runtime: createNoopSlackWebhookRuntime(), + state, + }; + await handleSlackWebhookAndFlush({ + request: slackWebhookRequest( + slackEnvelope({ + text: `<@${SLACK_BOT_USER_ID}> first`, + ts: "1712345.0001", + }), + ), + services: ingressServices, + }); + + const observed: string[][] = []; + const runtime: SlackWorkerOptions["runtime"] = { + handleNewMention: async (_thread, _message, hooks) => { + await hooks.ack?.(); + await appendInboundMessage({ + message: createSlackResourceEventInboundMessage({ + event: { + eventKey: "check-suite-1", + eventType: "check_suite.completed", + occurredAtMs: 2_000, + provider: "github", + resourceRef: "github:pull_request:getsentry/junior#1010", + }, + subscription: { + conversationId: CONVERSATION_ID, + destination: SLACK_DESTINATION, + id: "sub-1", + }, + text: "CI failed.", + }), + state, + }); + await hooks.drainSteeringMessages?.(async (steering) => { + observed.push(steering.map((candidate) => candidate.message.id)); + return steering.map((candidate) => candidate.inboundMessageId); + }); + }, + handleSubscribedMessage: async () => { + throw new Error("resource event should remain queued for follow-up"); + }, + }; + + await expect( + processNextQueuedSlackWork({ + getSlackAdapter: () => slackAdapter, + queue, + runtime, + state, + }), + ).resolves.toEqual({ status: "completed" }); + + expect(observed).toEqual([[]]); + const work = await getConversationWorkState({ + conversationId: CONVERSATION_ID, + state, + }); + expect(work?.execution.pendingMessages).toEqual([ + expect.objectContaining({ source: "resource_event" }), + ]); + }); + it("treats Slack assistant-thread user messages as active steering", async () => { const queue = createConversationWorkQueueTestAdapter(); const state = getStateAdapter(); From b3aa45577bb9109e4ad85cf14617a5f034f498aa Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 07:37:06 +0000 Subject: [PATCH 02/10] test(runtime): Expect resource event requeue --- .../component/task-execution/slack-conversation-work.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index 63255ea63..faddd92ba 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -999,7 +999,7 @@ describe("Slack conversation work execution", () => { runtime, state, }), - ).resolves.toEqual({ status: "completed" }); + ).resolves.toEqual({ status: "pending_requeued" }); expect(observed).toEqual([[]]); const work = await getConversationWorkState({ From f024fc2d5d17f371c5b79ac25e60c322f035b53b Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 15:42:28 +0000 Subject: [PATCH 03/10] refactor(runtime): Model mailbox routing explicitly Co-Authored-By: David Cramer --- .../junior/src/chat/task-execution/README.md | 8 +- .../src/chat/task-execution/slack-work.ts | 42 ++++------- .../junior/src/chat/task-execution/state.ts | 59 +++++++++++++++ .../junior/src/chat/task-execution/store.ts | 13 ++++ .../junior/src/chat/task-execution/worker.ts | 36 +++++++-- .../component/conversation-sql-store.test.ts | 2 +- .../task-execution/conversation-work.test.ts | 75 ++++++++++++++----- .../slack-conversation-work.test.ts | 8 +- 8 files changed, 185 insertions(+), 58 deletions(-) diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index c3fe75999..072b78789 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -6,7 +6,8 @@ source of truth. ## State Model -- A conversation mailbox contains normalized pending user work. +- A conversation mailbox contains normalized pending work with a durable routing + mode: `steer`, `follow_up`, or `implicit` when runtime policy must decide. - A queue payload identifies the conversation to wake; persisted conversation work owns destination and routing. - A lease grants one worker temporary execution ownership. @@ -30,8 +31,9 @@ source of truth. 6. Terminal delivery or intentional no-reply completion records the delivered turn before acknowledging work. -New messages that arrive during a run remain durable and are drained at the -next safe boundary or subsequent wake-up. +New messages that arrive during a run remain durable. Explicit `steer` work is +eligible at the next safe boundary, explicit `follow_up` work waits for the next +turn, and `implicit` work is classified by runtime policy. ## Queue And Lease Rules diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index 12a98374d..a8a6ed773 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -566,6 +566,7 @@ export function createSlackResourceEventInboundMessage( createdAtMs: input.event.occurredAtMs, destination, inboundMessageId: `resource-event:${input.subscription.id}:${input.event.eventKey}`, + routing: "follow_up", source: "resource_event", receivedAtMs: Date.now(), input: { @@ -837,38 +838,20 @@ export function createSlackConversationWorker( ); } }; - // Restore stored mailbox entries as Slack steering candidates; the - // runtime returns only the inbound ids it handled durably. + // Explicit follow-ups never enter steering. Implicit records are + // offered to runtime policy, while explicit steering bypasses that + // ambiguity through activeRequest. const drainSteeringMessages = async ( accept: ( messages: SteeringCandidateMessage[], ) => Promise, ): Promise => { - await context.attempt.drain(async (pendingRecords) => { - const messages = pendingRecords.flatMap((record) => { - const metadata = parseSlackMetadata(record.input.metadata); - if (!metadata) { - throw new Error( - "Conversation mailbox record is not Slack metadata", - ); - } - // Provider notifications are follow-up work, not user steering. - // Leave them pending so the next worker slice handles them as - // their own subscribed turn after the active answer completes. - if (record.source === "resource_event") { - return []; - } - const message = restoreMessage({ adapter, record }); - return [ - { - activeRequest: - metadata.route === "mention" || - isSlackAssistantThreadUserMessage(message), - inboundMessageId: record.inboundMessageId, - message, - }, - ]; - }); + await context.attempt.steer(async (pendingRecords) => { + const messages = pendingRecords.map((record) => ({ + activeRequest: record.routing === "steer", + inboundMessageId: record.inboundMessageId, + message: restoreMessage({ adapter, record }), + })); return await accept(messages); }); }; @@ -946,6 +929,11 @@ export function buildSlackInboundMessage(args: { args.conversationId, args.message.id, ].join(":"), + routing: + args.route === "mention" || + isSlackAssistantThreadUserMessage(args.message) + ? "steer" + : "implicit", source: "slack", createdAtMs: args.message.metadata.dateSent.getTime(), receivedAtMs: args.receivedAtMs, diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index ec0e1f607..b44078ec2 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -83,6 +83,8 @@ export interface AgentInput { text: string; } +export type InboundMessageRouting = "follow_up" | "implicit" | "steer"; + export interface InboundMessage { attemptCount?: number; conversationId: string; @@ -92,6 +94,7 @@ export interface InboundMessage { injectedAtMs?: number; input: AgentInput; receivedAtMs: number; + routing?: InboundMessageRouting; source: Source; } @@ -325,6 +328,10 @@ function normalizeInput(value: unknown): AgentInput | undefined { }; } +function normalizeInboundMessageRouting(value: unknown): InboundMessageRouting { + return value === "follow_up" || value === "steer" ? value : "implicit"; +} + function normalizeMessage(value: unknown): InboundMessage | undefined { if (!isRecord(value)) { return undefined; @@ -355,6 +362,7 @@ function normalizeMessage(value: unknown): InboundMessage | undefined { createdAtMs, receivedAtMs, input, + routing: normalizeInboundMessageRouting(value.routing), injectedAtMs: toOptionalNumber(value.injectedAtMs), attemptCount: toOptionalNumber(value.attemptCount), }; @@ -1483,6 +1491,57 @@ export async function checkInConversationWork(args: { }); } +/** Persist an explicit route for pending mailbox entries under the active lease. */ +export async function routeConversationMailboxMessages(args: { + conversationId: string; + inboundMessageIds: readonly string[]; + leaseToken: string; + nowMs?: number; + routing: Exclude; + state?: StateAdapter; +}): Promise { + if (args.inboundMessageIds.length === 0) { + return; + } + const routedIds = new Set(args.inboundMessageIds); + await withConversationMutation(args, async (state, lock) => { + const current = await readConversation(state, args.conversationId); + if (!current || current.execution.lease?.token !== args.leaseToken) { + throw new Error( + `Conversation lease is not held for ${args.conversationId}`, + ); + } + const pendingIds = new Set( + current.execution.pendingMessages.map( + (message) => message.inboundMessageId, + ), + ); + for (const inboundMessageId of routedIds) { + if (!pendingIds.has(inboundMessageId)) { + throw new Error( + `Conversation mailbox route is not pending for ${args.conversationId}`, + ); + } + } + await writeConversation( + state, + lock, + withExecutionUpdate( + current, + { + ...current.execution, + pendingMessages: current.execution.pendingMessages.map((message) => + routedIds.has(message.inboundMessageId) + ? { ...message, routing: args.routing } + : message, + ), + }, + args.nowMs ?? now(), + ), + ); + }); +} + /** * Drain pending mailbox entries after the caller acknowledges durable handling. * diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 28e8ee197..07e8f1a01 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -22,6 +22,7 @@ export { type ConversationWorkState, type ExecutionStatus, type InboundMessage, + type InboundMessageRouting, type Lease, type RequestConversationWorkResult, type Source, @@ -362,6 +363,18 @@ export async function checkInConversationWork(args: { return result; } +/** Persist an explicit route for pending mailbox entries under the active lease. */ +export async function routeConversationMailboxMessages( + args: Parameters[0] & { + conversationStore?: ConversationStore; + state?: StateAdapter; + }, +) { + const result = await workState.routeConversationMailboxMessages(args); + await recordExecutionMetadata(args); + return result; +} + /** Drain pending mailbox entries after the caller accepts responsibility. */ export async function drainConversationMailbox( args: Parameters[0] & { diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 526fbcd95..042573e60 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -24,6 +24,7 @@ import { recordAttemptFailure, releaseConversationWork, requestConversationContinuation, + routeConversationMailboxMessages, startConversationWork, type AttemptFailure, type ConversationWorkState, @@ -45,11 +46,12 @@ export interface InboxAttempt { ack(): Promise; conversationId: string; destination: Destination; - drain( - handle: (messages: InboundMessage[]) => Promise, - ): Promise; + followUp(inboundMessageIds: readonly string[]): Promise; isFinalAttempt: boolean; messages: InboundMessage[]; + steer( + handle: (messages: InboundMessage[]) => Promise, + ): Promise; } export interface ConversationWorkerResult { @@ -338,18 +340,39 @@ export async function processConversationWork( "Conversation work lease acquired", ); - const drainInbox = ( + const steer = ( handle: (messages: InboundMessage[]) => Promise, ) => drainConversationMailbox({ conversationId, leaseToken: lease.leaseToken, conversationStore: options.conversationStore, - handle, + handle: async (messages) => { + const candidates = messages.filter( + (message) => message.routing !== "follow_up", + ); + return ( + (await handle(candidates)) ?? + candidates.map((message) => message.inboundMessageId) + ); + }, nowMs: now(options), state: options.state, }); + const followUp = async ( + inboundMessageIds: readonly string[], + ): Promise => { + await routeConversationMailboxMessages({ + conversationId, + inboundMessageIds, + leaseToken: lease.leaseToken, + routing: "follow_up", + nowMs: now(options), + state: options.state, + }); + }; + const ack = async (): Promise => { const acknowledged = await ackMessages({ conversationId, @@ -372,11 +395,12 @@ export async function processConversationWork( ack, conversationId, destination, - drain: drainInbox, + followUp, isFinalAttempt: attemptMessages.some((message) => isFinalAttempt(message), ), messages: attemptMessages, + steer, }, conversationId, destination, diff --git a/packages/junior/tests/component/conversation-sql-store.test.ts b/packages/junior/tests/component/conversation-sql-store.test.ts index f1caaa4b2..9df7f2c72 100644 --- a/packages/junior/tests/component/conversation-sql-store.test.ts +++ b/packages/junior/tests/component/conversation-sql-store.test.ts @@ -1025,7 +1025,7 @@ WHERE conversation_id = $1 conversationStore: store, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 1b943a609..86a90c07c 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -573,7 +573,7 @@ describe("conversation work execution", () => { queue, run: async (context) => { runs += 1; - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; @@ -613,7 +613,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; @@ -725,7 +725,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); currentNowMs = 2_000; await requestConversationWork({ conversationId: context.conversationId, @@ -761,7 +761,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); currentNowMs = 2_000; await requestConversationWork({ conversationId: context.conversationId, @@ -806,7 +806,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); currentNowMs = 2_000; await requestConversationWork({ conversationId: context.conversationId, @@ -930,7 +930,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); currentNowMs = 2_000; await scheduleAgentContinue( { @@ -1090,7 +1090,7 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - injected.push(await context.attempt.drain(async () => {})); + injected.push(await context.attempt.steer(async () => {})); return { status: "completed" }; }, }), @@ -1123,7 +1123,7 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - await context.attempt.drain(async (messages) => { + await context.attempt.steer(async (messages) => { injected.push(messages.map((message) => message.inboundMessageId)); return messages .filter((message) => message.inboundMessageId === "m1") @@ -1144,6 +1144,43 @@ describe("conversation work execution", () => { ]); }); + it("routes explicit follow-ups separately from steering", async () => { + const queue = createConversationWorkQueueTestAdapter(); + await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); + await appendInboundMessage({ + message: inboundMessage("m2", { + createdAtMs: 2_000, + receivedAtMs: 2_100, + }), + nowMs: 2_100, + }); + const steered: string[][] = []; + + await expect( + processConversationWork(conversationQueueMessage(), { + queue, + run: async (context) => { + await context.attempt.followUp(["m2"]); + await context.attempt.steer(async (messages) => { + steered.push(messages.map((message) => message.inboundMessageId)); + }); + return { status: "completed" }; + }, + }), + ).resolves.toEqual({ status: "pending_requeued" }); + + expect(steered).toEqual([["m1"]]); + const state = await getConversationWorkState({ + conversationId: CONVERSATION_ID, + }); + expect(state?.messages).toEqual([ + expect.objectContaining({ + inboundMessageId: "m2", + routing: "follow_up", + }), + ]); + }); + it("rejects mailbox acknowledgements outside the pending set", async () => { const queue = createConversationWorkQueueTestAdapter(); await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); @@ -1154,7 +1191,7 @@ describe("conversation work execution", () => { queue, run: async (context) => { try { - await context.attempt.drain(async () => ["different-message"]); + await context.attempt.steer(async () => ["different-message"]); } catch (error) { drainError = error; throw error; @@ -1195,7 +1232,7 @@ describe("conversation work execution", () => { queue, state: observed.state, run: async (context) => { - const drain = context.attempt.drain(async () => { + const drain = context.attempt.steer(async () => { expect(observed.isHeld()).toBe(false); injectionStarted.resolve(); await finishInjection.promise; @@ -1244,7 +1281,7 @@ describe("conversation work execution", () => { checkInIntervalMs: 15_000, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; @@ -1284,7 +1321,7 @@ describe("conversation work execution", () => { checkInIntervalMs: 15_000, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); entered.resolve({ shouldYield: context.shouldYield, }); @@ -1520,7 +1557,7 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - const first = await context.attempt.drain(async () => {}); + const first = await context.attempt.steer(async () => {}); injected.push(first.map((message) => message.inboundMessageId)); await appendInboundMessage({ message: inboundMessage("m2", { @@ -1529,7 +1566,7 @@ describe("conversation work execution", () => { }), nowMs: 2_100, }); - const second = await context.attempt.drain(async () => {}); + const second = await context.attempt.steer(async () => {}); injected.push(second.map((message) => message.inboundMessageId)); return { status: "completed" }; }, @@ -1547,7 +1584,7 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); await appendInboundMessage({ message: inboundMessage("m2", { createdAtMs: 2_000, @@ -1555,7 +1592,7 @@ describe("conversation work execution", () => { }), nowMs: 2_100, }); - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); return { status: "completed" }; }, }), @@ -1578,7 +1615,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); currentNowMs = 2_100; await appendInboundMessage({ message: inboundMessage("m2", { @@ -1609,7 +1646,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.drain(async () => {}); + await context.attempt.steer(async () => {}); currentNowMs = 242_000; expect(context.shouldYield()).toBe(true); return { status: "yielded" }; @@ -1806,7 +1843,7 @@ describe("conversation work execution", () => { processConversationQueueMessage(conversationQueueMessage(), { queue, run: async (context) => { - const messages = await context.attempt.drain(async () => {}); + const messages = await context.attempt.steer(async () => {}); injected.push(...messages.map((message) => message.inboundMessageId)); return { status: "completed" }; }, diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index faddd92ba..55ed182f2 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -343,9 +343,10 @@ describe("Slack conversation work execution", () => { ack: async () => {}, conversationId: CONVERSATION_ID, destination: SLACK_DESTINATION, - drain: async () => [], + followUp: async () => {}, isFinalAttempt: false, messages: [malformed], + steer: async () => [], }, checkIn: async () => true, conversationId: CONVERSATION_ID, @@ -1007,7 +1008,10 @@ describe("Slack conversation work execution", () => { state, }); expect(work?.execution.pendingMessages).toEqual([ - expect.objectContaining({ source: "resource_event" }), + expect.objectContaining({ + routing: "follow_up", + source: "resource_event", + }), ]); }); From 127ec3aeeab315fdc78c2d16292bfe1e54fb6d9f Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 16:11:54 +0000 Subject: [PATCH 04/10] refactor(runtime): Clarify mailbox routing contract Co-Authored-By: David Cramer --- TERMINOLOGY.md | 3 ++ .../junior/src/chat/runtime/slack-runtime.ts | 32 ++++++++++++------- .../junior/src/chat/task-execution/README.md | 4 +-- .../src/chat/task-execution/slack-work.ts | 24 ++++++++------ .../junior/src/chat/task-execution/state.ts | 6 ++-- .../slack-conversation-work.test.ts | 15 +++++++-- 6 files changed, 55 insertions(+), 29 deletions(-) diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index bfce487f6..cc7393736 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -17,6 +17,9 @@ Canonical words used across Junior's code and documentation. batched into that turn. - **Follow-up message**: a user message that waits for the active turn to finish before starting the next turn. +- **Message routing**: the durable mailbox instruction for an inbound message: + `steer` interrupts at the next safe boundary, `follow_up` waits for the active + turn to finish, and `auto` asks runtime policy to choose between them. - **Turn**: one request-to-final-response cycle. It may span multiple runs and execution slices; one model invocation is not a turn. - **Run**: one bounded attempt to advance a turn. A later run may resume the diff --git a/packages/junior/src/chat/runtime/slack-runtime.ts b/packages/junior/src/chat/runtime/slack-runtime.ts index 91d6dec29..8b15cc57c 100644 --- a/packages/junior/src/chat/runtime/slack-runtime.ts +++ b/packages/junior/src/chat/runtime/slack-runtime.ts @@ -56,7 +56,7 @@ export interface AssistantLifecycleEvent { userId?: string; } -type SteeringMode = "defer" | "interrupt"; +type SteeringMode = "follow_up" | "steer"; export interface SteeringCandidateMessage { activeRequest: boolean; @@ -64,12 +64,17 @@ export interface SteeringCandidateMessage { message: Message; } +export interface SteeringDisposition { + followUp: readonly string[]; + handled: readonly string[]; +} + export interface ReplyHooks { beforeFirstResponsePost?: () => Promise; drainSteeringMessages?: ( accept: ( messages: SteeringCandidateMessage[], - ) => Promise, + ) => Promise, ) => Promise; messageContext?: MessageContext; ack?: () => Promise; @@ -295,20 +300,25 @@ function createAcceptedSteeringDrain( await hooks.drainSteeringMessages!(async (messages) => { const selection = await options.selectMessages(messages, context); await options.onSkipped?.(selection.skipped); - // Deferred accepted messages stay pending so a later worker slice handles - // them after the active answer is delivered. + // Follow-ups stay pending so a later worker slice handles them after the + // active answer is delivered. const interrupted = selection.accepted - .filter((accepted) => accepted.mode === "interrupt") + .filter((accepted) => accepted.mode === "steer") .map((accepted) => accepted.message); await accept(getQueuedMessagesFromSlackMessages(interrupted, options)); interruptedMessages = interrupted; await options.onAcceptedForProcessing?.(interrupted); - return [ - ...selection.accepted - .filter((accepted) => accepted.mode === "interrupt") + return { + followUp: selection.accepted + .filter((accepted) => accepted.mode === "follow_up") .map((accepted) => accepted.inboundMessageId), - ...selection.skipped.map((skipped) => skipped.inboundMessageId), - ]; + handled: [ + ...selection.accepted + .filter((accepted) => accepted.mode === "steer") + .map((accepted) => accepted.inboundMessageId), + ...selection.skipped.map((skipped) => skipped.inboundMessageId), + ], + }; }); return getQueuedMessagesFromSlackMessages( interruptedMessages ?? [], @@ -556,7 +566,7 @@ export function createSlackTurnRuntime< return { context, decision, - mode: isActiveRequest ? "interrupt" : "defer", + mode: isActiveRequest ? "steer" : "follow_up", text, }; }; diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index 072b78789..9745b803b 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -7,7 +7,7 @@ source of truth. ## State Model - A conversation mailbox contains normalized pending work with a durable routing - mode: `steer`, `follow_up`, or `implicit` when runtime policy must decide. + mode: `steer`, `follow_up`, or `auto` when runtime policy must decide. - A queue payload identifies the conversation to wake; persisted conversation work owns destination and routing. - A lease grants one worker temporary execution ownership. @@ -33,7 +33,7 @@ source of truth. New messages that arrive during a run remain durable. Explicit `steer` work is eligible at the next safe boundary, explicit `follow_up` work waits for the next -turn, and `implicit` work is classified by runtime policy. +turn, and `auto` work is classified by runtime policy. ## Queue And Lease Rules diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index a8a6ed773..cbed96c0d 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -11,6 +11,7 @@ import { z } from "zod"; import type { SlackTurnOptions, SteeringCandidateMessage, + SteeringDisposition, } from "@/chat/runtime/slack-runtime"; import { isCooperativeTurnYieldError, @@ -838,21 +839,24 @@ export function createSlackConversationWorker( ); } }; - // Explicit follow-ups never enter steering. Implicit records are + // Explicit follow-ups never enter steering. Auto-routed records are // offered to runtime policy, while explicit steering bypasses that - // ambiguity through activeRequest. + // decision through activeRequest. const drainSteeringMessages = async ( accept: ( messages: SteeringCandidateMessage[], - ) => Promise, + ) => Promise, ): Promise => { await context.attempt.steer(async (pendingRecords) => { - const messages = pendingRecords.map((record) => ({ - activeRequest: record.routing === "steer", - inboundMessageId: record.inboundMessageId, - message: restoreMessage({ adapter, record }), - })); - return await accept(messages); + const disposition = await accept( + pendingRecords.map((record) => ({ + activeRequest: record.routing === "steer", + inboundMessageId: record.inboundMessageId, + message: restoreMessage({ adapter, record }), + })), + ); + await context.attempt.followUp(disposition.followUp); + return disposition.handled; }); }; @@ -933,7 +937,7 @@ export function buildSlackInboundMessage(args: { args.route === "mention" || isSlackAssistantThreadUserMessage(args.message) ? "steer" - : "implicit", + : "auto", source: "slack", createdAtMs: args.message.metadata.dateSent.getTime(), receivedAtMs: args.receivedAtMs, diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index b44078ec2..5b4c9b248 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -83,7 +83,7 @@ export interface AgentInput { text: string; } -export type InboundMessageRouting = "follow_up" | "implicit" | "steer"; +export type InboundMessageRouting = "follow_up" | "auto" | "steer"; export interface InboundMessage { attemptCount?: number; @@ -329,7 +329,7 @@ function normalizeInput(value: unknown): AgentInput | undefined { } function normalizeInboundMessageRouting(value: unknown): InboundMessageRouting { - return value === "follow_up" || value === "steer" ? value : "implicit"; + return value === "follow_up" || value === "steer" ? value : "auto"; } function normalizeMessage(value: unknown): InboundMessage | undefined { @@ -1497,7 +1497,7 @@ export async function routeConversationMailboxMessages(args: { inboundMessageIds: readonly string[]; leaseToken: string; nowMs?: number; - routing: Exclude; + routing: Exclude; state?: StateAdapter; }): Promise { if (args.inboundMessageIds.length === 0) { diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index 55ed182f2..e4d4bcb4d 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -905,7 +905,10 @@ describe("Slack conversation work execution", () => { }); await hooks.drainSteeringMessages?.(async (steering) => { injected.push(steering.map((candidate) => candidate.message.id)); - return steering.map((candidate) => candidate.inboundMessageId); + return { + followUp: [], + handled: steering.map((candidate) => candidate.inboundMessageId), + }; }); }, handleSubscribedMessage: async () => { @@ -985,7 +988,10 @@ describe("Slack conversation work execution", () => { }); await hooks.drainSteeringMessages?.(async (steering) => { observed.push(steering.map((candidate) => candidate.message.id)); - return steering.map((candidate) => candidate.inboundMessageId); + return { + followUp: [], + handled: steering.map((candidate) => candidate.inboundMessageId), + }; }); }, handleSubscribedMessage: async () => { @@ -1096,7 +1102,10 @@ describe("Slack conversation work execution", () => { id: candidate.message.id, })), ); - return steering.map((candidate) => candidate.inboundMessageId); + return { + followUp: [], + handled: steering.map((candidate) => candidate.inboundMessageId), + }; }); }, handleSubscribedMessage: async () => { From 7597948f5f5cb15eae8d51b1b8f860ee4b871ec8 Mon Sep 17 00:00:00 2001 From: David Cramer Date: Thu, 23 Jul 2026 09:47:05 -0700 Subject: [PATCH 05/10] ref(runtime): Simplify durable mailbox routing Resolve mailbox acknowledgements and deferrals through one atomic drain operation while keeping agent steering terminology at the runtime boundary. --- TERMINOLOGY.md | 4 +- .../junior/src/chat/runtime/slack-runtime.ts | 26 ++-- .../junior/src/chat/task-execution/README.md | 6 +- .../src/chat/task-execution/slack-work.ts | 22 ++-- .../junior/src/chat/task-execution/state.ts | 119 +++++++----------- .../junior/src/chat/task-execution/store.ts | 27 ++-- .../junior/src/chat/task-execution/worker.ts | 40 +++--- .../component/conversation-sql-store.test.ts | 2 +- .../services/turn-session-record.test.ts | 1 + .../task-execution/conversation-work.test.ts | 67 +++++----- .../slack-conversation-work.test.ts | 22 ++-- .../tests/component/vercel-queue-dev.test.ts | 1 + .../tests/fixtures/conversation-work.ts | 1 + 13 files changed, 156 insertions(+), 182 deletions(-) diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index cc7393736..8c5185963 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -18,8 +18,8 @@ Canonical words used across Junior's code and documentation. - **Follow-up message**: a user message that waits for the active turn to finish before starting the next turn. - **Message routing**: the durable mailbox instruction for an inbound message: - `steer` interrupts at the next safe boundary, `follow_up` waits for the active - turn to finish, and `auto` asks runtime policy to choose between them. + `interrupt` is eligible at the next safe boundary, `defer` waits for the + active turn to finish, and `auto` asks runtime policy to choose between them. - **Turn**: one request-to-final-response cycle. It may span multiple runs and execution slices; one model invocation is not a turn. - **Run**: one bounded attempt to advance a turn. A later run may resume the diff --git a/packages/junior/src/chat/runtime/slack-runtime.ts b/packages/junior/src/chat/runtime/slack-runtime.ts index 8b15cc57c..9484456d6 100644 --- a/packages/junior/src/chat/runtime/slack-runtime.ts +++ b/packages/junior/src/chat/runtime/slack-runtime.ts @@ -56,7 +56,7 @@ export interface AssistantLifecycleEvent { userId?: string; } -type SteeringMode = "follow_up" | "steer"; +type SteeringMode = "defer" | "interrupt"; export interface SteeringCandidateMessage { activeRequest: boolean; @@ -64,9 +64,9 @@ export interface SteeringCandidateMessage { message: Message; } -export interface SteeringDisposition { - followUp: readonly string[]; - handled: readonly string[]; +export interface SteeringDrainResult { + ack: readonly string[]; + defer: readonly string[]; } export interface ReplyHooks { @@ -74,7 +74,7 @@ export interface ReplyHooks { drainSteeringMessages?: ( accept: ( messages: SteeringCandidateMessage[], - ) => Promise, + ) => Promise, ) => Promise; messageContext?: MessageContext; ack?: () => Promise; @@ -300,21 +300,21 @@ function createAcceptedSteeringDrain( await hooks.drainSteeringMessages!(async (messages) => { const selection = await options.selectMessages(messages, context); await options.onSkipped?.(selection.skipped); - // Follow-ups stay pending so a later worker slice handles them after the - // active answer is delivered. + // Deferred messages stay pending so a later worker slice handles them + // after the active answer is delivered. const interrupted = selection.accepted - .filter((accepted) => accepted.mode === "steer") + .filter((accepted) => accepted.mode === "interrupt") .map((accepted) => accepted.message); await accept(getQueuedMessagesFromSlackMessages(interrupted, options)); interruptedMessages = interrupted; await options.onAcceptedForProcessing?.(interrupted); return { - followUp: selection.accepted - .filter((accepted) => accepted.mode === "follow_up") + defer: selection.accepted + .filter((accepted) => accepted.mode === "defer") .map((accepted) => accepted.inboundMessageId), - handled: [ + ack: [ ...selection.accepted - .filter((accepted) => accepted.mode === "steer") + .filter((accepted) => accepted.mode === "interrupt") .map((accepted) => accepted.inboundMessageId), ...selection.skipped.map((skipped) => skipped.inboundMessageId), ], @@ -566,7 +566,7 @@ export function createSlackTurnRuntime< return { context, decision, - mode: isActiveRequest ? "steer" : "follow_up", + mode: isActiveRequest ? "interrupt" : "defer", text, }; }; diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index 9745b803b..c3c3922b5 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -7,7 +7,7 @@ source of truth. ## State Model - A conversation mailbox contains normalized pending work with a durable routing - mode: `steer`, `follow_up`, or `auto` when runtime policy must decide. + mode: `interrupt`, `defer`, or `auto` when runtime policy must decide. - A queue payload identifies the conversation to wake; persisted conversation work owns destination and routing. - A lease grants one worker temporary execution ownership. @@ -31,8 +31,8 @@ source of truth. 6. Terminal delivery or intentional no-reply completion records the delivered turn before acknowledging work. -New messages that arrive during a run remain durable. Explicit `steer` work is -eligible at the next safe boundary, explicit `follow_up` work waits for the next +New messages that arrive during a run remain durable. Explicit `interrupt` work +is eligible at the next safe boundary, explicit `defer` work waits for the next turn, and `auto` work is classified by runtime policy. ## Queue And Lease Rules diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index cbed96c0d..fd70d4d86 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -11,7 +11,7 @@ import { z } from "zod"; import type { SlackTurnOptions, SteeringCandidateMessage, - SteeringDisposition, + SteeringDrainResult, } from "@/chat/runtime/slack-runtime"; import { isCooperativeTurnYieldError, @@ -567,7 +567,7 @@ export function createSlackResourceEventInboundMessage( createdAtMs: input.event.occurredAtMs, destination, inboundMessageId: `resource-event:${input.subscription.id}:${input.event.eventKey}`, - routing: "follow_up", + routing: "defer", source: "resource_event", receivedAtMs: Date.now(), input: { @@ -839,24 +839,22 @@ export function createSlackConversationWorker( ); } }; - // Explicit follow-ups never enter steering. Auto-routed records are - // offered to runtime policy, while explicit steering bypasses that - // decision through activeRequest. + // Explicitly deferred records never enter steering. Auto-routed records + // are offered to runtime policy, while interrupts bypass that decision. const drainSteeringMessages = async ( accept: ( messages: SteeringCandidateMessage[], - ) => Promise, + ) => Promise, ): Promise => { - await context.attempt.steer(async (pendingRecords) => { - const disposition = await accept( + await context.attempt.drain(async (pendingRecords) => { + const result = await accept( pendingRecords.map((record) => ({ - activeRequest: record.routing === "steer", + activeRequest: record.routing === "interrupt", inboundMessageId: record.inboundMessageId, message: restoreMessage({ adapter, record }), })), ); - await context.attempt.followUp(disposition.followUp); - return disposition.handled; + return result; }); }; @@ -936,7 +934,7 @@ export function buildSlackInboundMessage(args: { routing: args.route === "mention" || isSlackAssistantThreadUserMessage(args.message) - ? "steer" + ? "interrupt" : "auto", source: "slack", createdAtMs: args.message.metadata.dateSent.getTime(), diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 5b4c9b248..7742ccd76 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -83,7 +83,12 @@ export interface AgentInput { text: string; } -export type InboundMessageRouting = "follow_up" | "auto" | "steer"; +export type InboundMessageRouting = "auto" | "defer" | "interrupt"; + +export interface MailboxDrainResult { + ack: readonly string[]; + defer: readonly string[]; +} export interface InboundMessage { attemptCount?: number; @@ -94,7 +99,7 @@ export interface InboundMessage { injectedAtMs?: number; input: AgentInput; receivedAtMs: number; - routing?: InboundMessageRouting; + routing: InboundMessageRouting; source: Source; } @@ -328,8 +333,15 @@ function normalizeInput(value: unknown): AgentInput | undefined { }; } -function normalizeInboundMessageRouting(value: unknown): InboundMessageRouting { - return value === "follow_up" || value === "steer" ? value : "auto"; +function normalizeInboundMessageRouting( + value: unknown, +): InboundMessageRouting | undefined { + if (value === undefined) { + return "auto"; + } + return value === "auto" || value === "defer" || value === "interrupt" + ? value + : undefined; } function normalizeMessage(value: unknown): InboundMessage | undefined { @@ -343,6 +355,7 @@ function normalizeMessage(value: unknown): InboundMessage | undefined { const createdAtMs = toOptionalNumber(value.createdAtMs); const receivedAtMs = toOptionalNumber(value.receivedAtMs); const input = normalizeInput(value.input); + const routing = normalizeInboundMessageRouting(value.routing); if ( !conversationId || !destination || @@ -350,7 +363,8 @@ function normalizeMessage(value: unknown): InboundMessage | undefined { !source || typeof createdAtMs !== "number" || typeof receivedAtMs !== "number" || - !input + !input || + !routing ) { return undefined; } @@ -362,7 +376,7 @@ function normalizeMessage(value: unknown): InboundMessage | undefined { createdAtMs, receivedAtMs, input, - routing: normalizeInboundMessageRouting(value.routing), + routing, injectedAtMs: toOptionalNumber(value.injectedAtMs), attemptCount: toOptionalNumber(value.attemptCount), }; @@ -1491,66 +1505,15 @@ export async function checkInConversationWork(args: { }); } -/** Persist an explicit route for pending mailbox entries under the active lease. */ -export async function routeConversationMailboxMessages(args: { - conversationId: string; - inboundMessageIds: readonly string[]; - leaseToken: string; - nowMs?: number; - routing: Exclude; - state?: StateAdapter; -}): Promise { - if (args.inboundMessageIds.length === 0) { - return; - } - const routedIds = new Set(args.inboundMessageIds); - await withConversationMutation(args, async (state, lock) => { - const current = await readConversation(state, args.conversationId); - if (!current || current.execution.lease?.token !== args.leaseToken) { - throw new Error( - `Conversation lease is not held for ${args.conversationId}`, - ); - } - const pendingIds = new Set( - current.execution.pendingMessages.map( - (message) => message.inboundMessageId, - ), - ); - for (const inboundMessageId of routedIds) { - if (!pendingIds.has(inboundMessageId)) { - throw new Error( - `Conversation mailbox route is not pending for ${args.conversationId}`, - ); - } - } - await writeConversation( - state, - lock, - withExecutionUpdate( - current, - { - ...current.execution, - pendingMessages: current.execution.pendingMessages.map((message) => - routedIds.has(message.inboundMessageId) - ? { ...message, routing: args.routing } - : message, - ), - }, - args.nowMs ?? now(), - ), - ); - }); -} - /** - * Drain pending mailbox entries after the caller acknowledges durable handling. + * Resolve pending mailbox entries after the caller accepts responsibility. * - * Returning ids acknowledges only that subset; returning nothing acknowledges - * every pending entry passed to the handler. + * Acknowledged entries are removed while deferred entries remain for the next + * turn. Returning nothing acknowledges every entry passed to the handler. */ export async function drainConversationMailbox(args: { conversationId: string; - handle: (messages: InboundMessage[]) => Promise; + handle: (messages: InboundMessage[]) => Promise; leaseToken: string; nowMs?: number; state?: StateAdapter; @@ -1569,20 +1532,28 @@ export async function drainConversationMailbox(args: { return []; } - const acknowledgedIds = await args.handle(pending); + const result = await args.handle(pending); const pendingIds = new Set( pending.map((message) => message.inboundMessageId), ); - for (const inboundMessageId of acknowledgedIds ?? []) { + const acknowledgedIds = new Set( + result?.ack ?? pending.map((message) => message.inboundMessageId), + ); + const deferredIds = new Set(result?.defer ?? []); + for (const inboundMessageId of [...acknowledgedIds, ...deferredIds]) { if (!pendingIds.has(inboundMessageId)) { throw new Error( - `Conversation mailbox acknowledgement is not pending for ${args.conversationId}`, + `Conversation mailbox drain result is not pending for ${args.conversationId}`, + ); + } + } + for (const inboundMessageId of deferredIds) { + if (acknowledgedIds.has(inboundMessageId)) { + throw new Error( + `Conversation mailbox drain result overlaps for ${args.conversationId}`, ); } } - const drainedIds = new Set( - acknowledgedIds ?? pending.map((message) => message.inboundMessageId), - ); await withConversationMutation(args, async (state, lock) => { const current = await readConversation(state, args.conversationId); @@ -1591,9 +1562,13 @@ export async function drainConversationMailbox(args: { `Conversation lease is not held for ${args.conversationId}`, ); } - const pendingMessages = current.execution.pendingMessages.filter( - (message) => !drainedIds.has(message.inboundMessageId), - ); + const pendingMessages = current.execution.pendingMessages + .filter((message) => !acknowledgedIds.has(message.inboundMessageId)) + .map((message) => + deferredIds.has(message.inboundMessageId) + ? { ...message, routing: "defer" as const } + : message, + ); await writeConversation( state, lock, @@ -1612,7 +1587,9 @@ export async function drainConversationMailbox(args: { ), ); }); - return pending.filter((message) => drainedIds.has(message.inboundMessageId)); + return pending.filter((message) => + acknowledgedIds.has(message.inboundMessageId), + ); } /** Acknowledge leased mailbox entries after the handler accepts responsibility. */ diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 07e8f1a01..eec156a96 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -24,6 +24,7 @@ export { type InboundMessage, type InboundMessageRouting, type Lease, + type MailboxDrainResult, type RequestConversationWorkResult, type Source, type StartConversationWorkAcquired, @@ -363,18 +364,6 @@ export async function checkInConversationWork(args: { return result; } -/** Persist an explicit route for pending mailbox entries under the active lease. */ -export async function routeConversationMailboxMessages( - args: Parameters[0] & { - conversationStore?: ConversationStore; - state?: StateAdapter; - }, -) { - const result = await workState.routeConversationMailboxMessages(args); - await recordExecutionMetadata(args); - return result; -} - /** Drain pending mailbox entries after the caller accepts responsibility. */ export async function drainConversationMailbox( args: Parameters[0] & { @@ -382,8 +371,18 @@ export async function drainConversationMailbox( state?: StateAdapter; }, ) { - const result = await workState.drainConversationMailbox(args); - if (result.length > 0) { + let changed = false; + const result = await workState.drainConversationMailbox({ + ...args, + handle: async (messages) => { + const result = await args.handle(messages); + changed = result + ? result.ack.length > 0 || result.defer.length > 0 + : messages.length > 0; + return result; + }, + }); + if (changed) { await recordExecutionMetadata(args); } return result; diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 042573e60..8e0a20c42 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -24,11 +24,11 @@ import { recordAttemptFailure, releaseConversationWork, requestConversationContinuation, - routeConversationMailboxMessages, startConversationWork, type AttemptFailure, type ConversationWorkState, type InboundMessage, + type MailboxDrainResult, } from "./store"; export const CONVERSATION_WORK_DEFER_DELAY_MS = 15_000; @@ -46,12 +46,11 @@ export interface InboxAttempt { ack(): Promise; conversationId: string; destination: Destination; - followUp(inboundMessageIds: readonly string[]): Promise; + drain( + handle: (messages: InboundMessage[]) => Promise, + ): Promise; isFinalAttempt: boolean; messages: InboundMessage[]; - steer( - handle: (messages: InboundMessage[]) => Promise, - ): Promise; } export interface ConversationWorkerResult { @@ -340,8 +339,8 @@ export async function processConversationWork( "Conversation work lease acquired", ); - const steer = ( - handle: (messages: InboundMessage[]) => Promise, + const drain = ( + handle: (messages: InboundMessage[]) => Promise, ) => drainConversationMailbox({ conversationId, @@ -349,30 +348,22 @@ export async function processConversationWork( conversationStore: options.conversationStore, handle: async (messages) => { const candidates = messages.filter( - (message) => message.routing !== "follow_up", + (message) => message.routing !== "defer", ); + if (candidates.length === 0) { + return { ack: [], defer: [] }; + } return ( - (await handle(candidates)) ?? - candidates.map((message) => message.inboundMessageId) + (await handle(candidates)) ?? { + ack: candidates.map((message) => message.inboundMessageId), + defer: [], + } ); }, nowMs: now(options), state: options.state, }); - const followUp = async ( - inboundMessageIds: readonly string[], - ): Promise => { - await routeConversationMailboxMessages({ - conversationId, - inboundMessageIds, - leaseToken: lease.leaseToken, - routing: "follow_up", - nowMs: now(options), - state: options.state, - }); - }; - const ack = async (): Promise => { const acknowledged = await ackMessages({ conversationId, @@ -395,12 +386,11 @@ export async function processConversationWork( ack, conversationId, destination, - followUp, + drain, isFinalAttempt: attemptMessages.some((message) => isFinalAttempt(message), ), messages: attemptMessages, - steer, }, conversationId, destination, diff --git a/packages/junior/tests/component/conversation-sql-store.test.ts b/packages/junior/tests/component/conversation-sql-store.test.ts index 9df7f2c72..f1caaa4b2 100644 --- a/packages/junior/tests/component/conversation-sql-store.test.ts +++ b/packages/junior/tests/component/conversation-sql-store.test.ts @@ -1025,7 +1025,7 @@ WHERE conversation_id = $1 conversationStore: store, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; diff --git a/packages/junior/tests/component/services/turn-session-record.test.ts b/packages/junior/tests/component/services/turn-session-record.test.ts index d293f8913..98bbcc7c7 100644 --- a/packages/junior/tests/component/services/turn-session-record.test.ts +++ b/packages/junior/tests/component/services/turn-session-record.test.ts @@ -131,6 +131,7 @@ describe("persistAuthPauseSessionRecord", () => { text: "start", }, receivedAtMs: 9_000, + routing: "auto", source: "slack", }, nowMs: 9_000, diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 86a90c07c..d58d75f4b 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -573,7 +573,7 @@ describe("conversation work execution", () => { queue, run: async (context) => { runs += 1; - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; @@ -613,7 +613,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; @@ -725,7 +725,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); currentNowMs = 2_000; await requestConversationWork({ conversationId: context.conversationId, @@ -761,7 +761,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); currentNowMs = 2_000; await requestConversationWork({ conversationId: context.conversationId, @@ -806,7 +806,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); currentNowMs = 2_000; await requestConversationWork({ conversationId: context.conversationId, @@ -930,7 +930,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); currentNowMs = 2_000; await scheduleAgentContinue( { @@ -1090,7 +1090,7 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - injected.push(await context.attempt.steer(async () => {})); + injected.push(await context.attempt.drain(async () => {})); return { status: "completed" }; }, }), @@ -1123,11 +1123,12 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - await context.attempt.steer(async (messages) => { + await context.attempt.drain(async (messages) => { injected.push(messages.map((message) => message.inboundMessageId)); - return messages - .filter((message) => message.inboundMessageId === "m1") - .map((message) => message.inboundMessageId); + return { + ack: ["m1"], + defer: [], + }; }); return { status: "completed" }; }, @@ -1144,7 +1145,7 @@ describe("conversation work execution", () => { ]); }); - it("routes explicit follow-ups separately from steering", async () => { + it("keeps deferred messages separate from interruptions", async () => { const queue = createConversationWorkQueueTestAdapter(); await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); await appendInboundMessage({ @@ -1154,29 +1155,32 @@ describe("conversation work execution", () => { }), nowMs: 2_100, }); - const steered: string[][] = []; + const drained: string[][] = []; await expect( processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - await context.attempt.followUp(["m2"]); - await context.attempt.steer(async (messages) => { - steered.push(messages.map((message) => message.inboundMessageId)); + await context.attempt.drain(async (messages) => { + drained.push(messages.map((message) => message.inboundMessageId)); + return { + ack: ["m1"], + defer: ["m2"], + }; }); return { status: "completed" }; }, }), ).resolves.toEqual({ status: "pending_requeued" }); - expect(steered).toEqual([["m1"]]); + expect(drained).toEqual([["m1", "m2"]]); const state = await getConversationWorkState({ conversationId: CONVERSATION_ID, }); expect(state?.messages).toEqual([ expect.objectContaining({ inboundMessageId: "m2", - routing: "follow_up", + routing: "defer", }), ]); }); @@ -1191,7 +1195,10 @@ describe("conversation work execution", () => { queue, run: async (context) => { try { - await context.attempt.steer(async () => ["different-message"]); + await context.attempt.drain(async () => ({ + ack: ["different-message"], + defer: [], + })); } catch (error) { drainError = error; throw error; @@ -1202,7 +1209,7 @@ describe("conversation work execution", () => { ).resolves.toEqual({ status: "failed" }); expect(String(drainError)).toContain( - "Conversation mailbox acknowledgement is not pending for", + "Conversation mailbox drain result is not pending for", ); const state = await getConversationWorkState({ @@ -1232,7 +1239,7 @@ describe("conversation work execution", () => { queue, state: observed.state, run: async (context) => { - const drain = context.attempt.steer(async () => { + const drain = context.attempt.drain(async () => { expect(observed.isHeld()).toBe(false); injectionStarted.resolve(); await finishInjection.promise; @@ -1281,7 +1288,7 @@ describe("conversation work execution", () => { checkInIntervalMs: 15_000, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); entered.resolve(); await finish.promise; return { status: "completed" }; @@ -1321,7 +1328,7 @@ describe("conversation work execution", () => { checkInIntervalMs: 15_000, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); entered.resolve({ shouldYield: context.shouldYield, }); @@ -1557,7 +1564,7 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - const first = await context.attempt.steer(async () => {}); + const first = await context.attempt.drain(async () => {}); injected.push(first.map((message) => message.inboundMessageId)); await appendInboundMessage({ message: inboundMessage("m2", { @@ -1566,7 +1573,7 @@ describe("conversation work execution", () => { }), nowMs: 2_100, }); - const second = await context.attempt.steer(async () => {}); + const second = await context.attempt.drain(async () => {}); injected.push(second.map((message) => message.inboundMessageId)); return { status: "completed" }; }, @@ -1584,7 +1591,7 @@ describe("conversation work execution", () => { processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); await appendInboundMessage({ message: inboundMessage("m2", { createdAtMs: 2_000, @@ -1592,7 +1599,7 @@ describe("conversation work execution", () => { }), nowMs: 2_100, }); - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); return { status: "completed" }; }, }), @@ -1615,7 +1622,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); currentNowMs = 2_100; await appendInboundMessage({ message: inboundMessage("m2", { @@ -1646,7 +1653,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, run: async (context) => { - await context.attempt.steer(async () => {}); + await context.attempt.drain(async () => {}); currentNowMs = 242_000; expect(context.shouldYield()).toBe(true); return { status: "yielded" }; @@ -1843,7 +1850,7 @@ describe("conversation work execution", () => { processConversationQueueMessage(conversationQueueMessage(), { queue, run: async (context) => { - const messages = await context.attempt.steer(async () => {}); + const messages = await context.attempt.drain(async () => {}); injected.push(...messages.map((message) => message.inboundMessageId)); return { status: "completed" }; }, diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index e4d4bcb4d..ccd8ecb1e 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -271,6 +271,7 @@ describe("Slack conversation work execution", () => { ...conversationQueueMessage(), destination: SLACK_DESTINATION, inboundMessageId: "malformed-slack-metadata", + routing: "auto" as const, source: "slack" as const, createdAtMs: 1_000, receivedAtMs: 1_100, @@ -343,10 +344,9 @@ describe("Slack conversation work execution", () => { ack: async () => {}, conversationId: CONVERSATION_ID, destination: SLACK_DESTINATION, - followUp: async () => {}, + drain: async () => [], isFinalAttempt: false, messages: [malformed], - steer: async () => [], }, checkIn: async () => true, conversationId: CONVERSATION_ID, @@ -906,8 +906,8 @@ describe("Slack conversation work execution", () => { await hooks.drainSteeringMessages?.(async (steering) => { injected.push(steering.map((candidate) => candidate.message.id)); return { - followUp: [], - handled: steering.map((candidate) => candidate.inboundMessageId), + ack: steering.map((candidate) => candidate.inboundMessageId), + defer: [], }; }); }, @@ -943,7 +943,7 @@ describe("Slack conversation work execution", () => { }); }); - it("leaves resource events queued as follow-up work during an active turn", async () => { + it("leaves resource events deferred during an active turn", async () => { const queue = createConversationWorkQueueTestAdapter(); const state = getStateAdapter(); await state.connect(); @@ -989,8 +989,8 @@ describe("Slack conversation work execution", () => { await hooks.drainSteeringMessages?.(async (steering) => { observed.push(steering.map((candidate) => candidate.message.id)); return { - followUp: [], - handled: steering.map((candidate) => candidate.inboundMessageId), + ack: steering.map((candidate) => candidate.inboundMessageId), + defer: [], }; }); }, @@ -1008,14 +1008,14 @@ describe("Slack conversation work execution", () => { }), ).resolves.toEqual({ status: "pending_requeued" }); - expect(observed).toEqual([[]]); + expect(observed).toEqual([]); const work = await getConversationWorkState({ conversationId: CONVERSATION_ID, state, }); expect(work?.execution.pendingMessages).toEqual([ expect.objectContaining({ - routing: "follow_up", + routing: "defer", source: "resource_event", }), ]); @@ -1103,8 +1103,8 @@ describe("Slack conversation work execution", () => { })), ); return { - followUp: [], - handled: steering.map((candidate) => candidate.inboundMessageId), + ack: steering.map((candidate) => candidate.inboundMessageId), + defer: [], }; }); }, diff --git a/packages/junior/tests/component/vercel-queue-dev.test.ts b/packages/junior/tests/component/vercel-queue-dev.test.ts index 8be603641..80ae7e95c 100644 --- a/packages/junior/tests/component/vercel-queue-dev.test.ts +++ b/packages/junior/tests/component/vercel-queue-dev.test.ts @@ -283,6 +283,7 @@ describe("registerVercelConversationWorkDevConsumer", () => { message: { conversationId: "slack:C123:1712345.0001", inboundMessageId: "m1", + routing: "auto", destination: { channelId: "C123", platform: "slack", diff --git a/packages/junior/tests/fixtures/conversation-work.ts b/packages/junior/tests/fixtures/conversation-work.ts index 0a6e58946..a1d9729b6 100644 --- a/packages/junior/tests/fixtures/conversation-work.ts +++ b/packages/junior/tests/fixtures/conversation-work.ts @@ -197,6 +197,7 @@ export function inboundMessage( conversationId: CONVERSATION_ID, inboundMessageId, destination: SLACK_DESTINATION, + routing: "auto", source: "slack", createdAtMs: 1_000, receivedAtMs: 1_100, From 495295aacbab71aa2322f4da4c2fd349f1892fc6 Mon Sep 17 00:00:00 2001 From: David Cramer Date: Thu, 23 Jul 2026 10:13:32 -0700 Subject: [PATCH 06/10] ref(runtime): Rename mailbox routing to delivery Describe how pending input is delivered into turn execution without overloading routing terminology used for destinations and handlers. --- TERMINOLOGY.md | 2 +- .../junior/src/chat/task-execution/README.md | 6 +++--- .../junior/src/chat/task-execution/slack-work.ts | 10 +++++----- packages/junior/src/chat/task-execution/state.ts | 16 ++++++++-------- packages/junior/src/chat/task-execution/store.ts | 2 +- .../junior/src/chat/task-execution/worker.ts | 2 +- .../services/turn-session-record.test.ts | 2 +- .../task-execution/conversation-work.test.ts | 2 +- .../slack-conversation-work.test.ts | 4 ++-- .../tests/component/vercel-queue-dev.test.ts | 2 +- .../junior/tests/fixtures/conversation-work.ts | 2 +- 11 files changed, 25 insertions(+), 25 deletions(-) diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index 8c5185963..bb950cac2 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -17,7 +17,7 @@ Canonical words used across Junior's code and documentation. batched into that turn. - **Follow-up message**: a user message that waits for the active turn to finish before starting the next turn. -- **Message routing**: the durable mailbox instruction for an inbound message: +- **Mailbox delivery**: the durable instruction for an inbound message: `interrupt` is eligible at the next safe boundary, `defer` waits for the active turn to finish, and `auto` asks runtime policy to choose between them. - **Turn**: one request-to-final-response cycle. It may span multiple runs and diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index c3c3922b5..c70ea67e5 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -6,10 +6,10 @@ source of truth. ## State Model -- A conversation mailbox contains normalized pending work with a durable routing - mode: `interrupt`, `defer`, or `auto` when runtime policy must decide. +- A conversation mailbox contains normalized pending work with a durable + delivery mode: `interrupt`, `defer`, or `auto` when runtime policy must decide. - A queue payload identifies the conversation to wake; persisted conversation - work owns destination and routing. + work owns destination and delivery. - A lease grants one worker temporary execution ownership. - Check-ins extend active ownership and allow heartbeat recovery to distinguish slow work from abandoned work. diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index fd70d4d86..b0e6d0802 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -567,7 +567,7 @@ export function createSlackResourceEventInboundMessage( createdAtMs: input.event.occurredAtMs, destination, inboundMessageId: `resource-event:${input.subscription.id}:${input.event.eventKey}`, - routing: "defer", + delivery: "defer", source: "resource_event", receivedAtMs: Date.now(), input: { @@ -839,8 +839,8 @@ export function createSlackConversationWorker( ); } }; - // Explicitly deferred records never enter steering. Auto-routed records - // are offered to runtime policy, while interrupts bypass that decision. + // Explicitly deferred records never enter steering. Auto-delivered + // records are offered to runtime policy, while interrupts bypass it. const drainSteeringMessages = async ( accept: ( messages: SteeringCandidateMessage[], @@ -849,7 +849,7 @@ export function createSlackConversationWorker( await context.attempt.drain(async (pendingRecords) => { const result = await accept( pendingRecords.map((record) => ({ - activeRequest: record.routing === "interrupt", + activeRequest: record.delivery === "interrupt", inboundMessageId: record.inboundMessageId, message: restoreMessage({ adapter, record }), })), @@ -931,7 +931,7 @@ export function buildSlackInboundMessage(args: { args.conversationId, args.message.id, ].join(":"), - routing: + delivery: args.route === "mention" || isSlackAssistantThreadUserMessage(args.message) ? "interrupt" diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 7742ccd76..24dbfb0f6 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -83,7 +83,7 @@ export interface AgentInput { text: string; } -export type InboundMessageRouting = "auto" | "defer" | "interrupt"; +export type InboundMessageDelivery = "auto" | "defer" | "interrupt"; export interface MailboxDrainResult { ack: readonly string[]; @@ -99,7 +99,7 @@ export interface InboundMessage { injectedAtMs?: number; input: AgentInput; receivedAtMs: number; - routing: InboundMessageRouting; + delivery: InboundMessageDelivery; source: Source; } @@ -333,9 +333,9 @@ function normalizeInput(value: unknown): AgentInput | undefined { }; } -function normalizeInboundMessageRouting( +function normalizeInboundMessageDelivery( value: unknown, -): InboundMessageRouting | undefined { +): InboundMessageDelivery | undefined { if (value === undefined) { return "auto"; } @@ -355,7 +355,7 @@ function normalizeMessage(value: unknown): InboundMessage | undefined { const createdAtMs = toOptionalNumber(value.createdAtMs); const receivedAtMs = toOptionalNumber(value.receivedAtMs); const input = normalizeInput(value.input); - const routing = normalizeInboundMessageRouting(value.routing); + const delivery = normalizeInboundMessageDelivery(value.delivery); if ( !conversationId || !destination || @@ -364,7 +364,7 @@ function normalizeMessage(value: unknown): InboundMessage | undefined { typeof createdAtMs !== "number" || typeof receivedAtMs !== "number" || !input || - !routing + !delivery ) { return undefined; } @@ -376,7 +376,7 @@ function normalizeMessage(value: unknown): InboundMessage | undefined { createdAtMs, receivedAtMs, input, - routing, + delivery, injectedAtMs: toOptionalNumber(value.injectedAtMs), attemptCount: toOptionalNumber(value.attemptCount), }; @@ -1566,7 +1566,7 @@ export async function drainConversationMailbox(args: { .filter((message) => !acknowledgedIds.has(message.inboundMessageId)) .map((message) => deferredIds.has(message.inboundMessageId) - ? { ...message, routing: "defer" as const } + ? { ...message, delivery: "defer" as const } : message, ); await writeConversation( diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index eec156a96..7f69b8c5f 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -22,7 +22,7 @@ export { type ConversationWorkState, type ExecutionStatus, type InboundMessage, - type InboundMessageRouting, + type InboundMessageDelivery, type Lease, type MailboxDrainResult, type RequestConversationWorkResult, diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 8e0a20c42..60a457a3b 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -348,7 +348,7 @@ export async function processConversationWork( conversationStore: options.conversationStore, handle: async (messages) => { const candidates = messages.filter( - (message) => message.routing !== "defer", + (message) => message.delivery !== "defer", ); if (candidates.length === 0) { return { ack: [], defer: [] }; diff --git a/packages/junior/tests/component/services/turn-session-record.test.ts b/packages/junior/tests/component/services/turn-session-record.test.ts index 98bbcc7c7..69cbb0eaf 100644 --- a/packages/junior/tests/component/services/turn-session-record.test.ts +++ b/packages/junior/tests/component/services/turn-session-record.test.ts @@ -131,7 +131,7 @@ describe("persistAuthPauseSessionRecord", () => { text: "start", }, receivedAtMs: 9_000, - routing: "auto", + delivery: "auto", source: "slack", }, nowMs: 9_000, diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index d58d75f4b..33d998eb7 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -1180,7 +1180,7 @@ describe("conversation work execution", () => { expect(state?.messages).toEqual([ expect.objectContaining({ inboundMessageId: "m2", - routing: "defer", + delivery: "defer", }), ]); }); diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index ccd8ecb1e..cd4fd1de6 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -271,7 +271,7 @@ describe("Slack conversation work execution", () => { ...conversationQueueMessage(), destination: SLACK_DESTINATION, inboundMessageId: "malformed-slack-metadata", - routing: "auto" as const, + delivery: "auto" as const, source: "slack" as const, createdAtMs: 1_000, receivedAtMs: 1_100, @@ -1015,7 +1015,7 @@ describe("Slack conversation work execution", () => { }); expect(work?.execution.pendingMessages).toEqual([ expect.objectContaining({ - routing: "defer", + delivery: "defer", source: "resource_event", }), ]); diff --git a/packages/junior/tests/component/vercel-queue-dev.test.ts b/packages/junior/tests/component/vercel-queue-dev.test.ts index 80ae7e95c..b5fc44d9f 100644 --- a/packages/junior/tests/component/vercel-queue-dev.test.ts +++ b/packages/junior/tests/component/vercel-queue-dev.test.ts @@ -283,7 +283,7 @@ describe("registerVercelConversationWorkDevConsumer", () => { message: { conversationId: "slack:C123:1712345.0001", inboundMessageId: "m1", - routing: "auto", + delivery: "auto", destination: { channelId: "C123", platform: "slack", diff --git a/packages/junior/tests/fixtures/conversation-work.ts b/packages/junior/tests/fixtures/conversation-work.ts index a1d9729b6..cbdfc3538 100644 --- a/packages/junior/tests/fixtures/conversation-work.ts +++ b/packages/junior/tests/fixtures/conversation-work.ts @@ -197,7 +197,7 @@ export function inboundMessage( conversationId: CONVERSATION_ID, inboundMessageId, destination: SLACK_DESTINATION, - routing: "auto", + delivery: "auto", source: "slack", createdAtMs: 1_000, receivedAtMs: 1_100, From 66914c58219ec20aff4c4519f5700810fa1e659f Mon Sep 17 00:00:00 2001 From: David Cramer Date: Thu, 23 Jul 2026 11:16:01 -0700 Subject: [PATCH 07/10] fix(runtime): Harden mailbox delivery persistence Version and strictly validate durable mailbox messages, preserve interrupt delivery across duplicate ingress, and keep metadata mirroring aligned with actual drain mutations. --- .../junior/src/chat/runtime/slack-runtime.ts | 13 +- .../junior/src/chat/task-execution/README.md | 9 + .../src/chat/task-execution/slack-work.ts | 22 +- .../junior/src/chat/task-execution/state.ts | 255 ++++++++---------- .../junior/src/chat/task-execution/store.ts | 16 +- .../task-execution/conversation-work.test.ts | 112 ++++++++ .../slack-conversation-work.test.ts | 20 +- 7 files changed, 275 insertions(+), 172 deletions(-) diff --git a/packages/junior/src/chat/runtime/slack-runtime.ts b/packages/junior/src/chat/runtime/slack-runtime.ts index 9484456d6..278eb962f 100644 --- a/packages/junior/src/chat/runtime/slack-runtime.ts +++ b/packages/junior/src/chat/runtime/slack-runtime.ts @@ -59,14 +59,15 @@ export interface AssistantLifecycleEvent { type SteeringMode = "defer" | "interrupt"; export interface SteeringCandidateMessage { - activeRequest: boolean; + delivery: "auto" | "interrupt"; inboundMessageId: string; message: Message; } +/** Runtime-owned handling result for drained Slack steering candidates. */ export interface SteeringDrainResult { - ack: readonly string[]; - defer: readonly string[]; + deferred: readonly string[]; + handled: readonly string[]; } export interface ReplyHooks { @@ -309,10 +310,10 @@ function createAcceptedSteeringDrain( interruptedMessages = interrupted; await options.onAcceptedForProcessing?.(interrupted); return { - defer: selection.accepted + deferred: selection.accepted .filter((accepted) => accepted.mode === "defer") .map((accepted) => accepted.inboundMessageId), - ack: [ + handled: [ ...selection.accepted .filter((accepted) => accepted.mode === "interrupt") .map((accepted) => accepted.inboundMessageId), @@ -552,7 +553,7 @@ export function createSlackTurnRuntime< userText: appendSlackLegacyAttachmentText(strippedUserText, message.raw), }; const isActiveRequest = - candidate.activeRequest || Boolean(message.isMention); + candidate.delivery === "interrupt" || Boolean(message.isMention); const decision = await deps.decideSubscribedReply({ rawText: text.rawText, diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index c70ea67e5..01ebcae31 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -15,6 +15,9 @@ source of truth. slow work from abandoned work. - Delivery state prevents a completed turn from being posted twice. +Schema-v1 mailbox entries without `delivery` migrate to `auto`. Schema-v2 +entries require a valid delivery mode and reject invalid pending work. + `state.ts`, `store.ts`, and their runtime schemas define the persisted shapes. ## Execution @@ -35,6 +38,12 @@ New messages that arrive during a run remain durable. Explicit `interrupt` work is eligible at the next safe boundary, explicit `defer` work waits for the next turn, and `auto` work is classified by runtime policy. +Slack mentions and assistant-thread user messages use `interrupt`, subscribed +resource events use `defer`, and other subscribed messages use `auto`. + +One lease-bound drain removes disjoint `ack` ids and retains disjoint `defer` +ids with `delivery: "defer"`. + ## Queue And Lease Rules - Duplicate queue delivery is expected and must be idempotent. diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index b0e6d0802..244cbaa78 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -848,13 +848,23 @@ export function createSlackConversationWorker( ): Promise => { await context.attempt.drain(async (pendingRecords) => { const result = await accept( - pendingRecords.map((record) => ({ - activeRequest: record.delivery === "interrupt", - inboundMessageId: record.inboundMessageId, - message: restoreMessage({ adapter, record }), - })), + pendingRecords.map((record) => { + if (record.delivery === "defer") { + throw new Error( + "Deferred mailbox message reached Slack steering", + ); + } + return { + delivery: record.delivery, + inboundMessageId: record.inboundMessageId, + message: restoreMessage({ adapter, record }), + }; + }), ); - return result; + return { + ack: result.handled, + defer: result.deferred, + }; }); }; diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 24dbfb0f6..ff4489165 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -9,7 +9,8 @@ */ import { randomUUID } from "node:crypto"; import type { Lock, StateAdapter } from "chat"; -import type { Destination } from "@sentry/junior-plugin-api"; +import { destinationSchema, type Destination } from "@sentry/junior-plugin-api"; +import { z } from "zod"; import { isRecord, toOptionalNumber, toOptionalString } from "@/chat/coerce"; import { getChatConfig } from "@/chat/config"; import { parseDestination, sameDestination } from "@/chat/destination"; @@ -21,7 +22,7 @@ import { import { JUNIOR_THREAD_STATE_TTL_MS } from "@/chat/state/ttl"; const CONVERSATION_PREFIX = "junior:conversation"; -const CONVERSATION_SCHEMA_VERSION = 1; +const CONVERSATION_SCHEMA_VERSION = 2; const CONVERSATION_ACTIVITY_INDEX_MAX_LENGTH = 10_000; const CONVERSATION_INDEX_LOCK_TTL_MS = 10_000; const CONVERSATION_INDEX_LOCK_WAIT_MS = 2_000; @@ -60,14 +61,17 @@ export const CONVERSATION_WORK_CHECK_IN_INTERVAL_MS = 15_000; export const CONVERSATION_WORK_STALE_ENQUEUE_MS = 60_000; export const CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS = 5; -export type Source = - | "api" - | "internal" - | "local" - | "plugin" - | "resource_event" - | "scheduler" - | "slack"; +const inboundMessageSourceSchema = z.enum([ + "api", + "internal", + "local", + "plugin", + "resource_event", + "scheduler", + "slack", +]); + +export type Source = z.output; export type ExecutionStatus = | "awaiting_resume" @@ -76,32 +80,54 @@ export type ExecutionStatus = | "pending" | "running"; -export interface AgentInput { - attachments?: unknown[]; - authorId?: string; - metadata?: Record; - text: string; -} +const agentInputSchema = z + .object({ + attachments: z.array(z.unknown()).optional(), + authorId: z.string().min(1).optional(), + metadata: z.record(z.string(), z.unknown()).optional(), + text: z.string(), + }) + .strict() + .refine((input) => input.text.trim() || input.attachments?.length, { + message: "agent input requires text or attachments", + }); + +export type AgentInput = z.output; -export type InboundMessageDelivery = "auto" | "defer" | "interrupt"; +/** Durable delivery modes for pending inbound mailbox work. */ +export const inboundMessageDeliverySchema = z.enum([ + "auto", + "defer", + "interrupt", +]); +export type InboundMessageDelivery = z.output< + typeof inboundMessageDeliverySchema +>; + +/** Mailbox ids to remove or retain as deferred in one lease-bound drain. */ export interface MailboxDrainResult { ack: readonly string[]; defer: readonly string[]; } -export interface InboundMessage { - attemptCount?: number; - conversationId: string; - createdAtMs: number; - destination: Destination; - inboundMessageId: string; - injectedAtMs?: number; - input: AgentInput; - receivedAtMs: number; - delivery: InboundMessageDelivery; - source: Source; -} +/** Canonical durable mailbox entry owned by task execution. */ +export const inboundMessageSchema = z + .object({ + attemptCount: z.number().finite().optional(), + conversationId: z.string().refine((value) => value.trim().length > 0), + createdAtMs: z.number().finite(), + delivery: inboundMessageDeliverySchema, + destination: destinationSchema, + inboundMessageId: z.string().refine((value) => value.trim().length > 0), + injectedAtMs: z.number().finite().optional(), + input: agentInputSchema, + receivedAtMs: z.number().finite(), + source: inboundMessageSourceSchema, + }) + .strict(); + +export type InboundMessage = z.output; export interface Lease { acquiredAtMs: number; @@ -130,7 +156,7 @@ export interface Conversation { execution: ConversationExecution; lastActivityAtMs: number; actor?: StoredSlackActor; - schemaVersion: 1; + schemaVersion: 2; source?: Source; title?: string; updatedAtMs: number; @@ -244,13 +270,18 @@ function upgradedPendingMessage( stored: InboundMessage, duplicate: InboundMessage, ): InboundMessage { - if ( - !inputHasAttachments(duplicate.input) || - inputHasAttachments(stored.input) - ) { + const input = + inputHasAttachments(duplicate.input) && !inputHasAttachments(stored.input) + ? duplicate.input + : stored.input; + const delivery = + stored.delivery === "auto" && duplicate.delivery === "interrupt" + ? "interrupt" + : stored.delivery; + if (input === stored.input && delivery === stored.delivery) { return stored; } - return { ...stored, input: duplicate.input }; + return { ...stored, delivery, input }; } function compareIndexDescending( @@ -278,18 +309,8 @@ function uniqueStrings(values: string[]): string[] { } function normalizeSource(value: unknown): Source | undefined { - if ( - value === "api" || - value === "internal" || - value === "local" || - value === "plugin" || - value === "resource_event" || - value === "scheduler" || - value === "slack" - ) { - return value; - } - return undefined; + const parsed = inboundMessageSourceSchema.safeParse(value); + return parsed.success ? parsed.data : undefined; } function normalizeExecutionStatus(value: unknown): ExecutionStatus | undefined { @@ -305,81 +326,17 @@ function normalizeExecutionStatus(value: unknown): ExecutionStatus | undefined { return undefined; } -function normalizeMetadata( +function normalizeMessage( value: unknown, -): Record | undefined { - if (!isRecord(value)) { - return undefined; - } - return value; -} - -function normalizeInput(value: unknown): AgentInput | undefined { - if (!isRecord(value)) { - return undefined; - } - const text = typeof value.text === "string" ? value.text : undefined; - const attachments = Array.isArray(value.attachments) - ? [...value.attachments] - : undefined; - if (text === undefined || (!text.trim() && !attachments?.length)) { - return undefined; - } - return { - text, - authorId: toOptionalString(value.authorId), - attachments, - metadata: normalizeMetadata(value.metadata), - }; -} - -function normalizeInboundMessageDelivery( - value: unknown, -): InboundMessageDelivery | undefined { - if (value === undefined) { - return "auto"; - } - return value === "auto" || value === "defer" || value === "interrupt" - ? value - : undefined; -} - -function normalizeMessage(value: unknown): InboundMessage | undefined { - if (!isRecord(value)) { - return undefined; - } - const conversationId = toOptionalString(value.conversationId); - const inboundMessageId = toOptionalString(value.inboundMessageId); - const source = normalizeSource(value.source); - const destination = parseDestination(value.destination); - const createdAtMs = toOptionalNumber(value.createdAtMs); - const receivedAtMs = toOptionalNumber(value.receivedAtMs); - const input = normalizeInput(value.input); - const delivery = normalizeInboundMessageDelivery(value.delivery); - if ( - !conversationId || - !destination || - !inboundMessageId || - !source || - typeof createdAtMs !== "number" || - typeof receivedAtMs !== "number" || - !input || - !delivery - ) { - return undefined; - } - return { - conversationId, - destination, - inboundMessageId, - source, - createdAtMs, - receivedAtMs, - input, - delivery, - injectedAtMs: toOptionalNumber(value.injectedAtMs), - attemptCount: toOptionalNumber(value.attemptCount), - }; + schemaVersion: 1 | 2, +): InboundMessage | undefined { + // TODO(v0.110.0): Remove schema-v1 mailbox delivery migration after old records expire. + const candidate = + schemaVersion === 1 && isRecord(value) && value.delivery === undefined + ? { ...value, delivery: "auto" } + : value; + const parsed = inboundMessageSchema.safeParse(candidate); + return parsed.success ? parsed.data : undefined; } /** Whether this is the final attempt before an unacked message is dead-lettered. */ @@ -428,6 +385,7 @@ function normalizeLease(value: unknown): Lease | undefined { function normalizeExecution( conversationId: string, value: unknown, + schemaVersion: 1 | 2, ): ConversationExecution | undefined { if (!isRecord(value)) { return undefined; @@ -436,14 +394,21 @@ function normalizeExecution( if (!status) { return undefined; } - const pendingMessages = Array.isArray(value.pendingMessages) - ? value.pendingMessages - .map(normalizeMessage) - .filter((message): message is InboundMessage => Boolean(message)) - .filter((message) => message.conversationId === conversationId) - .filter((message) => message.injectedAtMs === undefined) - .sort(compareMessages) - : []; + const pendingMessages: InboundMessage[] = []; + if (Array.isArray(value.pendingMessages)) { + for (const rawMessage of value.pendingMessages) { + const message = normalizeMessage(rawMessage, schemaVersion); + if ( + !message || + message.conversationId !== conversationId || + message.injectedAtMs !== undefined + ) { + return undefined; + } + pendingMessages.push(message); + } + pendingMessages.sort(compareMessages); + } const inboundMessageIds = Array.isArray(value.inboundMessageIds) ? uniqueStrings( value.inboundMessageIds @@ -477,21 +442,29 @@ function normalizeExecution( } /** - * Decode schema-v1 conversation records and reject runnable mailbox state whose - * pending entries do not belong to the conversation destination. + * Decode current conversation records and migrate schema-v1 mailbox delivery. */ function normalizeConversation( conversationId: string, value: unknown, ): Conversation | undefined { - if (!isRecord(value) || value.schemaVersion !== CONVERSATION_SCHEMA_VERSION) { + if ( + !isRecord(value) || + (value.schemaVersion !== 1 && + value.schemaVersion !== CONVERSATION_SCHEMA_VERSION) + ) { return undefined; } + const schemaVersion = value.schemaVersion; const storedConversationId = toOptionalString(value.conversationId); const createdAtMs = toOptionalNumber(value.createdAtMs); const lastActivityAtMs = toOptionalNumber(value.lastActivityAtMs); const updatedAtMs = toOptionalNumber(value.updatedAtMs); - const execution = normalizeExecution(conversationId, value.execution); + const execution = normalizeExecution( + conversationId, + value.execution, + schemaVersion, + ); const destination = value.destination === undefined ? undefined @@ -1517,7 +1490,7 @@ export async function drainConversationMailbox(args: { leaseToken: string; nowMs?: number; state?: StateAdapter; -}): Promise { +}): Promise<{ changed: boolean; messages: InboundMessage[] }> { const nowMs = args.nowMs ?? now(); const pending = await withConversationMutation(args, async (state) => { const current = await readConversation(state, args.conversationId); @@ -1529,7 +1502,7 @@ export async function drainConversationMailbox(args: { return pendingMessages(current); }); if (pending.length === 0) { - return []; + return { changed: false, messages: [] }; } const result = await args.handle(pending); @@ -1554,6 +1527,9 @@ export async function drainConversationMailbox(args: { ); } } + if (acknowledgedIds.size === 0 && deferredIds.size === 0) { + return { changed: false, messages: [] }; + } await withConversationMutation(args, async (state, lock) => { const current = await readConversation(state, args.conversationId); @@ -1587,9 +1563,12 @@ export async function drainConversationMailbox(args: { ), ); }); - return pending.filter((message) => - acknowledgedIds.has(message.inboundMessageId), - ); + return { + changed: true, + messages: pending.filter((message) => + acknowledgedIds.has(message.inboundMessageId), + ), + }; } /** Acknowledge leased mailbox entries after the handler accepts responsibility. */ diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 7f69b8c5f..35919e992 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -371,21 +371,11 @@ export async function drainConversationMailbox( state?: StateAdapter; }, ) { - let changed = false; - const result = await workState.drainConversationMailbox({ - ...args, - handle: async (messages) => { - const result = await args.handle(messages); - changed = result - ? result.ack.length > 0 || result.defer.length > 0 - : messages.length > 0; - return result; - }, - }); - if (changed) { + const result = await workState.drainConversationMailbox(args); + if (result.changed) { await recordExecutionMetadata(args); } - return result; + return result.messages; } /** Acknowledge leased mailbox entries after the handler accepts responsibility. */ diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 33d998eb7..9bc60862d 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -231,6 +231,60 @@ describe("conversation work execution", () => { ); }); + it("migrates schema-v1 mailbox messages without delivery as auto", async () => { + const state = getStateAdapter(); + await state.connect(); + await appendInboundMessage({ + message: inboundMessage("legacy-delivery"), + nowMs: 1_000, + state, + }); + const raw = (await state.get(CONVERSATION_WORK_STATE_KEY)) as { + execution: { pendingMessages: Array> }; + schemaVersion: number; + }; + raw.schemaVersion = 1; + delete raw.execution.pendingMessages[0]?.delivery; + await state.set(CONVERSATION_WORK_STATE_KEY, raw); + const observed: string[] = []; + + await expect( + processConversationWork(conversationQueueMessage(), { + queue: createConversationWorkQueueTestAdapter(), + run: async (context) => { + observed.push( + ...context.attempt.messages.map((message) => message.delivery), + ); + await context.attempt.drain(async () => {}); + return { status: "completed" }; + }, + state, + }), + ).resolves.toEqual({ status: "completed" }); + + expect(observed).toEqual(["auto"]); + }); + + it("rejects unknown mailbox delivery without dropping pending work", async () => { + const state = getStateAdapter(); + await state.connect(); + await appendInboundMessage({ + message: inboundMessage("unknown-delivery"), + nowMs: 1_000, + state, + }); + const raw = (await state.get(CONVERSATION_WORK_STATE_KEY)) as { + execution: { pendingMessages: Array> }; + }; + raw.execution.pendingMessages[0]!.delivery = "future-mode"; + await state.set(CONVERSATION_WORK_STATE_KEY, raw); + + await expect( + getConversationWorkState({ conversationId: CONVERSATION_ID, state }), + ).rejects.toThrow(`Conversation record is invalid for ${CONVERSATION_ID}`); + await expect(state.get(CONVERSATION_WORK_STATE_KEY)).resolves.toEqual(raw); + }); + it("repairs duplicate inbound work when no queue marker was recorded", async () => { const queue = createConversationWorkQueueTestAdapter(); await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); @@ -259,6 +313,7 @@ describe("conversation work execution", () => { await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); const twinWithFiles = inboundMessage("m1", { + delivery: "interrupt", receivedAtMs: 2_000, input: { text: "message m1", @@ -276,6 +331,7 @@ describe("conversation work execution", () => { }); expect(work?.messages).toEqual([ expect.objectContaining({ + delivery: "interrupt", inboundMessageId: "m1", receivedAtMs: 1_100, input: expect.objectContaining({ @@ -1185,6 +1241,62 @@ describe("conversation work execution", () => { ]); }); + it("does not rewrite mailbox state for an empty drain result", async () => { + const events: string[] = []; + const conversationStore = metadataEventsStore(events); + const state = getStateAdapter(); + await state.connect(); + await appendInboundMessage({ + conversationStore, + message: inboundMessage("m1"), + nowMs: 1_000, + state, + }); + const lease = await startConversationWork({ + conversationId: CONVERSATION_ID, + conversationStore, + nowMs: 2_000, + state, + }); + expect(lease.status).toBe("acquired"); + if (lease.status !== "acquired") { + throw new Error("Expected conversation work lease"); + } + await appendInboundMessage({ + conversationStore, + message: inboundMessage("m2", { + createdAtMs: 2_100, + delivery: "defer", + receivedAtMs: 2_100, + }), + nowMs: 2_100, + state, + }); + const before = await getConversationWorkState({ + conversationId: CONVERSATION_ID, + state, + }); + const metadataCount = events.length; + + await expect( + drainConversationMailbox({ + conversationId: CONVERSATION_ID, + conversationStore, + handle: async () => ({ ack: [], defer: [] }), + leaseToken: lease.leaseToken, + nowMs: 3_000, + state, + }), + ).resolves.toEqual([]); + + const after = await getConversationWorkState({ + conversationId: CONVERSATION_ID, + state, + }); + expect(after?.execution.updatedAtMs).toBe(before?.execution.updatedAtMs); + expect(events).toHaveLength(metadataCount); + }); + it("rejects mailbox acknowledgements outside the pending set", async () => { const queue = createConversationWorkQueueTestAdapter(); await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index cd4fd1de6..94921d084 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -906,8 +906,8 @@ describe("Slack conversation work execution", () => { await hooks.drainSteeringMessages?.(async (steering) => { injected.push(steering.map((candidate) => candidate.message.id)); return { - ack: steering.map((candidate) => candidate.inboundMessageId), - defer: [], + deferred: [], + handled: steering.map((candidate) => candidate.inboundMessageId), }; }); }, @@ -989,8 +989,8 @@ describe("Slack conversation work execution", () => { await hooks.drainSteeringMessages?.(async (steering) => { observed.push(steering.map((candidate) => candidate.message.id)); return { - ack: steering.map((candidate) => candidate.inboundMessageId), - defer: [], + deferred: [], + handled: steering.map((candidate) => candidate.inboundMessageId), }; }); }, @@ -1048,7 +1048,9 @@ describe("Slack conversation work execution", () => { services: ingressServices, }); - const observed: Array> = []; + const observed: Array< + Array<{ delivery: "auto" | "interrupt"; id: string }> + > = []; const runtime: SlackWorkerOptions["runtime"] = { handleNewMention: async (_thread, _message, hooks) => { await hooks.ack?.(); @@ -1098,13 +1100,13 @@ describe("Slack conversation work execution", () => { await hooks.drainSteeringMessages?.(async (steering) => { observed.push( steering.map((candidate) => ({ - activeRequest: candidate.activeRequest, + delivery: candidate.delivery, id: candidate.message.id, })), ); return { - ack: steering.map((candidate) => candidate.inboundMessageId), - defer: [], + deferred: [], + handled: steering.map((candidate) => candidate.inboundMessageId), }; }); }, @@ -1122,7 +1124,7 @@ describe("Slack conversation work execution", () => { }), ).resolves.toEqual({ status: "completed" }); - expect(observed).toEqual([[{ activeRequest: true, id: "1712345.1002" }]]); + expect(observed).toEqual([[{ delivery: "interrupt", id: "1712345.1002" }]]); }); it("does not replay injected Slack mailbox records after lease recovery", async () => { From df3835ddfbd53528345f3b9a0a71c45efe326ef7 Mon Sep 17 00:00:00 2001 From: David Cramer Date: Thu, 23 Jul 2026 13:36:09 -0700 Subject: [PATCH 08/10] fix(runtime): Enforce interrupt and defer delivery Assign interrupts only to Slack mentions, queue all other inbound work as deferred, and batch pending work by delivery priority. --- TERMINOLOGY.md | 8 +- .../junior/src/chat/runtime/slack-runtime.ts | 47 +++------- .../junior/src/chat/task-execution/README.md | 20 ++--- .../src/chat/task-execution/slack-work.ts | 47 ++-------- .../junior/src/chat/task-execution/state.ts | 51 ++++------- .../junior/src/chat/task-execution/store.ts | 1 - .../junior/src/chat/task-execution/worker.ts | 27 +++--- .../services/turn-session-record.test.ts | 2 +- .../task-execution/conversation-work.test.ts | 33 +++---- .../slack-conversation-work.test.ts | 74 ++++++++------- .../tests/component/vercel-queue-dev.test.ts | 2 +- .../tests/fixtures/conversation-work.ts | 2 +- ...onversation-turn-steering-behavior.test.ts | 89 ++++++++----------- 13 files changed, 162 insertions(+), 241 deletions(-) diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index bb950cac2..817bea68e 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -13,13 +13,13 @@ Canonical words used across Junior's code and documentation. - **Agent input**: the inbound content, context, and runtime metadata selected for a turn. - **Steering message**: a user message that interrupts the active turn at the - next safe boundary and starts a new turn. Messages arriving together may be - batched into that turn. + next safe boundary and joins that turn. Messages arriving together may be + batched into the same turn. - **Follow-up message**: a user message that waits for the active turn to finish before starting the next turn. - **Mailbox delivery**: the durable instruction for an inbound message: - `interrupt` is eligible at the next safe boundary, `defer` waits for the - active turn to finish, and `auto` asks runtime policy to choose between them. + `interrupt` is eligible at the next safe boundary, while `defer` follows + normal ordering and waits when a turn is already active. - **Turn**: one request-to-final-response cycle. It may span multiple runs and execution slices; one model invocation is not a turn. - **Run**: one bounded attempt to advance a turn. A later run may resume the diff --git a/packages/junior/src/chat/runtime/slack-runtime.ts b/packages/junior/src/chat/runtime/slack-runtime.ts index 278eb962f..521b3df27 100644 --- a/packages/junior/src/chat/runtime/slack-runtime.ts +++ b/packages/junior/src/chat/runtime/slack-runtime.ts @@ -56,26 +56,17 @@ export interface AssistantLifecycleEvent { userId?: string; } -type SteeringMode = "defer" | "interrupt"; - export interface SteeringCandidateMessage { - delivery: "auto" | "interrupt"; inboundMessageId: string; message: Message; } -/** Runtime-owned handling result for drained Slack steering candidates. */ -export interface SteeringDrainResult { - deferred: readonly string[]; - handled: readonly string[]; -} - export interface ReplyHooks { beforeFirstResponsePost?: () => Promise; drainSteeringMessages?: ( accept: ( messages: SteeringCandidateMessage[], - ) => Promise, + ) => Promise, ) => Promise; messageContext?: MessageContext; ack?: () => Promise; @@ -259,7 +250,6 @@ interface SteeringMessageDecision { context: TurnContext; decision: SubscribedReplyDecision; inboundMessageId: string; - mode: SteeringMode; message: Message; text: TurnMessageText; } @@ -268,12 +258,11 @@ interface SteeringMessageSelection { accepted: Array<{ inboundMessageId: string; message: Message; - mode: SteeringMode; }>; skipped: SteeringMessageDecision[]; } -/** Drain mailbox steering messages after classifying interrupt, defer, and skip. */ +/** Drain explicit mailbox interruptions after applying subscribed reply policy. */ function createAcceptedSteeringDrain( hooks: ReplyHooks, options: { @@ -301,25 +290,16 @@ function createAcceptedSteeringDrain( await hooks.drainSteeringMessages!(async (messages) => { const selection = await options.selectMessages(messages, context); await options.onSkipped?.(selection.skipped); - // Deferred messages stay pending so a later worker slice handles them - // after the active answer is delivered. - const interrupted = selection.accepted - .filter((accepted) => accepted.mode === "interrupt") - .map((accepted) => accepted.message); + const interrupted = selection.accepted.map( + (accepted) => accepted.message, + ); await accept(getQueuedMessagesFromSlackMessages(interrupted, options)); interruptedMessages = interrupted; await options.onAcceptedForProcessing?.(interrupted); - return { - deferred: selection.accepted - .filter((accepted) => accepted.mode === "defer") - .map((accepted) => accepted.inboundMessageId), - handled: [ - ...selection.accepted - .filter((accepted) => accepted.mode === "interrupt") - .map((accepted) => accepted.inboundMessageId), - ...selection.skipped.map((skipped) => skipped.inboundMessageId), - ], - }; + return [ + ...selection.accepted.map((accepted) => accepted.inboundMessageId), + ...selection.skipped.map((skipped) => skipped.inboundMessageId), + ]; }); return getQueuedMessagesFromSlackMessages( interruptedMessages ?? [], @@ -534,7 +514,6 @@ export function createSlackTurnRuntime< ): Promise<{ context: TurnContext; decision: SubscribedReplyDecision; - mode: SteeringMode; text: TurnMessageText; }> => { const { message } = candidate; @@ -552,22 +531,18 @@ export function createSlackTurnRuntime< rawText: appendSlackLegacyAttachmentText(message.text, message.raw), userText: appendSlackLegacyAttachmentText(strippedUserText, message.raw), }; - const isActiveRequest = - candidate.delivery === "interrupt" || Boolean(message.isMention); - const decision = await deps.decideSubscribedReply({ rawText: text.rawText, text: text.userText, conversationContext, hasAttachments: message.attachments.length > 0 || legacyAttachmentText !== "", - isExplicitMention: isActiveRequest, + isExplicitMention: true, context, }); return { context, decision, - mode: isActiveRequest ? "interrupt" : "defer", text, }; }; @@ -644,7 +619,6 @@ export function createSlackTurnRuntime< context: decision.context, decision: decision.decision, inboundMessageId: candidate.inboundMessageId, - mode: decision.mode, message: candidate.message, text: decision.text, }); @@ -671,7 +645,6 @@ export function createSlackTurnRuntime< .map((message) => ({ inboundMessageId: message.inboundMessageId, message: message.message, - mode: message.mode, })), skipped: selected.filter((message) => !message.decision.shouldReply), }; diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index 01ebcae31..945018500 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -7,7 +7,7 @@ source of truth. ## State Model - A conversation mailbox contains normalized pending work with a durable - delivery mode: `interrupt`, `defer`, or `auto` when runtime policy must decide. + delivery mode: `interrupt` or `defer`. - A queue payload identifies the conversation to wake; persisted conversation work owns destination and delivery. - A lease grants one worker temporary execution ownership. @@ -15,8 +15,8 @@ source of truth. slow work from abandoned work. - Delivery state prevents a completed turn from being posted twice. -Schema-v1 mailbox entries without `delivery` migrate to `auto`. Schema-v2 -entries require a valid delivery mode and reject invalid pending work. +Schema-v1 mailbox entries migrate to deferred delivery. Schema-v2 entries +require a valid delivery value and reject invalid pending work. `state.ts`, `store.ts`, and their runtime schemas define the persisted shapes. @@ -34,15 +34,15 @@ entries require a valid delivery mode and reject invalid pending work. 6. Terminal delivery or intentional no-reply completion records the delivered turn before acknowledging work. -New messages that arrive during a run remain durable. Explicit `interrupt` work -is eligible at the next safe boundary, explicit `defer` work waits for the next -turn, and `auto` work is classified by runtime policy. +New messages that arrive during a run remain durable. `interrupt` work is +eligible at the next safe boundary, while `defer` work follows normal ordering +and waits for the next turn. -Slack mentions and assistant-thread user messages use `interrupt`, subscribed -resource events use `defer`, and other subscribed messages use `auto`. +Slack `@` mentions use `interrupt` delivery; all other inbound messages use +`defer`. -One lease-bound drain removes disjoint `ack` ids and retains disjoint `defer` -ids with `delivery: "defer"`. +An active turn drains only `interrupt` messages. When a new turn starts, +interrupt messages are handled before queued deferred work. ## Queue And Lease Rules diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index 244cbaa78..07641e629 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -11,7 +11,6 @@ import { z } from "zod"; import type { SlackTurnOptions, SteeringCandidateMessage, - SteeringDrainResult, } from "@/chat/runtime/slack-runtime"; import { isCooperativeTurnYieldError, @@ -629,18 +628,6 @@ function routeForRecords(records: InboundMessage[]): SlackConversationRoute { : "subscribed"; } -function isSlackAssistantThreadUserMessage(message: Message): boolean { - const raw = - message.raw && typeof message.raw === "object" - ? (message.raw as Record) - : undefined; - return ( - raw?.channel_type === "im" && - typeof raw.thread_ts === "string" && - raw.thread_ts.trim().length > 0 - ); -} - function isResourceEventNotificationMessage(message: Message): boolean { const raw = message.raw && typeof message.raw === "object" @@ -839,32 +826,20 @@ export function createSlackConversationWorker( ); } }; - // Explicitly deferred records never enter steering. Auto-delivered - // records are offered to runtime policy, while interrupts bypass it. + // Only explicit interruptions enter the active turn. Deferred work + // stays pending for a normal turn. const drainSteeringMessages = async ( accept: ( messages: SteeringCandidateMessage[], - ) => Promise, + ) => Promise, ): Promise => { await context.attempt.drain(async (pendingRecords) => { - const result = await accept( - pendingRecords.map((record) => { - if (record.delivery === "defer") { - throw new Error( - "Deferred mailbox message reached Slack steering", - ); - } - return { - delivery: record.delivery, - inboundMessageId: record.inboundMessageId, - message: restoreMessage({ adapter, record }), - }; - }), + return await accept( + pendingRecords.map((record) => ({ + inboundMessageId: record.inboundMessageId, + message: restoreMessage({ adapter, record }), + })), ); - return { - ack: result.handled, - defer: result.deferred, - }; }); }; @@ -941,11 +916,7 @@ export function buildSlackInboundMessage(args: { args.conversationId, args.message.id, ].join(":"), - delivery: - args.route === "mention" || - isSlackAssistantThreadUserMessage(args.message) - ? "interrupt" - : "auto", + delivery: args.route === "mention" ? "interrupt" : "defer", source: "slack", createdAtMs: args.message.metadata.dateSent.getTime(), receivedAtMs: args.receivedAtMs, diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index ff4489165..5322f562e 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -95,22 +95,12 @@ const agentInputSchema = z export type AgentInput = z.output; /** Durable delivery modes for pending inbound mailbox work. */ -export const inboundMessageDeliverySchema = z.enum([ - "auto", - "defer", - "interrupt", -]); +export const inboundMessageDeliverySchema = z.enum(["defer", "interrupt"]); export type InboundMessageDelivery = z.output< typeof inboundMessageDeliverySchema >; -/** Mailbox ids to remove or retain as deferred in one lease-bound drain. */ -export interface MailboxDrainResult { - ack: readonly string[]; - defer: readonly string[]; -} - /** Canonical durable mailbox entry owned by task execution. */ export const inboundMessageSchema = z .object({ @@ -275,7 +265,7 @@ function upgradedPendingMessage( ? duplicate.input : stored.input; const delivery = - stored.delivery === "auto" && duplicate.delivery === "interrupt" + stored.delivery === "defer" && duplicate.delivery === "interrupt" ? "interrupt" : stored.delivery; if (input === stored.input && delivery === stored.delivery) { @@ -330,10 +320,13 @@ function normalizeMessage( value: unknown, schemaVersion: 1 | 2, ): InboundMessage | undefined { - // TODO(v0.110.0): Remove schema-v1 mailbox delivery migration after old records expire. + // TODO(v0.110.0): Remove schema-v1 mailbox migration after old records expire. const candidate = schemaVersion === 1 && isRecord(value) && value.delivery === undefined - ? { ...value, delivery: "auto" } + ? { + ...value, + delivery: "defer", + } : value; const parsed = inboundMessageSchema.safeParse(candidate); return parsed.success ? parsed.data : undefined; @@ -1481,12 +1474,12 @@ export async function checkInConversationWork(args: { /** * Resolve pending mailbox entries after the caller accepts responsibility. * - * Acknowledged entries are removed while deferred entries remain for the next - * turn. Returning nothing acknowledges every entry passed to the handler. + * Returning ids acknowledges only that subset; returning nothing acknowledges + * every pending entry passed to the handler. */ export async function drainConversationMailbox(args: { conversationId: string; - handle: (messages: InboundMessage[]) => Promise; + handle: (messages: InboundMessage[]) => Promise; leaseToken: string; nowMs?: number; state?: StateAdapter; @@ -1510,24 +1503,16 @@ export async function drainConversationMailbox(args: { pending.map((message) => message.inboundMessageId), ); const acknowledgedIds = new Set( - result?.ack ?? pending.map((message) => message.inboundMessageId), + result ?? pending.map((message) => message.inboundMessageId), ); - const deferredIds = new Set(result?.defer ?? []); - for (const inboundMessageId of [...acknowledgedIds, ...deferredIds]) { + for (const inboundMessageId of acknowledgedIds) { if (!pendingIds.has(inboundMessageId)) { throw new Error( `Conversation mailbox drain result is not pending for ${args.conversationId}`, ); } } - for (const inboundMessageId of deferredIds) { - if (acknowledgedIds.has(inboundMessageId)) { - throw new Error( - `Conversation mailbox drain result overlaps for ${args.conversationId}`, - ); - } - } - if (acknowledgedIds.size === 0 && deferredIds.size === 0) { + if (acknowledgedIds.size === 0) { return { changed: false, messages: [] }; } @@ -1538,13 +1523,9 @@ export async function drainConversationMailbox(args: { `Conversation lease is not held for ${args.conversationId}`, ); } - const pendingMessages = current.execution.pendingMessages - .filter((message) => !acknowledgedIds.has(message.inboundMessageId)) - .map((message) => - deferredIds.has(message.inboundMessageId) - ? { ...message, delivery: "defer" as const } - : message, - ); + const pendingMessages = current.execution.pendingMessages.filter( + (message) => !acknowledgedIds.has(message.inboundMessageId), + ); await writeConversation( state, lock, diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 35919e992..09a92ae9e 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -24,7 +24,6 @@ export { type InboundMessage, type InboundMessageDelivery, type Lease, - type MailboxDrainResult, type RequestConversationWorkResult, type Source, type StartConversationWorkAcquired, diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 60a457a3b..e4bf518bd 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -28,7 +28,6 @@ import { type AttemptFailure, type ConversationWorkState, type InboundMessage, - type MailboxDrainResult, } from "./store"; export const CONVERSATION_WORK_DEFER_DELAY_MS = 15_000; @@ -47,7 +46,7 @@ export interface InboxAttempt { conversationId: string; destination: Destination; drain( - handle: (messages: InboundMessage[]) => Promise, + handle: (messages: InboundMessage[]) => Promise, ): Promise; isFinalAttempt: boolean; messages: InboundMessage[]; @@ -82,6 +81,14 @@ function now(options: ProcessConversationWorkOptions): number { return options.nowMs?.() ?? Date.now(); } +/** Prioritize the interrupt batch; otherwise process the deferred batch. */ +function selectAttemptMessages(messages: InboundMessage[]): InboundMessage[] { + const interrupts = messages.filter( + (message) => message.delivery === "interrupt", + ); + return interrupts.length > 0 ? interrupts : messages; +} + function nudgeIdempotencyKey( reason: string, conversationId: string, @@ -315,7 +322,9 @@ export async function processConversationWork( conversationId, state: options.state, }); - const attemptMessages = leasedWork?.messages ?? initial.messages; + const attemptMessages = selectAttemptMessages( + leasedWork?.messages ?? initial.messages, + ); const attemptMessageIds = attemptMessages.map( (message) => message.inboundMessageId, ); @@ -340,7 +349,7 @@ export async function processConversationWork( ); const drain = ( - handle: (messages: InboundMessage[]) => Promise, + handle: (messages: InboundMessage[]) => Promise, ) => drainConversationMailbox({ conversationId, @@ -348,16 +357,14 @@ export async function processConversationWork( conversationStore: options.conversationStore, handle: async (messages) => { const candidates = messages.filter( - (message) => message.delivery !== "defer", + (message) => message.delivery === "interrupt", ); if (candidates.length === 0) { - return { ack: [], defer: [] }; + return []; } return ( - (await handle(candidates)) ?? { - ack: candidates.map((message) => message.inboundMessageId), - defer: [], - } + (await handle(candidates)) ?? + candidates.map((message) => message.inboundMessageId) ); }, nowMs: now(options), diff --git a/packages/junior/tests/component/services/turn-session-record.test.ts b/packages/junior/tests/component/services/turn-session-record.test.ts index 69cbb0eaf..db0cb0b5d 100644 --- a/packages/junior/tests/component/services/turn-session-record.test.ts +++ b/packages/junior/tests/component/services/turn-session-record.test.ts @@ -131,7 +131,7 @@ describe("persistAuthPauseSessionRecord", () => { text: "start", }, receivedAtMs: 9_000, - delivery: "auto", + delivery: "defer", source: "slack", }, nowMs: 9_000, diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 9bc60862d..2c1a66b1d 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -231,7 +231,7 @@ describe("conversation work execution", () => { ); }); - it("migrates schema-v1 mailbox messages without delivery as auto", async () => { + it("migrates schema-v1 Slack mailbox messages for classification", async () => { const state = getStateAdapter(); await state.connect(); await appendInboundMessage({ @@ -255,14 +255,14 @@ describe("conversation work execution", () => { observed.push( ...context.attempt.messages.map((message) => message.delivery), ); - await context.attempt.drain(async () => {}); + await context.attempt.ack(); return { status: "completed" }; }, state, }), ).resolves.toEqual({ status: "completed" }); - expect(observed).toEqual(["auto"]); + expect(observed).toEqual(["defer"]); }); it("rejects unknown mailbox delivery without dropping pending work", async () => { @@ -287,7 +287,10 @@ describe("conversation work execution", () => { it("repairs duplicate inbound work when no queue marker was recorded", async () => { const queue = createConversationWorkQueueTestAdapter(); - await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); + await appendInboundMessage({ + message: inboundMessage("m1", { delivery: "defer" }), + nowMs: 1_000, + }); await expect( appendAndEnqueueInboundMessage({ @@ -1181,10 +1184,7 @@ describe("conversation work execution", () => { run: async (context) => { await context.attempt.drain(async (messages) => { injected.push(messages.map((message) => message.inboundMessageId)); - return { - ack: ["m1"], - defer: [], - }; + return ["m1"]; }); return { status: "completed" }; }, @@ -1201,12 +1201,13 @@ describe("conversation work execution", () => { ]); }); - it("keeps deferred messages separate from interruptions", async () => { + it("drains interrupts while deferred messages stay queued", async () => { const queue = createConversationWorkQueueTestAdapter(); await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); await appendInboundMessage({ message: inboundMessage("m2", { createdAtMs: 2_000, + delivery: "defer", receivedAtMs: 2_100, }), nowMs: 2_100, @@ -1219,17 +1220,14 @@ describe("conversation work execution", () => { run: async (context) => { await context.attempt.drain(async (messages) => { drained.push(messages.map((message) => message.inboundMessageId)); - return { - ack: ["m1"], - defer: ["m2"], - }; + return messages.map((message) => message.inboundMessageId); }); return { status: "completed" }; }, }), ).resolves.toEqual({ status: "pending_requeued" }); - expect(drained).toEqual([["m1", "m2"]]); + expect(drained).toEqual([["m1"]]); const state = await getConversationWorkState({ conversationId: CONVERSATION_ID, }); @@ -1282,7 +1280,7 @@ describe("conversation work execution", () => { drainConversationMailbox({ conversationId: CONVERSATION_ID, conversationStore, - handle: async () => ({ ack: [], defer: [] }), + handle: async () => [], leaseToken: lease.leaseToken, nowMs: 3_000, state, @@ -1307,10 +1305,7 @@ describe("conversation work execution", () => { queue, run: async (context) => { try { - await context.attempt.drain(async () => ({ - ack: ["different-message"], - defer: [], - })); + await context.attempt.drain(async () => ["different-message"]); } catch (error) { drainError = error; throw error; diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index 94921d084..e243d1df0 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -271,7 +271,7 @@ describe("Slack conversation work execution", () => { ...conversationQueueMessage(), destination: SLACK_DESTINATION, inboundMessageId: "malformed-slack-metadata", - delivery: "auto" as const, + delivery: "defer" as const, source: "slack" as const, createdAtMs: 1_000, receivedAtMs: 1_100, @@ -673,7 +673,7 @@ describe("Slack conversation work execution", () => { }); }); - it("keeps restored thread context aligned with promoted mention routing", async () => { + it("processes an interrupt before a queued deferred follow-up", async () => { const queue = createConversationWorkQueueTestAdapter(); const state = getStateAdapter(); await state.connect(); @@ -734,8 +734,14 @@ describe("Slack conversation work execution", () => { skipped: hooks.messageContext?.skipped ?? [], }); }, - handleSubscribedMessage: async () => { - throw new Error("mixed mention batches should promote to mention"); + handleSubscribedMessage: async (thread, message, hooks) => { + await hooks.ack?.(); + subscribedValues.push(await thread.isSubscribed()); + calls.push({ + thread, + message, + skipped: hooks.messageContext?.skipped ?? [], + }); }, }; await expect( @@ -745,14 +751,25 @@ describe("Slack conversation work execution", () => { runtime, state, }), - ).resolves.toEqual({ status: "completed" }); + ).resolves.toEqual({ status: "pending_requeued" }); expect(calls).toHaveLength(1); - expect(calls[0]?.message.id).toBe("1712345.0002"); - expect(calls[0]?.skipped.map((message) => message.id)).toEqual([ - "1712345.0001", - ]); + expect(calls[0]?.message.id).toBe("1712345.0001"); + expect(calls[0]?.skipped).toEqual([]); expect(subscribedValues).toEqual([false]); + + await expect( + processNextQueuedSlackWork({ + getSlackAdapter: () => slackAdapter, + queue, + runtime, + state, + }), + ).resolves.toEqual({ status: "completed" }); + expect(calls).toHaveLength(2); + expect(calls[1]?.message.id).toBe("1712345.0002"); + expect(calls[1]?.skipped).toEqual([]); + expect(subscribedValues).toEqual([false, true]); await expectRemainingQueuedSlackWorkIsNoop({ getSlackAdapter: () => slackAdapter, queue, @@ -905,10 +922,7 @@ describe("Slack conversation work execution", () => { }); await hooks.drainSteeringMessages?.(async (steering) => { injected.push(steering.map((candidate) => candidate.message.id)); - return { - deferred: [], - handled: steering.map((candidate) => candidate.inboundMessageId), - }; + return steering.map((candidate) => candidate.inboundMessageId); }); }, handleSubscribedMessage: async () => { @@ -988,10 +1002,7 @@ describe("Slack conversation work execution", () => { }); await hooks.drainSteeringMessages?.(async (steering) => { observed.push(steering.map((candidate) => candidate.message.id)); - return { - deferred: [], - handled: steering.map((candidate) => candidate.inboundMessageId), - }; + return steering.map((candidate) => candidate.inboundMessageId); }); }, handleSubscribedMessage: async () => { @@ -1021,7 +1032,7 @@ describe("Slack conversation work execution", () => { ]); }); - it("treats Slack assistant-thread user messages as active steering", async () => { + it("defers Slack assistant-thread messages without a mention", async () => { const queue = createConversationWorkQueueTestAdapter(); const state = getStateAdapter(); await state.connect(); @@ -1048,9 +1059,7 @@ describe("Slack conversation work execution", () => { services: ingressServices, }); - const observed: Array< - Array<{ delivery: "auto" | "interrupt"; id: string }> - > = []; + const observed: string[][] = []; const runtime: SlackWorkerOptions["runtime"] = { handleNewMention: async (_thread, _message, hooks) => { await hooks.ack?.(); @@ -1098,16 +1107,8 @@ describe("Slack conversation work execution", () => { state, }); await hooks.drainSteeringMessages?.(async (steering) => { - observed.push( - steering.map((candidate) => ({ - delivery: candidate.delivery, - id: candidate.message.id, - })), - ); - return { - deferred: [], - handled: steering.map((candidate) => candidate.inboundMessageId), - }; + observed.push(steering.map((candidate) => candidate.message.id)); + return steering.map((candidate) => candidate.inboundMessageId); }); }, handleSubscribedMessage: async () => { @@ -1122,9 +1123,16 @@ describe("Slack conversation work execution", () => { runtime, state, }), - ).resolves.toEqual({ status: "completed" }); + ).resolves.toEqual({ status: "pending_requeued" }); - expect(observed).toEqual([[{ delivery: "interrupt", id: "1712345.1002" }]]); + expect(observed).toEqual([]); + const work = await getConversationWorkState({ conversationId, state }); + expect(work?.execution.pendingMessages).toEqual([ + expect.objectContaining({ + delivery: "defer", + inboundMessageId: expect.stringContaining("1712345.1002"), + }), + ]); }); it("does not replay injected Slack mailbox records after lease recovery", async () => { diff --git a/packages/junior/tests/component/vercel-queue-dev.test.ts b/packages/junior/tests/component/vercel-queue-dev.test.ts index b5fc44d9f..1beddfd77 100644 --- a/packages/junior/tests/component/vercel-queue-dev.test.ts +++ b/packages/junior/tests/component/vercel-queue-dev.test.ts @@ -283,7 +283,7 @@ describe("registerVercelConversationWorkDevConsumer", () => { message: { conversationId: "slack:C123:1712345.0001", inboundMessageId: "m1", - delivery: "auto", + delivery: "defer", destination: { channelId: "C123", platform: "slack", diff --git a/packages/junior/tests/fixtures/conversation-work.ts b/packages/junior/tests/fixtures/conversation-work.ts index cbdfc3538..e8201d7d5 100644 --- a/packages/junior/tests/fixtures/conversation-work.ts +++ b/packages/junior/tests/fixtures/conversation-work.ts @@ -197,7 +197,7 @@ export function inboundMessage( conversationId: CONVERSATION_ID, inboundMessageId, destination: SLACK_DESTINATION, - delivery: "auto", + delivery: "interrupt", source: "slack", createdAtMs: 1_000, receivedAtMs: 1_100, diff --git a/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts b/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts index 6d2bc35e5..15d52fdf6 100644 --- a/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts +++ b/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts @@ -241,7 +241,7 @@ describe("Slack behavior: durable turn steering", () => { expect(work ? countPendingConversationMessages(work) : 0).toBe(1); }); - it("interrupts explicit follow-ups and defers passive follow-ups during an active turn", async () => { + it("interrupts explicit follow-ups and queues deferred follow-ups", async () => { const agentEntered = deferred(); const releaseAgent = deferred(); let blockingCallReleased = false; @@ -428,29 +428,18 @@ describe("Slack behavior: durable turn steering", () => { expect(work?.execution.inboundMessageIds).toHaveLength(5); expect(work ? countPendingConversationMessages(work) : 0).toBe(0); expect(work?.needsRun).toBe(false); - const persistedState = await getPersistedThreadState(conversationId); - const conversation = coerceThreadConversationState(persistedState); - await hydrateConversationMessages({ conversation, conversationId }); - expect(conversation.messages).toEqual( - expect.arrayContaining([ - expect.objectContaining({ - text: "thanks folks", - meta: expect.objectContaining({ - replied: false, - skippedReason: "side_conversation:passive side conversation", - }), - }), - ]), - ); - const expectedReactionTargets = (name: string) => - [THREAD_TS, "1712345.000200", "1712345.000300", "1712345.000400"].map( - (timestamp) => ({ - channel: CHANNEL_ID, - name, - timestamp, - }), - ); + [ + THREAD_TS, + "1712345.000200", + "1712345.000250", + "1712345.000300", + "1712345.000400", + ].map((timestamp) => ({ + channel: CHANNEL_ID, + name, + timestamp, + })); const expectedProcessingReactions = expectedReactionTargets("eyes"); const expectedCompletedReactions = expectedReactionTargets("white_check_mark"); @@ -537,7 +526,7 @@ describe("Slack behavior: durable turn steering", () => { expect(replyCalls).toEqual(["start the incident summary"]); }); - it("applies opt-out decisions from drained steering messages without reacting to them", async () => { + it("applies queued opt-out decisions on the next turn", async () => { const agentEntered = deferred(); const releaseAgent = deferred(); const drainedTexts: string[] = []; @@ -559,25 +548,26 @@ describe("Slack behavior: durable turn steering", () => { diagnostics: makeDiagnostics(), }); }; - const { conversationId, runNextQueuedWork, services } = createTurnHarness({ - completeObject: completeObjectWithDecision((prompt) => - prompt.includes("stop watching") - ? { - should_reply: false, - should_unsubscribe: true, - confidence: 1, - reason: "explicit stop instruction", - } - : { - should_reply: true, - should_unsubscribe: false, - confidence: 1, - reason: "active steering follow-up", - }, - ), - agentRunner: { run: executeAgentRun }, - state, - }); + const { conversationId, queue, runNextQueuedWork, services } = + createTurnHarness({ + completeObject: completeObjectWithDecision((prompt) => + prompt.includes("stop watching") + ? { + should_reply: false, + should_unsubscribe: true, + confidence: 1, + reason: "explicit stop instruction", + } + : { + should_reply: true, + should_unsubscribe: false, + confidence: 1, + reason: "active steering follow-up", + }, + ), + agentRunner: { run: executeAgentRun }, + state, + }); await createResourceEventSubscription( { conversationId, @@ -636,6 +626,10 @@ describe("Slack behavior: durable turn steering", () => { releaseAgent.resolve(); await expect(activeTurn).resolves.toEqual({ status: "completed" }); + expect(await state.isSubscribed(conversationId)).toBe(true); + while (queue.hasQueuedMessages()) { + await runNextQueuedWork(); + } expect(await state.isSubscribed(conversationId)).toBe(false); await expect( listResourceEventSubscriptions({ conversationId, state }), @@ -661,18 +655,11 @@ describe("Slack behavior: durable turn steering", () => { await hydrateConversationMessages({ conversation, conversationId }); expect(conversation.messages).toEqual( expect.arrayContaining([ - expect.objectContaining({ - text: "stop watching this thread", - meta: expect.objectContaining({ - replied: false, - skippedReason: "thread_opt_out:explicit stop instruction", - }), - }), expect.objectContaining({ text: "also add the rollout timeline", meta: expect.objectContaining({ replied: false, - skippedReason: "thread_opt_out:batch opt-out", + skippedReason: "thread_opt_out:explicit stop instruction", }), }), ]), From ff86163a75ef4a73d4e3b1dbd96631a69360cb8c Mon Sep 17 00:00:00 2001 From: David Cramer Date: Thu, 23 Jul 2026 13:56:01 -0700 Subject: [PATCH 09/10] fix(runtime): Preserve mention routing on duplicate upgrade When a duplicate Slack mention promotes deferred work to an interrupt, retain the mention metadata while preserving attachments already captured from its twin event. --- .../junior/src/chat/task-execution/state.ts | 18 ++++++--- .../task-execution/conversation-work.test.ts | 39 +++++++++++++++++++ 2 files changed, 51 insertions(+), 6 deletions(-) diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 5322f562e..7e7f81656 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -260,14 +260,20 @@ function upgradedPendingMessage( stored: InboundMessage, duplicate: InboundMessage, ): InboundMessage { - const input = - inputHasAttachments(duplicate.input) && !inputHasAttachments(stored.input) + const promotesInterrupt = + stored.delivery === "defer" && duplicate.delivery === "interrupt"; + const delivery = promotesInterrupt ? "interrupt" : stored.delivery; + const input = promotesInterrupt + ? { + ...duplicate.input, + ...(inputHasAttachments(duplicate.input) || + !inputHasAttachments(stored.input) + ? {} + : { attachments: stored.input.attachments }), + } + : inputHasAttachments(duplicate.input) && !inputHasAttachments(stored.input) ? duplicate.input : stored.input; - const delivery = - stored.delivery === "defer" && duplicate.delivery === "interrupt" - ? "interrupt" - : stored.delivery; if (input === stored.input && delivery === stored.delivery) { return stored; } diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 2c1a66b1d..b56a9293c 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -354,6 +354,45 @@ describe("conversation work execution", () => { expect(work?.messages[0]?.input.attachments).toEqual([{ id: "F123" }]); }); + it("promotes duplicate mention metadata without dropping attachments", async () => { + await appendInboundMessage({ + message: inboundMessage("m1", { + delivery: "defer", + input: { + attachments: [{ id: "F123" }], + authorId: "U123", + metadata: { platform: "slack", route: "subscribed" }, + text: "message m1", + }, + }), + nowMs: 1_000, + }); + await expect( + appendInboundMessage({ + message: inboundMessage("m1", { + delivery: "interrupt", + input: { + authorId: "U123", + metadata: { platform: "slack", route: "mention" }, + text: "message m1", + }, + }), + nowMs: 2_000, + }), + ).resolves.toEqual({ status: "duplicate" }); + + const work = await getConversationWorkState({ + conversationId: CONVERSATION_ID, + }); + expect(work?.messages[0]).toMatchObject({ + delivery: "interrupt", + input: { + attachments: [{ id: "F123" }], + metadata: { platform: "slack", route: "mention" }, + }, + }); + }); + it("retries transient conversation work index lock contention", async () => { const queue = createConversationWorkQueueTestAdapter(); const state = delayIndexLockOnce(getStateAdapter()); From d2e4ecd83d29aa3a7e40e24f4d86ea25ddfc89c5 Mon Sep 17 00:00:00 2001 From: David Cramer Date: Thu, 23 Jul 2026 15:31:10 -0700 Subject: [PATCH 10/10] fix(runtime): Preserve Slack delivery and legacy mailboxes Derive interrupts from explicit Slack mentions instead of broad DM routing, and ignore legacy injected markers without rejecting otherwise valid pending work. --- .../src/chat/task-execution/slack-work.ts | 2 +- .../junior/src/chat/task-execution/state.ts | 9 +++--- .../task-execution/conversation-work.test.ts | 29 +++++++++++++++++++ .../slack-conversation-work.test.ts | 3 +- 4 files changed, 36 insertions(+), 7 deletions(-) diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index 07641e629..c4b54cb1a 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -916,7 +916,7 @@ export function buildSlackInboundMessage(args: { args.conversationId, args.message.id, ].join(":"), - delivery: args.route === "mention" ? "interrupt" : "defer", + delivery: args.message.isMention ? "interrupt" : "defer", source: "slack", createdAtMs: args.message.metadata.dateSent.getTime(), receivedAtMs: args.receivedAtMs, diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 7e7f81656..caeafbf5b 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -397,13 +397,12 @@ function normalizeExecution( if (Array.isArray(value.pendingMessages)) { for (const rawMessage of value.pendingMessages) { const message = normalizeMessage(rawMessage, schemaVersion); - if ( - !message || - message.conversationId !== conversationId || - message.injectedAtMs !== undefined - ) { + if (!message || message.conversationId !== conversationId) { return undefined; } + if (message.injectedAtMs !== undefined) { + continue; + } pendingMessages.push(message); } pendingMessages.sort(compareMessages); diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index b56a9293c..1f78fca65 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -285,6 +285,35 @@ describe("conversation work execution", () => { await expect(state.get(CONVERSATION_WORK_STATE_KEY)).resolves.toEqual(raw); }); + it("ignores legacy injected markers without blocking pending work", async () => { + const state = getStateAdapter(); + await state.connect(); + await appendInboundMessage({ + message: inboundMessage("injected"), + nowMs: 1_000, + state, + }); + await appendInboundMessage({ + message: inboundMessage("pending", { + createdAtMs: 2_000, + receivedAtMs: 2_100, + }), + nowMs: 2_100, + state, + }); + const raw = (await state.get(CONVERSATION_WORK_STATE_KEY)) as { + execution: { pendingMessages: Array> }; + }; + raw.execution.pendingMessages[0]!.injectedAtMs = 1_500; + await state.set(CONVERSATION_WORK_STATE_KEY, raw); + + await expect( + getConversationWorkState({ conversationId: CONVERSATION_ID, state }), + ).resolves.toMatchObject({ + messages: [expect.objectContaining({ inboundMessageId: "pending" })], + }); + }); + it("repairs duplicate inbound work when no queue marker was recorded", async () => { const queue = createConversationWorkQueueTestAdapter(); await appendInboundMessage({ diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index e243d1df0..d9591e18e 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -212,6 +212,7 @@ describe("Slack conversation work execution", () => { isMe: false, }, }); + message.isMention = true; const thread = new ThreadImpl({ adapter: slackAdapter, stateAdapter: state, @@ -1101,7 +1102,7 @@ describe("Slack conversation work execution", () => { installation: { teamId: "T123" }, message: followUp, receivedAtMs: 1_100, - route: "subscribed", + route: "mention", thread, }), state,