fix(client): defer Codex turn custody mismatch

packages/client/src/__tests__/codex-app-server-handler.test.ts

@@ -467,6 +467,13 @@ function activeTurnNotSteerableError(): CodexAppServerRpcError {
   });
 }
 
+function activeTurnOwnershipMismatchError(expectedTurnId: string, foundTurnId: string): CodexAppServerRpcError {
+  return new CodexAppServerRpcError("turn/steer", {
+    code: -32602,
+    message: `expected active turn id \`${expectedTurnId}\` but found \`${foundTurnId}\``,
+  });
+}
+
 function stubLandingTrialHostEnv(suffix: string): void {
   const hostHome = join(workspaceRoot, "..", `host-home-${suffix}`);
   const codexHome = join(hostHome, ".codex");
@@ -1537,6 +1544,153 @@ describe("codex app-server handler", () => {
     await handler.shutdown();
   });
 
+  it("defers eight recovered inputs after active-turn ownership mismatch without a retry loop", async () => {
+    const fake = new FakeAppServerClient();
+    fake.deferNextSteer();
+    const finished: SessionMessage[][] = [];
+    const failSessionForRecovery = vi.fn<(reason: string, sessionId?: string) => void>();
+    const handler = makeHandler(fake);
+    const ctx = makeContext({
+      finishTurn: async (messages) => {
+        finished.push(Array.isArray(messages) ? [...messages] : [messages]);
+      },
+      failSessionForRecovery,
+    });
+
+    const startPromise = handler.start(makeMessage("m0", "active"), ctx);
+    await waitFor(() => fake.requests.some((request) => request.method === "turn/start"));
+
+    const recovered = Array.from({ length: 8 }, (_, index) =>
+      makeMessage(`m${index + 1}`, `recovered ${index + 1}`, 586_395 + index),
+    );
+    for (const message of recovered) handler.inject(message);
+    await waitFor(() => fake.requests.filter((request) => request.method === "turn/steer").length === 1);
+
+    fake.rejectSteer(activeTurnOwnershipMismatchError("turn-1", "provider-turn"));
+    await flushAsync();
+
+    expect(fake.requests.filter((request) => request.method === "turn/steer")).toHaveLength(1);
+    expect(failSessionForRecovery).not.toHaveBeenCalled();
+    expect(fake.isClosed).toBe(false);
+
+    completeTurn(fake, "turn-1", "active done");
+    await startPromise;
+    await waitFor(() => fake.requests.filter((request) => request.method === "turn/start").length === 2);
+
+    const recoveredStart = fake.requests.filter((request) => request.method === "turn/start")[1];
+    for (const message of recovered) {
+      expect(JSON.stringify(recoveredStart?.params)).toContain(message.content);
+    }
+
+    completeTurn(fake, "turn-2", "recovered done");
+    await waitFor(() => finished.length === 2);
+
+    expect(finished.map((messages) => messages.map((message) => message.id))).toEqual([
+      ["m0"],
+      recovered.map((message) => message.id),
+    ]);
+    expect(fake.requests.filter((request) => request.method === "turn/steer")).toHaveLength(1);
+
+    await handler.shutdown();
+  });
+
+  it("keeps new input behind a repeated active-turn ownership mismatch until settlement", async () => {
+    const fake = new FakeAppServerClient();
+    fake.steerError = activeTurnOwnershipMismatchError("turn-1", "provider-turn-1");
+    const finished: SessionMessage[][] = [];
+    const handler = makeHandler(fake);
+    const ctx = makeContext({
+      finishTurn: async (messages) => {
+        finished.push(Array.isArray(messages) ? [...messages] : [messages]);
+      },
+    });
+
+    const startPromise = handler.start(makeMessage("m1", "active"), ctx);
+    await waitFor(() => fake.requests.some((request) => request.method === "turn/start"));
+    handler.inject(makeMessage("m2", "first pending"));
+    await waitFor(() => fake.requests.filter((request) => request.method === "turn/steer").length === 1);
+    await flushAsync();
+
+    handler.inject(makeMessage("m3", "concurrent pending"));
+    handler.inject(makeMessage("m4", "later pending"));
+    await flushAsync();
+    await flushAsync();
+
+    expect(fake.requests.filter((request) => request.method === "turn/steer")).toHaveLength(1);
+
+    fake.steerError = activeTurnOwnershipMismatchError("turn-1", "provider-turn-2");
+    completeEmptyTurn(fake, "provider-turn-1");
+    await waitFor(() => fake.requests.filter((request) => request.method === "turn/steer").length === 2);
+    await flushAsync();
+
+    expect(fake.requests.filter((request) => request.method === "turn/steer")).toHaveLength(2);
+
+    fake.steerError = null;
+    completeEmptyTurn(fake, "provider-turn-2");
+    await waitFor(() => fake.requests.filter((request) => request.method === "turn/steer").length === 3);
+
+    completeTurn(fake, "turn-1", "active done");
+    await startPromise;
+    await waitFor(() => finished.length === 1);
+
+    expect(fake.requests.filter((request) => request.method === "turn/start")).toHaveLength(1);
+    expect(finished.map((messages) => messages.map((message) => message.id))).toEqual([["m1", "m2", "m3", "m4"]]);
+
+    await handler.shutdown();
+  });
+
+  it("retries after a concurrent active-turn settlement wins the mismatch response race", async () => {
+    const fake = new FakeAppServerClient();
+    fake.deferNextSteer();
+    const finished: SessionMessage[][] = [];
+    const handler = makeHandler(fake);
+    const ctx = makeContext({
+      finishTurn: async (messages) => {
+        finished.push(Array.isArray(messages) ? [...messages] : [messages]);
+      },
+    });
+
+    const startPromise = handler.start(makeMessage("m1", "active"), ctx);
+    await waitFor(() => fake.requests.some((request) => request.method === "turn/start"));
+    handler.inject(makeMessage("m2", "pending"));
+    await waitFor(() => fake.requests.filter((request) => request.method === "turn/steer").length === 1);
+
+    completeEmptyTurn(fake, "provider-turn");
+    fake.rejectSteer(activeTurnOwnershipMismatchError("turn-1", "provider-turn"));
+    await waitFor(() => fake.requests.filter((request) => request.method === "turn/steer").length === 2);
+
+    completeTurn(fake, "turn-1", "all done");
+    await startPromise;
+
+    expect(finished.map((messages) => messages.map((message) => message.id))).toEqual([["m1", "m2"]]);
+
+    await handler.shutdown();
+  });
+
+  it("does not defer an active-turn mismatch that names another expected turn", async () => {
+    const fake = new FakeAppServerClient();
+    fake.steerError = activeTurnOwnershipMismatchError("another-turn", "provider-turn");
+    const retryTurn = vi.fn<SessionContext["retryTurn"]>();
+    const failSessionForRecovery = vi.fn<(reason: string, sessionId?: string) => void>();
+    const handler = makeHandler(fake);
+    const ctx = makeContext({ retryTurn, failSessionForRecovery });
+
+    const startPromise = handler.start(makeMessage("m1", "active"), ctx);
+    await waitFor(() => fake.requests.some((request) => request.method === "turn/start"));
+    handler.inject(makeMessage("m2", "pending"));
+    await waitFor(() => failSessionForRecovery.mock.calls.length === 1);
+    await startPromise;
+
+    expect(retryTurn).toHaveBeenCalled();
+    expect(failSessionForRecovery).toHaveBeenCalledWith(
+      "codex_app_server_steer_unknown_custody_failed",
+      "thread-app-server",
+    );
+    expect(fake.isClosed).toBe(true);
+
+    await handler.shutdown();
+  });
+
   it("retries the ordered pending batch when input arrives during a no-custody steer", async () => {
     const fake = new FakeAppServerClient();
     fake.deferNextSteer();

packages/client/src/handlers/codex/app-server/index.ts

@@ -153,6 +153,12 @@ type CurrentTurn = {
    * generation and gives the still-ordered pending prefix one fresh attempt.
    */
   appendRejectedInputGeneration: number | null;
+  /**
+   * An exact active-turn ownership mismatch means Codex rejected the pending
+   * batch before custody. Keep the complete batch until the provider's active
+   * turn or this turn settles. New input must not reopen turn/steer early.
+   */
+  appendDeferredActiveTurnId: string | null;
   inFlightAppend: Promise<void> | null;
   finalAgentText: string;
   completedItemIds: Set<string>;
@@ -797,6 +803,14 @@ export const createCodexAppServerHandler: HandlerFactory = (config: HandlerConfi
       const error = params ? parseTurnError(params.error) : null;
       if (error) recordAppServerFailureSignal(turnStartAttempt, error);
     }
+    if (turn && notification.method === "turn/completed" && notificationTurnId === turn.appendDeferredActiveTurnId) {
+      turn.appendDeferredActiveTurnId = null;
+      sessionCtx.log(
+        `codex app-server prior active turn ${notificationTurnId} settled; pending input can retry turn/steer`,
+      );
+      schedulePendingDrain();
+      return;
+    }
     if (notificationTurnId && (!turn || turn.turnId !== notificationTurnId)) {
       const historicalTokenUsageRecorded = !turnStartInProgress && recordHistoricalTokenUsage(notification);
       if (!historicalTokenUsageRecorded) bufferNotification(notificationTurnId, notification);
@@ -1269,6 +1283,7 @@ export const createCodexAppServerHandler: HandlerFactory = (config: HandlerConfi
       primaryToken: token,
       acceptedMessages: [...messages],
       appendRejectedInputGeneration: null,
+      appendDeferredActiveTurnId: null,
       inFlightAppend: null,
       finalAgentText: "",
       completedItemIds: new Set(),
@@ -1571,6 +1586,7 @@ export const createCodexAppServerHandler: HandlerFactory = (config: HandlerConfi
     if (turn) {
       if (
         turn.status !== "inProgress" ||
+        turn.appendDeferredActiveTurnId !== null ||
         turn.appendRejectedInputGeneration === pendingInputGeneration ||
         turn.inFlightAppend
       ) {
@@ -1597,7 +1613,12 @@ export const createCodexAppServerHandler: HandlerFactory = (config: HandlerConfi
     pendingDrainInProgress = true;
     try {
       const turn = currentTurn;
-      if (turn && turn.status === "inProgress" && turn.appendRejectedInputGeneration !== pendingInputGeneration) {
+      if (
+        turn &&
+        turn.status === "inProgress" &&
+        turn.appendDeferredActiveTurnId === null &&
+        turn.appendRejectedInputGeneration !== pendingInputGeneration
+      ) {
         await appendPendingInputsToTurn(turn, sessionCtx);
         return;
       }
@@ -1649,7 +1670,11 @@ export const createCodexAppServerHandler: HandlerFactory = (config: HandlerConfi
     }
 
     if (currentTurn !== turn || turn.stopRequested || shutdownRequested) return;
-    if (turn.status !== "inProgress" || turn.appendRejectedInputGeneration === pendingInputGeneration) {
+    if (
+      turn.status !== "inProgress" ||
+      turn.appendDeferredActiveTurnId !== null ||
+      turn.appendRejectedInputGeneration === pendingInputGeneration
+    ) {
       return;
     }
 
@@ -1670,7 +1695,21 @@ export const createCodexAppServerHandler: HandlerFactory = (config: HandlerConfi
       for (const entry of batch) entry.token.processingStarted(entry.message);
       turn.acceptedMessages.push(...batch.map((entry) => entry.message));
     } catch (err) {
-      if (shouldFallbackSteerToNextTurn(err)) {
+      const ownershipMismatch = activeTurnOwnershipMismatch(err);
+      if (ownershipMismatch?.expectedTurnId === turn.turnId) {
+        if (consumeBufferedTurnCompletion(pendingNotificationsByTurn, ownershipMismatch.foundTurnId)) {
+          sessionCtx.log(
+            `codex app-server turn/steer rejected active-turn ownership after active turn ` +
+              `${ownershipMismatch.foundTurnId} settled; pending generation ${inputGeneration} can retry`,
+          );
+        } else {
+          turn.appendDeferredActiveTurnId = ownershipMismatch.foundTurnId;
+          sessionCtx.log(
+            `codex app-server turn/steer rejected active-turn ownership; pending generation ${inputGeneration} ` +
+              `will wait for active turn ${ownershipMismatch.foundTurnId} or turn ${turn.turnId} to settle`,
+          );
+        }
+      } else if (shouldFallbackSteerToNextTurn(err)) {
         // The provider definitively did not accept this batch. Block only the
         // queue generation that was attempted: a later input (including one
         // that arrived while this request was in flight) re-opens one ordered
@@ -2227,6 +2266,33 @@ function isTransientErrorInfo(value: unknown): boolean {
   );
 }
 
+type ActiveTurnOwnershipMismatch = {
+  expectedTurnId: string;
+  foundTurnId: string;
+};
+
+const ACTIVE_TURN_OWNERSHIP_MISMATCH_PATTERN = /^expected active turn id `([^`\s]+)` but found `([^`\s]+)`$/iu;
+
+function activeTurnOwnershipMismatch(err: unknown): ActiveTurnOwnershipMismatch | null {
+  if (!(err instanceof CodexAppServerRpcError) || err.code !== -32602) return null;
+  const match = ACTIVE_TURN_OWNERSHIP_MISMATCH_PATTERN.exec(err.message.trim());
+  if (!match?.[1] || !match[2] || match[1] === match[2]) return null;
+  return { expectedTurnId: match[1], foundTurnId: match[2] };
+}
+
+function consumeBufferedTurnCompletion(
+  pendingNotifications: Map<string, CodexAppServerNotification[]>,
+  turnId: string,
+): boolean {
+  const buffered = pendingNotifications.get(turnId);
+  if (!buffered) return false;
+  const completionIndex = buffered.findIndex((notification) => notification.method === "turn/completed");
+  if (completionIndex < 0) return false;
+  buffered.splice(completionIndex, 1);
+  if (buffered.length === 0) pendingNotifications.delete(turnId);
+  return true;
+}
+
 function shouldFallbackSteerToNextTurn(err: unknown): boolean {
   const message = err instanceof Error ? err.message.toLowerCase() : String(err).toLowerCase();
   if (

packages/qa/cases/runtime/session-resume-transient-retry-custody.md

@@ -146,6 +146,16 @@ must stop immediately: the second tail must not reach the provider, the
 already-terminal prefix may ACK, and the rejected tail plus untouched suffix
 must enter chat-scoped recovery without an ACK-prefix inversion.
 
+For a Codex app-server route, hold one valid active turn and recover exactly
+eight ordered inbox entries behind it. Make `turn/steer` return the structured
+active-turn ownership mismatch for that active turn. The handler must keep all
+eight entries pending without a session failure or a new handler. Add more
+input while the mismatch is pending and verify that the handler does not send
+another steer request. Complete the valid active turn, then verify that one
+new turn receives every pending entry once and in order. Repeat the mismatch
+and settlement boundary concurrently. The handler must not start a busy loop,
+duplicate an entry, or lose an entry.
+
 Also exercise an explicit control resume with no new user message. Inject the
 same transient pre-provider failure and verify the retry calls provider resume
 without a message. It must not synthesize an empty user turn or replay the most