diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 76c10382b3cb8..f855c2f718caf 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -7037,6 +7037,23 @@ describe("CodexAdapterV2 post-settle continuation", () => { expectedCode: "subscription_sharing_usage_limit_exceeded", expectedMessage: "Usage settings", }, + { + name: "invalid-encrypted-history", + code: "other", + message: JSON.stringify({ + error: { + type: "invalid_request_error", + code: "invalid_encrypted_content", + message: "Encrypted content could not be decrypted or parsed.", + }, + headers: { "x-codex-turn-state": "dummy-sensitive" }, + status: 400, + }), + notification: true, + expectedClass: "provider_error", + expectedCode: "invalid_encrypted_content", + expectedMessage: "encrypted conversation history", + }, { name: "sharing-code-mention-retry", code: "responseStreamDisconnected", @@ -7308,6 +7325,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { assert.equal(terminal?.status, "failed"); if (terminal?.status !== "failed") return; assert.equal(terminal.failure.class, scenario.expectedClass); + assert.notInclude(terminal.failure.message, "dummy-sensitive"); assert.equal(terminal.threadDisposition, "reusable"); if (scenario.name === "deferred-reset" || scenario.name === "first-request-reset") assert.equal(terminal.failure.resetAt, resetAt); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index f7d3b49a687e5..62d34fd3394f1 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1196,9 +1196,8 @@ function codexUsageLimitError( } /** - * ChatGPT sharing reports its own reason inside Codex's generic turn error, so - * the code it names replaces Codex's, and clients pick a recovery from it. Its - * allowance limits get the usage-limit recovery; other sharing failures keep their class. + * Codex wraps structured service failures in generic turn errors. Preserve the + * inner code and safe message; sharing allowance limits also change the class. */ function withCodexManagedFailure( failure: OrchestrationV2ProviderFailure, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 15792dacecfa2..8cb515589d059 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1,3 +1,4 @@ +import { classifyCodexManagedError } from "../provider/CodexManagedErrors.ts"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import { @@ -1258,15 +1259,82 @@ export const layer: Layer.Layer< handoff.delivery?.nativeThreadId === providerThread.nativeThreadRef?.nativeId && handoff.delivery?.status === "pending", ); + const rejectedHistory = yield* Effect.gen(function* () { + if (session.driver !== "codex") return false; + const acceptedAttemptIds = new Set( + projection.providerTurns.map((turn) => turn.runAttemptId), + ); + const previous = projection.runs.findLast( + (source) => + source.providerThreadId === providerThread.id && + source.ordinal < run.ordinal && + source.activeAttemptId !== null && + acceptedAttemptIds.has(source.activeAttemptId), + ); + if ( + previous === undefined || + (previous.status !== "failed" && + previous.status !== "interrupted" && + previous.status !== "cancelled") + ) + return false; + const previousAttempt = projection.attempts.find( + (source) => source.id === previous.activeAttemptId, + ); + if (previousAttempt === undefined) return false; + if (previousAttempt.nativeThreadId !== undefined) { + if (previousAttempt.nativeThreadId !== providerThread.nativeThreadRef?.nativeId) + return false; + } else if ( + projection.contextHandoffs.some( + (handoff) => + handoff.toProviderThreadId === providerThread.id && + handoff.strategy === "full_thread_summary" && + handoff.delivery?.nativeThreadId === providerThread.nativeThreadRef?.nativeId && + handoff.coveredRunOrdinals.to >= previous.ordinal, + ) + ) + return false; + // The native resume RPC can succeed while persisted input still rejects every turn. + return (yield* projectionStore.getTurnStartHistory(input.threadId, [previous.id])).some( + (item) => + item.type === "error" && + item.nodeId === previous.rootNodeId && + (item.failure.code === "invalid_encrypted_content" || + classifyCodexManagedError(item.failure.message)?.code === + "invalid_encrypted_content"), + ); + }); + // A fresh session would orphan background work still running on the native thread. + const hasPendingWork = (thread: OrchestrationV2ProviderThread) => + session.hasPendingBackgroundWorkForThread?.(thread, { includeSubagents: true }) ?? + session.hasPendingBackgroundWork ?? + Effect.succeed(false); + const failForPendingWork = loadFromProvider( + Effect.fail( + new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: session.driver, + threadId: projection.thread.id, + providerThreadId: providerThread.id, + runId, + cause: new ContextHandoffBackgroundWorkError(), + }), + ), + ); + if (rejectedHistory && (yield* hasPendingWork(providerThread))) { + return yield* failForPendingWork; + } const resumed = yield* Effect.result( - uncertainDelivery + uncertainDelivery || rejectedHistory ? Effect.fail( new ProviderAdapter.ProviderAdapterTurnStartError({ driver: session.driver, threadId: projection.thread.id, providerThreadId: providerThread.id, runId, - cause: "Uncertain native history injection", + cause: rejectedHistory + ? "Rejected native history" + : "Uncertain native history injection", }), ) : session.resumeThread({ @@ -1307,26 +1375,7 @@ export const layer: Layer.Layer< Effect.catchTags({ ContextHandoffBudgetError: () => Effect.succeed(false) }), ); if (!fitsFresh) return resumed.success; - const hasPendingWork = yield* ( - session.hasPendingBackgroundWorkForThread?.(resumed.success, { - includeSubagents: true, - }) ?? - session.hasPendingBackgroundWork ?? - Effect.succeed(false) - ); - if (hasPendingWork) { - return yield* loadFromProvider( - Effect.fail( - new ProviderAdapter.ProviderAdapterTurnStartError({ - driver: session.driver, - threadId: projection.thread.id, - providerThreadId: providerThread.id, - runId, - cause: new ContextHandoffBackgroundWorkError(), - }), - ), - ); - } + if (yield* hasPendingWork(resumed.success)) return yield* failForPendingWork; } resumedHandoffContext = undefined; resumedRetryHandoff = []; @@ -1346,9 +1395,11 @@ export const layer: Layer.Layer< runId, reason: uncertainDelivery ? "uncertain_history_delivery" - : resumed._tag === "Success" - ? "handoff_budget_exhausted" - : "resume_failed", + : rejectedHistory + ? "rejected_native_history" + : resumed._tag === "Success" + ? "handoff_budget_exhausted" + : "resume_failed", errorTag: resumed._tag === "Failure" ? resumed.failure._tag : "ContextHandoffBudgetError", }); const replacement = yield* loadFromProvider( diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 965e5bd44f2dc..4f4c7e213c8ee 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -115,6 +115,7 @@ function makeTestAdapter(input: { Record> >; readonly failure?: ReturnType; + readonly rejectedNativeThreads?: Ref.Ref>; readonly failedRunOrdinals?: ReadonlySet; readonly interruptedRunOrdinals?: ReadonlySet; readonly cancelledRunOrdinals?: ReadonlySet; @@ -345,13 +346,27 @@ function makeTestAdapter(input: { yield* Deferred.await(input.releaseFirstTurn); } const eventTime = yield* DateTime.now; - const terminalStatus = input.failedRunOrdinals?.has(turnInput.runOrdinal) - ? "failed" - : input.interruptedRunOrdinals?.has(turnInput.runOrdinal) - ? "interrupted" - : input.cancelledRunOrdinals?.has(turnInput.runOrdinal) - ? "cancelled" - : "completed"; + const nativeThreadId = turnInput.providerThread.nativeThreadRef?.nativeId ?? ""; + if ( + input.rejectedNativeThreads && + input.failedRunOrdinals?.has(turnInput.runOrdinal) + ) { + yield* Ref.update( + input.rejectedNativeThreads, + (ids) => new Set([...ids, nativeThreadId]), + ); + } + const rejectedHistory = + input.rejectedNativeThreads !== undefined && + (yield* Ref.get(input.rejectedNativeThreads)).has(nativeThreadId); + const terminalStatus = + input.failedRunOrdinals?.has(turnInput.runOrdinal) || rejectedHistory + ? "failed" + : input.interruptedRunOrdinals?.has(turnInput.runOrdinal) + ? "interrupted" + : input.cancelledRunOrdinals?.has(turnInput.runOrdinal) + ? "cancelled" + : "completed"; const response = input.responseByThreadId?.[turnInput.threadId]?.[turnInput.runOrdinal] ?? input.responseByRunOrdinal[turnInput.runOrdinal] ?? @@ -400,7 +415,7 @@ function makeTestAdapter(input: { messageId: MessageId.make( `message:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, ), - text: response, + text: rejectedHistory ? "" : response, streaming: false, }, }, @@ -1369,6 +1384,220 @@ describe("orchestration v2 provider switching", () => { }), ); + it.live.each([ + "live", + "legacy", + "raw-error", + "other-error", + "repaired-native", + "background", + ] as const)("recovers invalid encrypted Codex history on the next prompt: %s", (scenario) => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("codex-rejected-history"); + const capturedTurns = yield* Ref.make>([]); + const injectedHistory = yield* Ref.make>([]); + const generation = yield* Ref.make(0); + const rejectedNativeThreads = yield* Ref.make>(new Set()); + const pendingBackgroundWork = yield* Ref.make(false); + const code = + scenario === "raw-error" + ? "other" + : scenario === "other-error" + ? "serverOverloaded" + : "invalid_encrypted_content"; + const registry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + capturedTurns, + injectedHistory, + nativeThreadGeneration: generation, + rejectedNativeThreads, + pendingBackgroundWork, + subagentOnlyBackgroundWork: true, + failedRunOrdinals: new Set([2]), + failure: makeProviderFailure({ + code, + message: + scenario === "raw-error" + ? JSON.stringify({ + error: { + code: "invalid_encrypted_content", + message: "Encrypted content could not be verified", + }, + }) + : "Encrypted content could not be verified", + class: "provider_error", + }), + responseByRunOrdinal: { 1: "BASELINE_OK", 3: "RECOVERY_OK", 4: "FOLLOWUP_OK" }, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const dispatch = Effect.fn("rejectedHistory.dispatch")(function* ( + ordinal: number, + text: string, + ) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`rejected:${ordinal}`), + threadId, + messageId: MessageId.make(`rejected:${ordinal}`), + createdBy: "user", + creationSource: "web", + text, + attachments: [], + modelSelection: CODEX_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.ordinal === ordinal && + (event.payload.status === "completed" || event.payload.status === "failed"), + ), + Stream.runHead, + ); + yield* worker.drain(); + return yield* orchestrator.getThreadProjection(threadId); + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("rejected:create"), + threadId, + projectId, + createdBy: "user", + creationSource: "web", + title: "Rejected Codex history", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + assert.equal( + (yield* dispatch(1, "Reply BASELINE_OK; remember the constraint")).runs.at(-1)?.status, + "completed", + ); + const failed = yield* dispatch(2, "Continue using encrypted history"); + assert.equal(failed.runs.at(-1)?.status, "failed"); + assert.isTrue( + failed.turnItems.some((item) => item.type === "error" && item.failure.code === code), + ); + if (scenario === "legacy") { + yield* orchestrator.dispatch({ + type: "provider-session.detach", + commandId: CommandId.make("rejected:detach"), + threadId, + providerSessionId: failed.providerThreads[0]!.providerSessionId!, + }); + yield* worker.drain(); + } + if (scenario === "legacy") { + yield* (yield* EventSink.EventSinkV2).write({ + events: failed.attempts.map( + ({ nativeThreadId: _nativeThreadId, ...attempt }, index) => ({ + id: EventId.make(`rejected:legacy:${index}`), + type: "run-attempt.updated" as const, + threadId, + occurredAt: failed.thread.createdAt, + payload: attempt, + }), + ), + }); + } + if (scenario === "repaired-native") { + const providerThread = failed.providerThreads[0]!; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("rejected:repaired-native"), + type: "provider-thread.updated", + threadId, + occurredAt: failed.thread.createdAt, + payload: { + ...providerThread, + nativeThreadRef: { + ...providerThread.nativeThreadRef!, + nativeId: "already-repaired", + }, + }, + }, + ], + }); + } + if (scenario === "background") yield* Ref.set(pendingBackgroundWork, true); + const recovered = yield* dispatch(3, "huh"); + if (scenario === "repaired-native") { + assert.equal(recovered.runs.at(-1)?.status, "completed"); + assert.equal(yield* Ref.get(generation), 1); + assert.lengthOf(recovered.contextHandoffs, 0); + return; + } + if (scenario === "other-error" || scenario === "background") { + assert.equal(recovered.runs.at(-1)?.status, "failed"); + assert.equal(yield* Ref.get(generation), 1); + assert.lengthOf(yield* Ref.get(capturedTurns), scenario === "background" ? 2 : 3); + if (scenario === "background") { + yield* Ref.set(pendingBackgroundWork, false); + assert.equal( + (yield* dispatch(4, "Continue after the worker finished")).runs.at(-1)?.status, + "completed", + ); + assert.equal(yield* Ref.get(generation), 2); + } + return; + } + assert.equal(recovered.runs.at(-1)?.status, "completed"); + assert.equal(yield* Ref.get(generation), 2); + const turns = yield* Ref.get(capturedTurns); + assert.lengthOf(turns, 3); + assert.equal(turns[2]!.text, "huh"); + assert.notEqual(turns[2]!.nativeThreadId, turns[1]!.nativeThreadId); + assert.equal(turns[2]!.providerThreadId, turns[1]!.providerThreadId); + const history = yield* encodeJson(yield* Ref.get(injectedHistory)); + assert.include(history, "remember the constraint"); + assert.include(history, "BASELINE_OK"); + assert.notInclude(history, '"encrypted_content"'); + assert.notInclude(history, "huh"); + assert.isTrue( + recovered.turnItems.some( + (item) => item.type === "assistant_message" && item.text === "RECOVERY_OK", + ), + ); + const next = yield* dispatch(4, "ok"); + assert.equal(next.runs.at(-1)?.status, "completed"); + assert.equal(yield* Ref.get(generation), 2); + assert.equal( + (yield* Ref.get(capturedTurns))[3]!.nativeThreadId, + turns[2]!.nativeThreadId, + ); + assert.equal(yield* encodeJson(yield* Ref.get(injectedHistory)), history); + assert.equal(next.contextHandoffs.length, recovered.contextHandoffs.length); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name: "codex-rejected-history", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly" }, + }, + }, + registry, + ), + ), + ); + }), + ), + ); + it.live.each(["native", "legacy"] as const)( "starts a replacement native thread as new after the resume fallback with %s attempts", (attempts) => diff --git a/apps/server/src/provider/CodexManagedErrors.ts b/apps/server/src/provider/CodexManagedErrors.ts index 4cad5709d2eb0..a699d31eaae67 100644 --- a/apps/server/src/provider/CodexManagedErrors.ts +++ b/apps/server/src/provider/CodexManagedErrors.ts @@ -34,6 +34,10 @@ const legacyFailures = { }, } as const; const failures = { + invalid_encrypted_content: { + message: + "Codex could not verify this session's encrypted conversation history. Send another message to continue in a fresh session with the visible conversation history.", + }, ...legacyFailures, subscription_sharing_user_not_eligible: legacyFailures.subscription_sharing_v2_user_not_eligible, subscription_sharing_route_not_supported: