fix(task-execution): Supersede stale mailbox Turns on new input

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,39 +516,38 @@ 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
               // late callback a resume no-op, matching Slack supersede behavior.
               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({

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<string | undefined> {
+): Promise<TurnRecord[]> {
   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,22 +170,27 @@ 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" ||
         (!record.source &&
           (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<string | undefined> {
+  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;
 }
 

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();
+  });
+});