diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 369608a289d2e..18af3345cd5b2 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -185,6 +185,13 @@ function codexFileChangeApprovalPrompt(input: { return input.grantRoot?.trim() || undefined; } +export function normalizeCodexThreadLoadModelSelection(selection: ModelSelection): ModelSelection { + // Codex applies effort on turn/start; resume discards its pinned request effort. + const { options, ...rest } = selection; + const loadOptions = options?.filter((option) => option.id !== "reasoningEffort"); + return { ...rest, ...(loadOptions?.length ? { options: loadOptions } : {}) }; +} + // Reasoning-effort changes preserve the tokenizer and context window; other changes need fresh telemetry. export function canReuseCodexContextUsage(previous: ModelSelection, next: ModelSelection): boolean { return modelSelectionsEqual( @@ -7046,6 +7053,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( providerSession: session, events: Stream.fromEffectRepeat(Queue.take(events)), canReuseContextUsage: canReuseCodexContextUsage, + normalizeThreadLoadModelSelection: normalizeCodexThreadLoadModelSelection, // Idle release can beat delayed subagent resumes: Codex reports them completed between turns with no resume-expected signal. hasPendingBackgroundWork: Effect.gen(function* () { for (const items of (yield* Ref.get(runningCommandItemsByTurn)).values()) { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 7c0332059cefd..c1b7c8dca8662 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -41,7 +41,10 @@ import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderS import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; -import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import { + CodexProviderCapabilitiesV2, + normalizeCodexThreadLoadModelSelection, +} from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as EventStore from "./EventStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -422,6 +425,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( state: Ref.Ref, options: { readonly failEventStream?: boolean; + readonly normalizeThreadLoadModelSelection?: (selection: ModelSelection) => ModelSelection; readonly capabilities?: OrchestrationV2ProviderCapabilities; readonly extensionStatuses?: OrchestrationV2ProviderSession["extensionStatuses"]; readonly mcpConfigs?: Ref.Ref< @@ -508,6 +512,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( instanceId: ProviderInstanceId.make("codex"), driver: CODEX_DRIVER, providerSessionId: input.providerSessionId, + ...(options.normalizeThreadLoadModelSelection === undefined + ? {} + : { normalizeThreadLoadModelSelection: options.normalizeThreadLoadModelSelection }), get providerSession() { return session; }, @@ -576,6 +583,7 @@ function layerTest(input: { readonly idleTimeoutMs: number; readonly maxIdlePinMs?: number; readonly failEventStream?: boolean; + readonly normalizeThreadLoadModelSelection?: (selection: ModelSelection) => ModelSelection; readonly capabilities?: OrchestrationV2ProviderCapabilities; readonly extensionStatuses?: OrchestrationV2ProviderSession["extensionStatuses"]; readonly mcpConfigs?: Ref.Ref< @@ -618,6 +626,9 @@ function layerTest(input: { const layerRegistry = Layer.unwrap( makeProviderAdapter(input.state, { failEventStream: input.failEventStream ?? false, + ...(input.normalizeThreadLoadModelSelection === undefined + ? {} + : { normalizeThreadLoadModelSelection: input.normalizeThreadLoadModelSelection }), ...(input.capabilities === undefined ? {} : { capabilities: input.capabilities }), ...(input.extensionStatuses === undefined ? {} @@ -3626,9 +3637,15 @@ it.effect("ProviderSessionManagerV2 persists session-scoped runtime requests wit }), ); -it.effect( - "ProviderSessionManagerV2 opens one shared runtime, broadcasts events, and detaches threads independently", - () => +it.effect.each([ + { name: "default load key", normalizeThreadLoadModelSelection: undefined }, + { + name: "Codex load key", + normalizeThreadLoadModelSelection: normalizeCodexThreadLoadModelSelection, + }, +])( + "ProviderSessionManagerV2 opens one shared runtime, broadcasts events, and detaches threads independently ($name)", + ({ normalizeThreadLoadModelSelection }) => Effect.gen(function* () { const state = yield* Ref.make(emptyState); const settings = yield* ServerSettings.ServerSettingsService.pipe( @@ -3725,25 +3742,91 @@ it.effect( const resumeSecondThread = secondRuntime.resumeThread({ providerThread: secondProviderThread, threadId: secondThreadId, - modelSelection, + modelSelection: { + ...modelSelection, + options: [{ id: "reasoningEffort", value: "medium" }], + }, runtimePolicy, }); yield* resumeSecondThread; yield* resumeSecondThread; assert.equal((yield* Ref.get(state)).resumeCount, 1); + let resumeCount = 1; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const appThread = (yield* projectionStore.getThreadProjection(secondThreadId)).thread; + const turnEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(turnEvents); + for (const [index, effort] of ["medium", "high", "medium"].entries()) { + const selection = { + ...modelSelection, + options: [{ id: "reasoningEffort", value: effort }], + } satisfies ModelSelection; + yield* secondRuntime.resumeThread({ + providerThread: secondProviderThread, + threadId: secondThreadId, + modelSelection: selection, + runtimePolicy, + }); + const ordinal = index + 1; + const runId = idAllocator.derive.run({ threadId: secondThreadId, ordinal }); + yield* secondRuntime.startTurn({ + appThread, + threadId: secondThreadId, + runId, + runOrdinal: ordinal, + providerTurnOrdinal: ordinal, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread: secondProviderThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ + threadId: secondThreadId, + ordinal, + }), + text: "turn", + attachments: [], + }, + modelSelection: selection, + runtimePolicy, + }); + const subscription = yield* secondRuntime.subscribeEvents!; + const terminal = yield* subscription.events.pipe( + Stream.filter((event) => event.type === "turn.terminal"), + Stream.runHead, + Effect.forkScoped, + ); + yield* Queue.offer(turnEvents!, { + type: "turn.terminal", + driver: CODEX_DRIVER, + providerThreadId: secondProviderThread.id, + providerTurnId: idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: `native-effort-turn-${ordinal}`, + }), + runOrdinal: ordinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(terminal); + if (normalizeThreadLoadModelSelection === undefined && index > 0) resumeCount++; + assert.equal((yield* Ref.get(state)).resumeCount, resumeCount, `${effort} effort reload`); + } yield* secondRuntime.resumeThread({ providerThread: secondProviderThread, threadId: secondThreadId, modelSelection: { ...modelSelection, model: "gpt-5.4-mini" }, runtimePolicy, }); - assert.equal((yield* Ref.get(state)).resumeCount, 2); + assert.equal((yield* Ref.get(state)).resumeCount, ++resumeCount); yield* resumeSecondThread; - assert.equal((yield* Ref.get(state)).resumeCount, 3); + assert.equal((yield* Ref.get(state)).resumeCount, ++resumeCount); // regression: #26722 a Linked environments change reloads only the idle thread's tools. yield* settings.updateSettings({ linkedEnvironmentsEnabled: true }); yield* resumeSecondThread; - assert.equal((yield* Ref.get(state)).resumeCount, 4); + assert.equal((yield* Ref.get(state)).resumeCount, ++resumeCount); assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]); const subscribe = firstRuntime.subscribeEvents; assert.isDefined(subscribe); @@ -3776,10 +3859,43 @@ it.effect( runtimePolicy, }); yield* resumeSecondThread; - assert.equal((yield* Ref.get(state)).resumeCount, 5); + assert.equal((yield* Ref.get(state)).resumeCount, ++resumeCount); assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]); + const reloadCases = [ + { + name: "other model options", + providerThread: secondProviderThread, + selection: { + ...modelSelection, + options: [{ id: "serviceTier", value: "priority" }], + }, + policy: runtimePolicy, + }, + { + name: "runtime policy", + providerThread: secondProviderThread, + selection: modelSelection, + policy: { ...runtimePolicy, interactionMode: "plan" }, + }, + ] satisfies ReadonlyArray<{ + name: string; + providerThread: OrchestrationV2ProviderThread; + selection: ModelSelection; + policy: ProviderAdapter.ProviderAdapterV2RuntimePolicy; + }>; + for (const { name, providerThread, selection, policy } of reloadCases) { + const before = (yield* Ref.get(state)).resumeCount; + yield* secondRuntime.resumeThread({ + providerThread, + threadId: secondThreadId, + modelSelection: selection, + runtimePolicy: policy, + }); + assert.equal((yield* Ref.get(state)).resumeCount, before + 1, `${name} did not reload`); + } + yield* manager.detach({ providerSessionId, threadId: firstThreadId }); assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); assert.equal((yield* Ref.get(state)).closeCount, 0); @@ -3800,6 +3916,9 @@ it.effect( layerTest({ state, idleTimeoutMs: 1000, + ...(normalizeThreadLoadModelSelection === undefined + ? {} + : { normalizeThreadLoadModelSelection }), serverSettingsLayer: Layer.succeed(ServerSettings.ServerSettingsService, settings), }), ), diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index b36119eb581d9..b6159679a43c5 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -305,15 +305,21 @@ function providerThreadRuntimeKey( } function providerThreadLoadKey(input: { + readonly runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime; readonly providerThread: Parameters< ProviderAdapter.ProviderAdapterV2SessionRuntime["resumeThread"] >[0]["providerThread"]; readonly modelSelection?: ModelSelection; readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy; }): string { + const modelSelection = + input.modelSelection === undefined + ? undefined + : (input.runtime.normalizeThreadLoadModelSelection?.(input.modelSelection) ?? + input.modelSelection); return JSON.stringify({ providerThread: providerThreadRuntimeKey(input.providerThread), - modelSelection: input.modelSelection ?? null, + modelSelection: modelSelection ?? null, runtimePolicy: input.runtimePolicy ?? null, }); } @@ -1936,6 +1942,7 @@ export const layerWithOptions = ( providerSessionId, threadId: input.threadId, providerThreadKey: providerThreadLoadKey({ + runtime, providerThread, modelSelection: input.modelSelection, runtimePolicy: input.runtimePolicy, @@ -1951,6 +1958,7 @@ export const layerWithOptions = ( return runtime.resumeThread(input); } const providerThreadKey = providerThreadLoadKey({ + runtime, providerThread: input.providerThread, ...(input.modelSelection === undefined ? {} @@ -1995,6 +2003,7 @@ export const layerWithOptions = ( providerSessionId, threadId, providerThreadKey: providerThreadLoadKey({ + runtime, providerThread, ...(input.modelSelection === undefined ? {} @@ -2024,6 +2033,7 @@ export const layerWithOptions = ( providerSessionId, threadId: input.targetThreadId, providerThreadKey: providerThreadLoadKey({ + runtime, providerThread, ...(input.modelSelection === undefined ? {} diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index e94740e34712f..b518dbcae7013 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -627,6 +627,8 @@ export interface ProviderAdapterV2SessionRuntime { cwd?: string | null, ) => number | undefined; readonly canReuseContextUsage?: (previous: ModelSelection, next: ModelSelection) => boolean; + /** Excludes turn-scoped options from native thread load matching. */ + readonly normalizeThreadLoadModelSelection?: (selection: ModelSelection) => ModelSelection; readonly ensureThread: ( input: ProviderAdapterV2EnsureThreadInput, ) => Effect.Effect;