diff --git a/packages/client/src/__tests__/codex-app-server-handler.test.ts b/packages/client/src/__tests__/codex-app-server-handler.test.ts index 270e0456b..0a538a1e3 100644 --- a/packages/client/src/__tests__/codex-app-server-handler.test.ts +++ b/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(); + 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(); diff --git a/packages/client/src/handlers/codex/app-server/index.ts b/packages/client/src/handlers/codex/app-server/index.ts index 66736a560..b3277a270 100644 --- a/packages/client/src/handlers/codex/app-server/index.ts +++ b/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 | null; finalAgentText: string; completedItemIds: Set; @@ -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, + 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 ( diff --git a/packages/qa/cases/runtime/session-resume-transient-retry-custody.md b/packages/qa/cases/runtime/session-resume-transient-retry-custody.md index 4aad07efd..3fda29614 100644 --- a/packages/qa/cases/runtime/session-resume-transient-retry-custody.md +++ b/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