diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 98652036118fc..7a49f3c957568 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1260,15 +1260,83 @@ 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 === "subscription_sharing_unsupported_capability", + ); + }); + if ( + rejectedHistory && + (yield* ( + session.hasPendingBackgroundWorkForThread?.(providerThread, { + includeSubagents: true, + }) ?? + session.hasPendingBackgroundWork ?? + Effect.succeed(false) + )) + ) { + return yield* loadFromProvider( + Effect.fail( + new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: session.driver, + threadId: projection.thread.id, + providerThreadId: providerThread.id, + runId, + cause: new ContextHandoffBackgroundWorkError(), + }), + ), + ); + } 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({ @@ -1348,9 +1416,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 83eee8f66a5dc..1807e06eb232f 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -113,6 +113,7 @@ function makeTestAdapter(input: { >; readonly failure?: ReturnType; readonly failedRunOrdinals?: ReadonlySet; + readonly rejectedNativeThreads?: Ref.Ref>; readonly interruptedRunOrdinals?: ReadonlySet; readonly cancelledRunOrdinals?: ReadonlySet; readonly holdRunOrdinal?: number; @@ -292,13 +293,27 @@ function makeTestAdapter(input: { const providerTurnId = ProviderTurnId.make( `provider-turn:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}`, ); - 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] ?? @@ -347,7 +362,7 @@ function makeTestAdapter(input: { messageId: MessageId.make( `message:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, ), - text: response, + text: rejectedHistory ? "" : response, streaming: false, }, }, @@ -1229,6 +1244,235 @@ describe("orchestration v2 provider switching", () => { ), ); + it.live.each([ + "live", + "detached", + "legacy", + "interrupted", + "cancelled", + "other-error", + "other-provider", + "repaired-native", + "background", + ] as const)("recovers rejected 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 modelSelection = + scenario === "other-provider" ? CLAUDE_MODEL_SELECTION : CODEX_MODEL_SELECTION; + const code = + scenario === "other-error" + ? "serverOverloaded" + : "subscription_sharing_unsupported_capability"; + const registry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: modelSelection.instanceId, + driver: scenario === "other-provider" ? CLAUDE_DRIVER : CODEX_DRIVER, + capabilities: + scenario === "other-provider" + ? ClaudeProviderCapabilitiesV2 + : CodexProviderCapabilitiesV2, + modelSelection, + capturedTurns, + injectedHistory, + nativeThreadGeneration: generation, + rejectedNativeThreads, + pendingBackgroundWork, + subagentOnlyBackgroundWork: true, + failedRunOrdinals: new Set([2]), + failure: makeProviderFailure({ + code, + message: "Unsupported input history", + 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, + 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, + 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, "Use the unsupported native agent capability"); + assert.equal(failed.runs.at(-1)?.status, "failed"); + assert.isTrue( + failed.turnItems.some((item) => item.type === "error" && item.failure.code === code), + ); + if (scenario === "detached" || 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 === "interrupted" || scenario === "cancelled") { + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make(`rejected:${scenario}`), + type: "run.updated", + threadId, + occurredAt: failed.thread.createdAt, + payload: { ...failed.runs.at(-1)!, status: scenario }, + }, + ], + }); + } + 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 === "other-provider" || + 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, '"agent_message"'); + 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) =>