diff --git a/packages/junior/src/chat/task-execution/conversation-turn.ts b/packages/junior/src/chat/task-execution/conversation-turn.ts index 1a67c9b4d6..a4ad2cd8f8 100644 --- a/packages/junior/src/chat/task-execution/conversation-turn.ts +++ b/packages/junior/src/chat/task-execution/conversation-turn.ts @@ -63,7 +63,6 @@ import { buildDeterministicTurnId } from "@/chat/state/turn-id"; import { abandonTurnRecord, getTurnRecord, - listTurnSummaries, saveTurnCheckpoint, } from "@/chat/task-execution/checkpoint"; import type { @@ -77,6 +76,7 @@ import { import type { SandboxRef } from "@/chat/sandbox/ref"; import { getEventBatchDelayMs, + listActiveMailboxTurns, sourceFromTurnInput, turnInputFactsFromConversationMessage, type MailboxTurnWork, @@ -516,28 +516,23 @@ export function createConversationTurnWorker( return await completeCancelledTurn(); } if (!isResume) { - // Match Slack: new input supersedes an auth-paused Turn - // instead of leaving two active Turns or letting a late OAuth - // wake resume stale work. + // Match Slack: new input supersedes every other open mailbox + // Turn. New input starts only when no yielded Turn waits, so an + // open Turn here waits for OAuth or was left open by a failed or + // stopped Run. A second open Turn makes every later resume fail. const pendingAuthorizationActorId = conversation.processing.pendingAuth?.actorId; - const authParked = ( - await listTurnSummaries(context.conversationId) - ).filter( - (summary) => - summary.surface === "api" && - !summary.dispatchId && - summary.state === "paused" && - summary.resumeReason === "auth", - ); - for (const parked of authParked) { + const superseded = ( + await listActiveMailboxTurns(context.conversationId) + ).filter((record) => record.turnId !== turnId); + for (const stale of superseded) { if (stopSignal?.aborted) { return await completeCancelledTurn(); } await abandonTurnRecord({ conversationId: context.conversationId, - turnId: parked.turnId, - errorMessage: "Auth-paused Turn superseded by new input", + turnId: stale.turnId, + errorMessage: "Turn superseded by new input", }); // Keep pendingAuth: MCP OAuth still needs it to accept an in-flight // connect and store credentials. The abandoned turn record makes a @@ -545,10 +540,14 @@ export function createConversationTurnWorker( markTurnClosed({ conversation, nowMs: Date.now(), - sessionId: parked.turnId, + sessionId: stale.turnId, }); } - if (authParked.length > 0 && pendingAuthorizationActorId) { + const authParked = superseded.some( + (record) => + record.state === "paused" && record.resumeReason === "auth", + ); + if (authParked && pendingAuthorizationActorId) { // Drop the dashboard connect prompt so a superseded OAuth flow // cannot leave a stale banner after the user moves on. await deleteWebAuthorization({ diff --git a/packages/junior/src/chat/task-execution/mailbox-turn.ts b/packages/junior/src/chat/task-execution/mailbox-turn.ts index bd558aa70e..668db9d600 100644 --- a/packages/junior/src/chat/task-execution/mailbox-turn.ts +++ b/packages/junior/src/chat/task-execution/mailbox-turn.ts @@ -14,6 +14,7 @@ import type { import { getTurnRecord, listTurnSummaries, + type TurnRecord, } from "@/chat/task-execution/checkpoint"; import { isEventConversationMessage, @@ -151,12 +152,14 @@ export async function getActiveConversationTurnId( : undefined; } -/** Return the active web or event mailbox Turn. */ -async function getActiveMailboxTurnId( +/** Return the web and event mailbox Turns that are paused or running. */ +export async function listActiveMailboxTurns( conversationId: string, -): Promise { +): Promise { const summaries = await listTurnSummaries(conversationId); - const active = ( + // TODO(dcramer): Remove the legacy surface and Actor checks after deployed + // Turn cursors all store Source.kind. + return ( await Promise.all( summaries .filter( @@ -167,8 +170,8 @@ async function getActiveMailboxTurnId( .map((summary) => getTurnRecord(conversationId, summary.turnId)), ) ).filter( - (record) => - record && + (record): record is TurnRecord => + record !== undefined && !record.dispatchId && (record.source?.kind === "web" || record.source?.kind === "event" || @@ -176,13 +179,18 @@ async function getActiveMailboxTurnId( (record.surface === "api" || hasEventActor(record.actors)))) && (record.state === "paused" || record.state === "running"), ); +} + +/** Return the active web or event mailbox Turn. */ +async function getActiveMailboxTurnId( + conversationId: string, +): Promise { + const active = await listActiveMailboxTurns(conversationId); if (active.length > 1) { throw new Error( `Conversation ${conversationId} has multiple active mailbox Turns`, ); } - // TODO(dcramer): Remove the legacy surface and Actor checks after deployed - // Turn cursors all store Source.kind. return active[0]?.turnId; } diff --git a/packages/junior/tests/integration/conversation-stale-turn.test.ts b/packages/junior/tests/integration/conversation-stale-turn.test.ts new file mode 100644 index 0000000000..24c8a28339 --- /dev/null +++ b/packages/junior/tests/integration/conversation-stale-turn.test.ts @@ -0,0 +1,106 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { createWebSource } from "@sentry/junior-plugin-api"; +import { + appendAndEnqueueWebMessage, + buildWebInboundMessage, + conversationTurnIdForMessage, + createAndEnqueueConversation, +} from "@/chat/conversations/web-input"; +import { createConversationTurnWorker } from "@/chat/task-execution/conversation-turn"; +import { resolveMailboxTurnWork } from "@/chat/task-execution/mailbox-turn"; +import { + getTurnRecord, + saveTurnCheckpoint, +} from "@/chat/task-execution/checkpoint"; +import { + closeConversationFixture, + createConversationFixture, + emptyConversationTurnAttempt, +} from "../fixtures/conversation"; + +describe("Conversation mailbox Turn after a stopped Turn", () => { + afterEach(async () => { + await closeConversationFixture(); + }); + + // JUNIOR-9M: conversation work stopped after repeated failed attempts and + // left the yielded Turn cursor open. The next Turn could start, but its + // resume failed with "multiple active mailbox Turns". + it("supersedes the open Turn cursor when new input starts a Turn", async () => { + const { actor, conversationStore, queue, state } = + await createConversationFixture(); + const first = await createAndEnqueueConversation( + { actor, idempotencyKey: "stale-turn-1", message: "First request." }, + { conversationStore, queue, state }, + ); + const { conversationId } = first; + const destination = { platform: "local" as const, conversationId }; + const staleTurnId = conversationTurnIdForMessage(first.messageId); + await saveTurnCheckpoint({ + mode: "paused", + conversationId, + turnId: staleTurnId, + sliceId: 1, + reason: "yield", + messages: [ + { + role: "user", + content: [{ type: "text", text: "First request." }], + timestamp: Date.now(), + }, + ], + destination, + source: createWebSource(conversationId), + actor, + surface: "api", + }); + + const second = await appendAndEnqueueWebMessage( + { + actor, + conversationId, + idempotencyKey: "stale-turn-2", + message: "Second request.", + }, + { conversationStore, queue, state }, + ); + const worker = createConversationTurnWorker({ + run: async () => { + throw new Error("Run stops after the Turn starts"); + }, + }); + const context = { + attempt: { + ...emptyConversationTurnAttempt({ conversationId, destination }), + messages: [ + buildWebInboundMessage({ + actor, + conversationId, + destination, + message: "Second request.", + messageId: second.messageId, + }), + ], + }, + conversationId, + destination, + shouldYield: () => false, + checkIn: async () => true, + }; + const resolved = await resolveMailboxTurnWork(context); + if (!resolved) throw new Error("Expected Conversation mailbox work"); + await expect(worker(context, resolved)).rejects.toThrow( + "Run stops after the Turn starts", + ); + + await expect( + getTurnRecord(conversationId, staleTurnId), + ).resolves.toMatchObject({ state: "abandoned" }); + await expect( + resolveMailboxTurnWork({ + ...context, + attempt: emptyConversationTurnAttempt({ conversationId, destination }), + }), + ).resolves.toBeUndefined(); + }); +});