diff --git a/src/main/codex/codex-close-turn-end.test.ts b/src/main/codex/codex-close-turn-end.test.ts index dae441a3526b..eadf11c60055 100644 --- a/src/main/codex/codex-close-turn-end.test.ts +++ b/src/main/codex/codex-close-turn-end.test.ts @@ -64,7 +64,6 @@ function sessionWithRunningTurn() { prompts: new CodexPromptRegistry(), options: new Map(), reportedOptions: {}, - fastModeTierByModel: new Map(), dispatchEchoes: createCodexDispatchEchoes(), turnOpenWaits: createCodexTurnOpenWaits(), translator diff --git a/src/main/codex/codex-configured-launch-defaults.ts b/src/main/codex/codex-configured-launch-defaults.ts index f9b56acbbe88..3ccc9f5dcb06 100644 --- a/src/main/codex/codex-configured-launch-defaults.ts +++ b/src/main/codex/codex-configured-launch-defaults.ts @@ -12,13 +12,14 @@ function configuredText(value: unknown): string | null { /** The model and effort the user's Codex config selects, or null when Codex can't say. */ export async function readCodexConfiguredLaunchDefaults( - connection: Pick + connection: Pick, + timeoutMs = CONFIG_READ_TIMEOUT_MS ): Promise { try { const response = await connection.request( 'config/read', {}, - { timeoutMs: CONFIG_READ_TIMEOUT_MS } + { timeoutMs: Math.min(timeoutMs, CONFIG_READ_TIMEOUT_MS) } ) const config = typeof response === 'object' && response !== null && 'config' in response diff --git a/src/main/codex/codex-requested-close-turn-timing.test.ts b/src/main/codex/codex-requested-close-turn-timing.test.ts index 8652919924f9..b45ac6d98921 100644 --- a/src/main/codex/codex-requested-close-turn-timing.test.ts +++ b/src/main/codex/codex-requested-close-turn-timing.test.ts @@ -59,7 +59,6 @@ describe('requested-close durable turn timing', () => { prompts: new CodexPromptRegistry(), options: new Map(), reportedOptions: {}, - fastModeTierByModel: new Map(), dispatchEchoes: createCodexDispatchEchoes(), turnOpenWaits: createCodexTurnOpenWaits(), translator diff --git a/src/main/codex/codex-structured-acquire-catalog.ts b/src/main/codex/codex-structured-acquire-catalog.ts index 0e66d5c4ae02..3a44686f4d57 100644 --- a/src/main/codex/codex-structured-acquire-catalog.ts +++ b/src/main/codex/codex-structured-acquire-catalog.ts @@ -1,4 +1,3 @@ -import type { CodexAppServerConnection } from './codex-app-server-connection' import type { CodexOpenedThread } from './codex-structured-thread-open' import type { CodexSessionCatalogAccess, @@ -20,23 +19,16 @@ export function codexAcquireCatalogAccess( return agentModelCatalogSessionAccess(deps.modelCatalog, CODEX_STRUCTURED_AGENT, launch.codexHome) } -/** The one catalog read a fast-mode restore needs, store-first. Null degrades - * exactly as a failed listing always did: the restore proceeds without tiers. */ -export async function codexAcquireFastModeCatalog(input: { - connection: Pick +/** Use saved catalog knowledge for Fast restore without waiting on discovery. */ +export function codexAcquireFastModeCatalog(input: { catalogAccess: CodexSessionCatalogAccess | undefined opened: Pick restoreNeedsCatalog: boolean - timeoutMs: number | undefined -}): Promise { +}): CodexSessionOptionCatalog | null { if (!input.restoreNeedsCatalog) { return null } - const listing = await codexAcquireCatalogListing( - input.connection, - input.catalogAccess, - input.timeoutMs - ) + const listing = codexAcquireCatalogListing(input.catalogAccess) if (!listing) { return null } diff --git a/src/main/codex/codex-structured-acquisition-notification.ts b/src/main/codex/codex-structured-acquisition-notification.ts new file mode 100644 index 000000000000..bd0e3f3b6f64 --- /dev/null +++ b/src/main/codex/codex-structured-acquisition-notification.ts @@ -0,0 +1,39 @@ +import type { CodexAppServerConnectionHandlers } from './codex-app-server-connection' +import type { CodexAcquisitionWindow } from './codex-structured-acquisition-window' +import type { CodexDispatchEchoes } from './codex-structured-dispatch-echo' +import type { CodexStructuredNotificationRetry } from './codex-structured-notification-retry' +import { CODEX_RECEIPT_TIMED_METHODS } from './codex-structured-session-state' + +/** Stamp provider receipt before pre-publication buffering or retry. */ +export function codexAcquisitionNotificationHandler(input: { + acquisition: CodexAcquisitionWindow + sessionId: string + dispatchEchoes: CodexDispatchEchoes + notificationRetries: CodexStructuredNotificationRetry + deliver: ( + acquisition: CodexAcquisitionWindow, + sessionId: string, + event: () => unknown, + retainedBytes?: number + ) => void + now: () => number +}): NonNullable { + return (method, params) => { + const observedAt = CODEX_RECEIPT_TIMED_METHODS.has(method) ? input.now() : undefined + const dispatchSequenceAtReceipt = + method === 'turn/started' ? input.dispatchEchoes.latestSequence() : undefined + input.deliver( + input.acquisition, + input.sessionId, + () => + input.notificationRetries.handle( + input.sessionId, + method, + params, + observedAt, + dispatchSequenceAtReceipt + ), + Buffer.byteLength(JSON.stringify(params ?? null), 'utf8') + ) + } +} diff --git a/src/main/codex/codex-structured-background-catalog.ts b/src/main/codex/codex-structured-background-catalog.ts new file mode 100644 index 000000000000..ed4eda7aa11d --- /dev/null +++ b/src/main/codex/codex-structured-background-catalog.ts @@ -0,0 +1,110 @@ +import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { + fetchCodexModelCatalogListing, + type CodexModelCatalogListing +} from './codex-structured-model-catalog' +import type { CodexSession } from './codex-structured-session-state' +import { reconcileCodexFastModeOption } from './codex-structured-fast-mode' + +type BackgroundCatalogInput = { + session: CodexSession + sessionId: string + sessions: ReadonlyMap + timeoutMs: number | undefined + logger: StructuredAgentSessionLogger | undefined +} + +/** Turns read tiers from the store; only a legacy saved tier needs migrating here. */ +function applyListing(session: CodexSession, listing: CodexModelCatalogListing): void { + if (!session.options.has('serviceTier')) { + return + } + const model = + session.options.get('model') ?? + session.reportedOptions.model ?? + listing.models.find((entry) => entry.isDefault)?.id ?? + listing.models[0]?.id ?? + '' + reconcileCodexFastModeOption(session, { + fastModeTierByModel: listing.fastModeTierByModel, + currentFastMode: undefined, + model, + modelFastModeSupport: undefined + }) +} + +async function refresh(input: BackgroundCatalogInput): Promise { + const { session, sessionId, sessions } = input + const access = session.catalogAccess + const store = access?.store + const fingerprint = access?.fingerprint + const prior = fingerprint ? store?.get(fingerprint) : undefined + const model = session.options.get('model') ?? session.reportedOptions.model + const needsTier = + (session.options.get('fastMode') === 'true' || session.options.has('serviceTier')) && + (!model || !prior?.fastModeTierByModel[model]) + if ( + fingerprint && + store && + (store.hasActiveFailure(fingerprint) || (prior && !store.isStale(prior) && !needsTier)) + ) { + return + } + const isCurrent = (): boolean => sessions.get(sessionId) === session && !session.ended + try { + const listing = await fetchCodexModelCatalogListing({ + connection: session.connection, + timeoutMs: input.timeoutMs, + deadlineMs: Math.min(input.timeoutMs ?? 30_000, 30_000) + }) + if (listing.models.length === 0) { + throw new Error('codex app-server returned no available models') + } + if (!isCurrent()) { + return + } + const latest = fingerprint ? store?.get(fingerprint) : undefined + if (latest && latest !== prior) { + applyListing(session, { + models: latest.models, + fastModeTierByModel: new Map(Object.entries(latest.fastModeTierByModel)) + }) + return + } + if (fingerprint && store) { + store.recordSuccess(fingerprint, 'codex', { + models: listing.models, + fastModeTierByModel: listing.fastModeTierByModel, + origin: 'live-session' + }) + } + applyListing(session, listing) + } catch (error) { + if (!isCurrent()) { + return + } + if (fingerprint && store && store.get(fingerprint) === prior) { + store.recordFailure(fingerprint, error instanceof Error ? error.message : String(error)) + } + input.logger?.warn('Codex model catalog refresh failed', { + scope: 'codex-background-catalog', + sessionId, + error + }) + } +} + +/** Catalog work begins after ownership and never joins a chat action. */ +export function startBackgroundCodexCatalogRefresh(input: BackgroundCatalogInput): void { + void refresh(input).catch((error: unknown) => { + try { + input.logger?.warn('Codex model catalog refresh failed', { + scope: 'codex-background-catalog', + sessionId: input.sessionId, + error + }) + } catch { + // Reporting must never reject an unawaited task. + } + }) +} diff --git a/src/main/codex/codex-structured-catalog-entry.ts b/src/main/codex/codex-structured-catalog-entry.ts new file mode 100644 index 000000000000..eff6365d1ce1 --- /dev/null +++ b/src/main/codex/codex-structured-catalog-entry.ts @@ -0,0 +1,27 @@ +import type { AgentModelCatalogEntry } from '../native-chat/agent-model-catalog/agent-model-catalog-store' +import type { CodexSessionCatalogAccess } from './codex-structured-session-state' +import type { CodexModelCatalogListing } from './codex-structured-model-catalog' + +export function listingFromEntry(entry: AgentModelCatalogEntry): CodexModelCatalogListing { + return { + models: entry.models.map((model) => ({ ...model })), + fastModeTierByModel: new Map(Object.entries(entry.fastModeTierByModel)) + } +} + +/** The exact Fast tier the account's stored catalog names for a model; no I/O. */ +export function codexKnownFastModeTier( + catalogAccess: CodexSessionCatalogAccess | undefined, + model: string +): string | undefined { + const tiers = catalogAccess?.store.get(catalogAccess.fingerprint)?.fastModeTierByModel + return tiers && Object.hasOwn(tiers, model) ? tiers[model] : undefined +} + +/** Acquisition uses known choices only; picker reads discover new choices later. */ +export function codexAcquireCatalogListing( + catalogAccess: CodexSessionCatalogAccess | undefined +): CodexModelCatalogListing | null { + const entry = catalogAccess?.store.get(catalogAccess.fingerprint) + return entry ? listingFromEntry(entry) : null +} diff --git a/src/main/codex/codex-structured-fast-mode.test.ts b/src/main/codex/codex-structured-fast-mode.test.ts index 5a40d622af0a..4182014d14af 100644 --- a/src/main/codex/codex-structured-fast-mode.test.ts +++ b/src/main/codex/codex-structured-fast-mode.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, vi } from 'vitest' +import { describe, expect, it, vi, type Mock } from 'vitest' import { THREAD_ID, USER_MESSAGE, @@ -8,146 +8,400 @@ import { identityFor, type Route } from './codex-structured-session-adapter-fixture' +import { + AGENT_MODEL_CATALOG_FAILURE_TTL_MS, + AgentModelCatalogStore +} from '../native-chat/agent-model-catalog/agent-model-catalog-store' +import { agentModelCatalogSessionAccess } from '../native-chat/agent-model-catalog/agent-model-catalog-fingerprint' +import type { CodexStructuredSessionAdapter } from './codex-structured-session-adapter' +import { CODEX_STRUCTURED_AGENT } from './codex-structured-agent-definition' + +function listing(tier = 'priority-live-v2') { + return { + data: [ + { + model: 'gpt-live', + supportedReasoningEfforts: [], + serviceTiers: [{ id: tier, name: 'Fast' }] + } + ], + nextCursor: null + } +} -describe('Codex structured Fast mode dispatch', () => { - it('uses the provider-advertised Fast tier on the first turn after acquisition', async () => { +async function acquire(adapter: CodexStructuredSessionAdapter, sessionId = 'session-1') { + return adapter.acquire({ + identity: identityFor(sessionId), + fence: 7, + spawnToken: `spawn-${sessionId}`, + options: { fastMode: 'true' } + }) +} + +async function send(adapter: CodexStructuredSessionAdapter, id: string, sessionId = 'session-1') { + return adapter.dispatch({ sessionId, clientMessageId: id, body: USER_MESSAGE, fence: 7 }) +} + +function finishTurn(codex: ReturnType, turnId: string) { + codex.connections.at(-1)?.handlers.onNotification?.('turn/completed', { + threadId: THREAD_ID, + turn: { id: turnId, status: 'completed' } + }) +} + +describe('Codex structured Fast mode without send-path catalog waits', () => { + it('starts the chat and sends Standard when its own listing rejects', async () => { const codex = fakeCodex({ - 'model/list': () => ({ - data: [ - { - model: 'gpt-live', - supportedReasoningEfforts: [], - serviceTiers: [{ id: 'priority-live-v2', name: 'Fast' }] - } - ], - nextCursor: null - }) + 'model/list': () => { + throw new Error('model listing unavailable') + } }) - codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-fast') + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-standard') const adapter = adapterFor(codex) - await adapter.acquire({ - identity: identityFor('session-1'), - fence: 7, - spawnToken: 'spawn-9', - options: { fastMode: 'true' } + + await acquire(adapter) + expect(adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 7 })).toMatchObject({ + fastMode: 'true' }) + await send(adapter, 'first') + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'default' }) + }) - await adapter.dispatch({ - sessionId: 'session-1', - clientMessageId: 'client-fast', - body: USER_MESSAGE, - fence: 7 + it('uses a stored exact tier immediately on the first turn', async () => { + const codex = fakeCodex() + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-fast') + const modelCatalog = new AgentModelCatalogStore() + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + modelCatalog.recordSuccess(access.fingerprint, 'codex', { + models: [{ id: 'gpt-live', label: 'GPT Live', isDefault: true, efforts: [] }], + fastModeTierByModel: new Map([['gpt-live', 'priority-live-v2']]), + origin: 'live-session' }) + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { modelCatalog }) + await acquire(adapter) + await send(adapter, 'first') + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'priority-live-v2' }) + expect(codex.connections[0].calls.some((call) => call.method === 'model/list')).toBe(false) + }) + it('sends Standard while its own bounded listing is pending, then uses Fast on a later turn', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'first-turn') + const modelCatalog = new AgentModelCatalogStore() + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { modelCatalog }) + await acquire(adapter) + await vi.waitFor(() => + expect(codex.connections[0].calls.some((call) => call.method === 'model/list')).toBe(true) + ) + await send(adapter, 'first') expect( codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'default' }) + finishTurn(codex, 'first-turn') + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'second-turn') + pending.resolve(listing()) + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + await vi.waitFor(() => + expect(modelCatalog.get(access.fingerprint)?.fastModeTierByModel['gpt-live']).toBe( + 'priority-live-v2' + ) + ) + await send(adapter, 'second') + expect( + codex.connections[0].calls.filter((call) => call.method === 'turn/start')[1]?.params ).toMatchObject({ serviceTier: 'priority-live-v2' }) }) - it('uses Standard on the first turn after acquisition with Fast explicitly off', async () => { - const codex = fakeCodex() - codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-standard') + it('steers an active turn while catalog discovery remains pending', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'running-turn') + codex.routes['turn/steer'] = () => ({ turn: { id: 'running-turn' } }) const adapter = adapterFor(codex) - await adapter.acquire({ - identity: identityFor('session-1'), - fence: 7, - spawnToken: 'spawn-9', - options: { fastMode: 'false' } - }) + await acquire(adapter) + await send(adapter, 'first') + await send(adapter, 'follow-up') + expect(codex.connections[0].calls.some((call) => call.method === 'turn/steer')).toBe(true) + expect(codex.connections[0].calls.filter((call) => call.method === 'model/list')).toHaveLength( + 1 + ) + pending.resolve(listing()) + }) - await adapter.dispatch({ - sessionId: 'session-1', - clientMessageId: 'client-standard', - body: USER_MESSAGE, - fence: 7 - }) + it('stops a running turn and starts its replacement as Standard while listing stalls', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'running-turn') + const adapter = adapterFor(codex) + await acquire(adapter) + await send(adapter, 'first') + await expect(adapter.cancelTurn({ sessionId: 'session-1', fence: 7 })).resolves.toMatchObject({ + cancelled: true + }) + finishTurn(codex, 'running-turn') + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'replacement-turn') + await send(adapter, 'replacement') + expect(codex.connections[0].calls.filter((call) => call.method === 'turn/start')).toHaveLength( + 2 + ) expect( - codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + codex.connections[0].calls.filter((call) => call.method === 'turn/start')[1]?.params ).toMatchObject({ serviceTier: 'default' }) - expect(codex.connections[0].calls.some((call) => call.method === 'model/list')).toBe(false) + pending.resolve(listing()) }) - it.each(['absent', 'transient'] as const)( - 'uses Standard while restored Fast discovery is %s, then recovers the exact tier', - async (discovery) => { - const unavailableCatalog = () => { - if (discovery === 'transient') { - throw new Error('catalog temporarily unavailable') - } - return { - data: [{ model: 'gpt-live', supportedReasoningEfforts: [] }], - nextCursor: null - } - } - const listModels = vi.fn().mockImplementationOnce(unavailableCatalog) - if (discovery === 'absent') { - listModels.mockImplementationOnce(unavailableCatalog) - } - listModels.mockImplementation(() => ({ - data: [ - { - model: 'gpt-live', - supportedReasoningEfforts: [], - serviceTiers: [{ id: 'priority-recovered', name: 'Fast' }] - } - ], - nextCursor: null - })) - const codex = fakeCodex({ 'model/list': listModels }) - codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-recovered') - const adapter = adapterFor(codex) - await expect( - adapter.acquire({ - identity: identityFor('session-1'), - fence: 7, - spawnToken: 'spawn-9', - options: { fastMode: 'true' } - }) - ).resolves.toBeDefined() - - await expect( - adapter.dispatch({ - sessionId: 'session-1', - clientMessageId: 'client-unverified', - body: USER_MESSAGE, - fence: 7 - }) - // `admitted`, not `accepted`: a Codex send now settles its identity on - // the provider echo. What this test pins is the tier the turn carries. - ).resolves.toMatchObject({ state: 'admitted' }) + it('records a background failure without delaying a send and retries after the failure TTL', async () => { + let now = 1_000 + const clock = vi.spyOn(Date, 'now').mockImplementation(() => now) + const modelCatalog = new AgentModelCatalogStore({ now: () => now }) + const listModels: Mock = vi + .fn() + .mockImplementationOnce(() => { + throw new Error('catalog unavailable') + }) + .mockImplementation(() => listing()) + const codex = fakeCodex({ 'model/list': listModels }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-standard') + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { modelCatalog }) + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + try { + await acquire(adapter) + await vi.waitFor(() => expect(modelCatalog.hasActiveFailure(access.fingerprint)).toBe(true)) + await send(adapter, 'first') expect( codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params ).toMatchObject({ serviceTier: 'default' }) + await acquire(adapter, 'session-2') + expect(listModels).toHaveBeenCalledOnce() + now += AGENT_MODEL_CATALOG_FAILURE_TTL_MS + await acquire(adapter, 'session-3') + await vi.waitFor(() => expect(listModels).toHaveBeenCalledTimes(2)) + } finally { + clock.mockRestore() + } + }) - let options = await adapter.readOptions({ sessionId: 'session-1', fence: 7 }) - expect(options).toMatchObject({ - current: { fastMode: true } - }) - if (discovery === 'absent') { - expect(options.fastModeSupport).toBeUndefined() - expect(options.models[0]?.supportsFastMode).toBeUndefined() - options = await adapter.readOptions({ sessionId: 'session-1', fence: 7 }) - } - expect(options).toMatchObject({ - models: [expect.objectContaining({ supportsFastMode: true })], - fastModeSupport: { supported: true }, - current: { fastMode: true } - }) - // Ended, so the next send starts a turn rather than steering into this one. - codex.connections[0].handlers.onNotification?.('turn/completed', { - threadId: THREAD_ID, - turn: { id: 'turn-recovered', status: 'completed' } - }) - await adapter.dispatch({ - sessionId: 'session-1', - clientMessageId: 'client-recovered', - body: USER_MESSAGE, - fence: 7 + it('keeps a newer picker success when its own older request later fails', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ + 'model/list': vi + .fn() + .mockImplementationOnce(() => pending.promise) + .mockImplementation(() => listing('priority-picker')) + }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-fast') + const modelCatalog = new AgentModelCatalogStore() + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { modelCatalog }) + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + await acquire(adapter) + await adapter.readOptions({ sessionId: 'session-1', fence: 7 }) + await expect( + adapter.setOption({ sessionId: 'session-1', key: 'model', value: 'gpt-live', fence: 7 }) + ).resolves.toMatchObject({ model: 'gpt-live' }) + expect(codex.connections[0].calls.filter((call) => call.method === 'model/list')).toHaveLength( + 2 + ) + pending.reject(new Error('old request failed')) + await vi.waitFor(() => expect(modelCatalog.failureDetail(access.fingerprint)).toBeNull()) + await send(adapter, 'first') + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'priority-picker' }) + }) + + it('does not apply an obsolete child’s catalog result to its replacement', async () => { + const oldListing = Promise.withResolvers() + const newListing = Promise.withResolvers() + const codex = fakeCodex({ + 'model/list': vi + .fn() + .mockImplementationOnce(() => oldListing.promise) + .mockImplementationOnce(() => newListing.promise) + }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-fast') + const modelCatalog = new AgentModelCatalogStore() + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { modelCatalog }) + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + await acquire(adapter) + await acquire(adapter) + oldListing.resolve(listing('priority-old')) + newListing.resolve(listing('priority-new')) + await vi.waitFor(() => + expect(modelCatalog.get(access.fingerprint)?.fastModeTierByModel['gpt-live']).toBe( + 'priority-new' + ) + ) + await send(adapter, 'first') + expect( + codex.connections.at(-1)?.calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'priority-new' }) + }) + + it('lists on its own connection while a separate store probe is still pending', async () => { + const probe = Promise.withResolvers<{ + models: [{ id: string; label: string; isDefault: boolean; efforts: [] }] + fastModeTierByModel: Map + origin: 'probe' + }>() + const modelCatalog = new AgentModelCatalogStore() + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + const pendingProbe = () => probe.promise + const probing = modelCatalog.refresh(access.fingerprint, 'codex', pendingProbe, pendingProbe) + const codex = fakeCodex({ 'model/list': () => listing('priority-own') }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-fast') + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { modelCatalog }) + + await acquire(adapter) + await vi.waitFor(() => + expect(codex.connections[0].calls.some((call) => call.method === 'model/list')).toBe(true) + ) + await send(adapter, 'first') + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'priority-own' }) + probe.resolve({ + models: [{ id: 'gpt-live', label: 'GPT Live', isDefault: true, efforts: [] }], + fastModeTierByModel: new Map([['gpt-live', 'priority-probe']]), + origin: 'probe' + }) + await probing + }) + + it('migrates a legacy saved tier after its own listing arrives', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'first-turn') + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { + modelCatalog: new AgentModelCatalogStore() + }) + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-legacy', + options: { serviceTier: 'priority-live-v2' } + }) + await send(adapter, 'first') + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'default' }) + finishTurn(codex, 'first-turn') + pending.resolve(listing()) + await vi.waitFor(() => + expect(adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 7 })).toMatchObject({ + fastMode: 'true' }) + ) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'second-turn') + await send(adapter, 'second') + expect( + codex.connections[0].calls.filter((call) => call.method === 'turn/start')[1]?.params + ).toMatchObject({ serviceTier: 'priority-live-v2' }) + }) + + it('sends a tier another writer stored even when its own listing fails', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-fast') + const modelCatalog = new AgentModelCatalogStore() + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { modelCatalog }) + await acquire(adapter) + await vi.waitFor(() => + expect(codex.connections[0].calls.some((call) => call.method === 'model/list')).toBe(true) + ) + modelCatalog.recordSuccess(access.fingerprint, 'codex', { + models: [{ id: 'gpt-live', label: 'GPT Live', isDefault: true, efforts: [] }], + fastModeTierByModel: new Map([['gpt-live', 'priority-probe']]), + origin: 'probe' + }) + pending.reject(new Error('own listing failed')) + await vi.waitFor(() => expect( - codex.connections[0].calls.filter((call) => call.method === 'turn/start')[1]?.params - ).toMatchObject({ serviceTier: 'priority-recovered' }) - } - ) + codex.connections[0].calls.filter((call) => call.method === 'model/list') + ).toHaveLength(1) + ) + await send(adapter, 'first') + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ serviceTier: 'priority-probe' }) + }) + + it('restores saved Fast against the saved model, not the resumed thread model', async () => { + const codex = fakeCodex() + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-fast') + const modelCatalog = new AgentModelCatalogStore() + const access = agentModelCatalogSessionAccess( + modelCatalog, + CODEX_STRUCTURED_AGENT, + '/codex/home' + )! + const medium = [{ value: 'medium', label: 'Medium' }] + modelCatalog.recordSuccess(access.fingerprint, 'codex', { + models: [ + { + id: 'gpt-live', + label: 'Live', + isDefault: true, + efforts: medium, + supportsFastMode: false + }, + { id: 'gpt-next', label: 'Next', isDefault: false, efforts: medium, supportsFastMode: true } + ], + fastModeTierByModel: new Map([['gpt-next', 'priority-next']]), + origin: 'live-session' + }) + const adapter = adapterFor(codex, { codexHome: '/codex/home', resumeThreadId: THREAD_ID }, [], { + modelCatalog + }) + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-resume', + options: { model: 'gpt-next', fastMode: 'true' } + }) + + expect(codex.connections[0].calls.some((call) => call.method === 'thread/resume')).toBe(true) + expect(adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 7 })).toEqual({ + model: 'gpt-next', + fastMode: 'true' + }) + await send(adapter, 'first') + const turn = codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + expect(turn).toMatchObject({ model: 'gpt-next', serviceTier: 'priority-next' }) + expect(turn).not.toHaveProperty('effort') + }) }) diff --git a/src/main/codex/codex-structured-fast-mode.ts b/src/main/codex/codex-structured-fast-mode.ts index 145d3ff2a8f5..fbc692817768 100644 --- a/src/main/codex/codex-structured-fast-mode.ts +++ b/src/main/codex/codex-structured-fast-mode.ts @@ -80,7 +80,6 @@ export function reconcileCodexFastModeOption( modelFastModeSupport: boolean | undefined } ): void { - session.fastModeTierByModel = input.fastModeTierByModel const encoded = session.options.get('fastMode') if (encoded !== undefined && decodeCodexFastMode(session.options) === undefined) { session.options.delete('fastMode') diff --git a/src/main/codex/codex-structured-model-catalog.ts b/src/main/codex/codex-structured-model-catalog.ts index e8164494dfd8..ea47d4d6a1c3 100644 --- a/src/main/codex/codex-structured-model-catalog.ts +++ b/src/main/codex/codex-structured-model-catalog.ts @@ -96,7 +96,19 @@ export type CodexModelCatalogListing = { export async function fetchCodexModelCatalogListing(input: { connection: Pick timeoutMs?: number + deadlineMs?: number }): Promise { + const deadline = input.deadlineMs === undefined ? null : Date.now() + input.deadlineMs + const remainingTimeout = (): number | undefined => { + if (deadline === null) { + return input.timeoutMs + } + const remaining = deadline - Date.now() + if (remaining <= 0) { + throw new Error('codex model listing deadline exceeded') + } + return Math.min(remaining, input.timeoutMs ?? remaining) + } const parsedModels: ParsedCodexModelOption[] = [] let cursor: string | null = null for (let page = 0; page < MAX_MODEL_PAGES; page += 1) { @@ -104,7 +116,7 @@ export async function fetchCodexModelCatalogListing(input: { await input.connection.request( 'model/list', { limit: MODEL_PAGE_LIMIT, includeHidden: false, ...(cursor ? { cursor } : {}) }, - { timeoutMs: input.timeoutMs } + { timeoutMs: remainingTimeout() } ) ) const rows = Array.isArray(response?.data) ? response.data : [] @@ -119,7 +131,7 @@ export async function fetchCodexModelCatalogListing(input: { break } } - const configured = await readCodexConfiguredLaunchDefaults(input.connection) + const configured = await readCodexConfiguredLaunchDefaults(input.connection, remainingTimeout()) return { models: applyCodexConfiguredLaunchDefaults( parsedModels.map((entry) => entry.option), diff --git a/src/main/codex/codex-structured-session-acquire.ts b/src/main/codex/codex-structured-session-acquire.ts index 61856ef5f24e..189a0e9eb5c4 100644 --- a/src/main/codex/codex-structured-session-acquire.ts +++ b/src/main/codex/codex-structured-session-acquire.ts @@ -28,6 +28,7 @@ import { handleCodexSessionExit } from './codex-structured-session-close' import { restoredCodexSessionOptions } from './codex-structured-session-options' +import { startBackgroundCodexCatalogRefresh } from './codex-structured-background-catalog' import { codexAcquireCatalogAccess, codexAcquireFastModeCatalog @@ -38,7 +39,6 @@ import { } from './codex-structured-fast-mode' import { assertCodexConnectionOpen, - CODEX_RECEIPT_TIMED_METHODS, codexSessionLifecycle, mintCodexAcquisitionGeneration, type CodexAcquisitionRegistry, @@ -49,6 +49,7 @@ import { import type { CodexStructuredSessionTeardown } from './codex-structured-session-teardown' import type { CodexStructuredNotificationRetry } from './codex-structured-notification-retry' import type { deliverCodexServerRequest } from './codex-structured-provider-events' +import { codexAcquisitionNotificationHandler } from './codex-structured-acquisition-notification' export async function acquireCodexStructuredSession(input: { input: StructuredAgentSessionAcquireInput @@ -141,27 +142,14 @@ export async function acquireCodexStructuredSession(input: { env: buildCodexStructuredChildEnvironment(launch, acquireInput.spawnToken, sessionId) }, { - onNotification: (method, params) => { - // Stamped at receipt, ahead of any pre-publication buffering or retry. - const observedAt = CODEX_RECEIPT_TIMED_METHODS.has(method) - ? (deps.now?.() ?? Date.now()) - : undefined - const dispatchSequenceAtReceipt = - method === 'turn/started' ? dispatchEchoes.latestSequence() : undefined - input.deliver( - acquisition, - sessionId, - () => - notificationRetries.handle( - sessionId, - method, - params, - observedAt, - dispatchSequenceAtReceipt - ), - Buffer.byteLength(JSON.stringify(params ?? null), 'utf8') - ) - }, + onNotification: codexAcquisitionNotificationHandler({ + acquisition, + sessionId, + dispatchEchoes, + notificationRetries, + deliver: input.deliver, + now: deps.now ?? Date.now + }), onServerRequest: (request) => input.deliver( acquisition, @@ -235,12 +223,10 @@ export async function acquireCodexStructuredSession(input: { acquisitions.assertCurrent(sessionId, attempt) const options = restoredCodexSessionOptions(acquireInput.options) const catalogAccess = codexAcquireCatalogAccess(deps, launch) - const fastModeCatalog = await codexAcquireFastModeCatalog({ - connection, + const fastModeCatalog = codexAcquireFastModeCatalog({ catalogAccess, opened, - restoreNeedsCatalog: options.get('fastMode') === 'true' || options.has('serviceTier'), - timeoutMs: deps.requestTimeoutMs + restoreNeedsCatalog: options.get('fastMode') === 'true' || options.has('serviceTier') }) acquisitions.assertCurrent(sessionId, attempt) assertCodexConnectionOpen(connection, sessionId) @@ -256,7 +242,6 @@ export async function acquireCodexStructuredSession(input: { prompts: acquisition.prompts, options, reportedOptions: reportedCodexThreadOptions(opened), - fastModeTierByModel: fastModeCatalog?.fastModeTierByModel ?? new Map(), ...(catalogAccess ? { catalogAccess } : {}), dispatchEchoes, translator, @@ -271,7 +256,8 @@ export async function acquireCodexStructuredSession(input: { ...(unbindReadingControl ? { unbindReadingControl } : {}) } if (fastModeCatalog) { - const model = opened.model ?? fastModeCatalog.result.current.model + // The model the next turn sends, as turn/start and the background refresh resolve it. + const model = options.get('model') ?? opened.model ?? fastModeCatalog.result.current.model reconcileCodexFastModeOption(session, { fastModeTierByModel: fastModeCatalog.fastModeTierByModel, currentFastMode: true, @@ -284,6 +270,13 @@ export async function acquireCodexStructuredSession(input: { for (const event of acquisition.drain()) { event() } + startBackgroundCodexCatalogRefresh({ + session, + sessionId, + sessions, + timeoutMs: deps.requestTimeoutMs, + logger: deps.logger + }) return acquired } catch (error) { if (sessions.get(sessionId)?.connection !== acquisition.connection) { diff --git a/src/main/codex/codex-structured-session-adapter.test.ts b/src/main/codex/codex-structured-session-adapter.test.ts index c9e3c14cac91..b2154b196cb5 100644 --- a/src/main/codex/codex-structured-session-adapter.test.ts +++ b/src/main/codex/codex-structured-session-adapter.test.ts @@ -1,5 +1,4 @@ import { describe, expect, it, vi } from 'vitest' -import { AgentSessionPromptAnswerRejectedError } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { CodexAppServerRequestError, type openCodexAppServerConnection @@ -21,8 +20,108 @@ import { identityFor } from './codex-structured-session-adapter-fixture' import { codexProviderHandle } from '../../shared/agent-session-provider-handle-encoding' +import { AgentModelCatalogStore } from '../native-chat/agent-model-catalog/agent-model-catalog-store' describe('CodexStructuredSessionAdapter.acquire', () => { + it('keeps a started thread usable when its model listing never answers', async () => { + const codex = fakeCodex({ 'model/list': () => new Promise(() => {}) }) + const adapter = adapterFor(codex) + + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + options: { model: 'gpt-saved', effort: 'low', fastMode: 'true' } + }) + + expect(adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 7 })).toEqual({ + model: 'gpt-saved', + effort: 'low', + fastMode: 'true' + }) + expect(codex.connections[0].calls.map((call) => call.method)).toEqual([ + 'thread/start', + 'model/list' + ]) + }) + + it('persists saved model and effort as the next turn uses them', async () => { + const codex = fakeCodex() + const adapter = adapterFor(codex) + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + options: { model: 'gpt-saved', effort: 'low', personality: 'concise' } + }) + + expect( + adapter.readAcquisitionOptions({ + sessionId: 'session-1', + fence: 7, + priorOptions: { model: 'gpt-saved', effort: 'low', personality: 'concise' } + }) + ).toEqual({ model: 'gpt-saved', effort: 'low', personality: 'concise' }) + }) + + it('keeps a saved next-turn pick when resuming a thread that reports its previous pick', async () => { + const codex = fakeCodex() + codex.routes['turn/start'] = () => ({ turn: { id: 'turn-1' } }) + const adapter = adapterFor(codex, { resumeThreadId: THREAD_ID }) + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + options: { model: 'gpt-next', effort: 'low' } + }) + + const saved = adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 7 }) + expect(saved).toMatchObject({ model: 'gpt-next', effort: 'low' }) + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'next-turn', + body: USER_MESSAGE, + fence: 7 + }) + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ + model: 'gpt-next', + effort: 'low' + }) + await adapter.closeSession('session-1') + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 8, + spawnToken: 'spawn-10', + options: saved + }) + expect(adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 8 })).toMatchObject({ + model: 'gpt-next', + effort: 'low' + }) + }) + + it('leaves a rejected listing as a catalog error after the thread opens', async () => { + const codex = fakeCodex({ + 'model/list': () => { + throw new Error('stub model/list rejected') + } + }) + const adapter = adapterFor(codex) + + await adapter.acquire({ identity: identityFor('session-1'), fence: 7, spawnToken: 'spawn-9' }) + + expect(adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 7 })).toEqual({ + model: 'gpt-live', + effort: 'medium' + }) + await expect(adapter.readOptions({ sessionId: 'session-1', fence: 7 })).rejects.toThrow( + 'stub model/list rejected' + ) + expect(codex.connections[0].closeCount).toBe(0) + }) + it('starts a new thread and reports the process and link the lease will prove', async () => { const codex = fakeCodex() const adapter = adapterFor(codex, { codexHome: '/codex/home' }) @@ -127,7 +226,7 @@ describe('CodexStructuredSessionAdapter.acquire', () => { spawnToken: 'spawn-9' }) - expect(codex.connections[0].calls.map((call) => call.method)).toEqual([ + expect(codex.connections[0].calls.slice(0, 2).map((call) => call.method)).toEqual([ 'thread/resume', 'thread/start' ]) @@ -386,15 +485,17 @@ describe('CodexStructuredSessionAdapter.dispatch', () => { // Identity is not knowable here: a send coalesced into a running turn shares // that turn's id, so the echo settles which message landed where. expect(outcome).toEqual({ state: 'admitted' }) - expect(codex.connections[0].calls[1].params).toEqual({ - threadId: THREAD_ID, - clientUserMessageId: 'client-1', - input: [ - { type: 'text', text: 'ship it' }, - { type: 'localImage', path: '/tmp/shot.png' }, - { type: 'image', url: 'https://example.test/a.png' } - ] - }) + expect(codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params).toEqual( + { + threadId: THREAD_ID, + clientUserMessageId: 'client-1', + input: [ + { type: 'text', text: 'ship it' }, + { type: 'localImage', path: '/tmp/shot.png' }, + { type: 'image', url: 'https://example.test/a.png' } + ] + } + ) }) it('admits a send on a build whose turn/start answers before the turn is named', async () => { @@ -509,7 +610,11 @@ describe('CodexStructuredSessionAdapter.dispatch', () => { }) }) codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-1') - const adapter = await acquired(codex) + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { + modelCatalog: new AgentModelCatalogStore() + }) + await adapter.acquire({ identity: identityFor('session-1'), fence: 7, spawnToken: 'spawn-9' }) + await adapter.readOptions({ sessionId: 'session-1', fence: 7 }) await adapter.setOption({ sessionId: 'session-1', key: 'model', value: 'gpt-5', fence: 7 }) await adapter.setOption({ sessionId: 'session-1', key: 'effort', value: 'high', fence: 7 }) @@ -531,277 +636,3 @@ describe('CodexStructuredSessionAdapter.dispatch', () => { expect(turnStart?.params).not.toHaveProperty('sandboxEscape') }) }) - -describe('CodexStructuredSessionAdapter prompts', () => { - function askApproval(codex: ReturnType): void { - codex.connections[0].handlers.onServerRequest?.({ - id: 11, - method: 'item/commandExecution/requestApproval', - params: { itemId: 'codex-item-1', threadId: THREAD_ID, turnId: 'turn-1' } - }) - } - - it('surfaces an approval request and answers it exactly once', async () => { - const codex = fakeCodex() - const events: CodexStructuredSessionEvent[] = [] - const adapter = await acquired(codex, {}, events) - - askApproval(codex) - adapter.bindPromptItemId('session-1', 'codex:thread-abc:turn-1:3', 'codex-item-1') - await adapter.answerPrompt({ - sessionId: 'session-1', - itemId: 'codex:thread-abc:turn-1:3', - kind: 'approval', - response: { kind: 'option', optionId: 'accept' }, - fence: 7, - commit: async () => undefined - }) - - expect(events.at(-1)).toMatchObject({ type: 'prompt', codexItemId: 'codex-item-1' }) - expect(codex.connections[0].replies).toEqual([{ id: 11, result: { decision: 'accept' } }]) - - await expect( - adapter.answerPrompt({ - sessionId: 'session-1', - itemId: 'codex:thread-abc:turn-1:3', - kind: 'approval', - response: { kind: 'option', optionId: 'decline' }, - fence: 7, - commit: async () => undefined - }) - ).rejects.toThrow('no longer waiting on') - expect(codex.connections[0].replies).toHaveLength(1) - }) - - it('responds with an error when a prompt cannot be admitted to the journal sink', async () => { - const codex = fakeCodex() - const events: CodexStructuredSessionEvent[] = [] - const sink: StructuredAgentSessionEventSink = { - appendItem: vi.fn(), - appendTombstone: vi.fn(), - publish: vi.fn(), - tryAppendItem: vi.fn(() => ({ accepted: false as const, reason: 'closed' as const })) - } - const adapter = adapterFor(codex, {}, events) - await adapter.acquire({ - identity: identityFor('session-1'), - fence: 7, - spawnToken: 'spawn-9', - events: sink - }) - - askApproval(codex) - - expect(events.filter((event) => event.type === 'prompt')).toEqual([]) - expect(codex.connections[0].replies).toEqual([ - { - id: 11, - code: -32001, - message: - 'Orca could not durably record item/commandExecution/requestApproval prompt (closed)' - } - ]) - await expect( - adapter.answerPrompt({ - sessionId: 'session-1', - itemId: 'codex-item-1', - kind: 'approval', - response: { kind: 'option', optionId: 'accept' }, - fence: 7, - commit: async () => undefined - }) - ).rejects.toThrow('no longer waiting on') - }) - - it('force-closes when an unhandled provider frame cannot be admitted', async () => { - const codex = fakeCodex() - const events: CodexStructuredSessionEvent[] = [] - const sink: StructuredAgentSessionEventSink = { - appendItem: vi.fn(), - appendTombstone: vi.fn(), - publish: vi.fn(), - tryAppendItem: vi.fn(() => ({ accepted: false as const, reason: 'backpressure' as const })) - } - const adapter = adapterFor(codex, {}, events) - await adapter.acquire({ - identity: identityFor('session-1'), - fence: 7, - spawnToken: 'spawn-9', - events: sink - }) - - codex.connections[0].handlers.onUnhandledFrame?.('frame:invalid-json', '{') - - await vi.waitFor(() => expect(codex.connections[0].closeCount).toBe(1)) - expect(events.filter((event) => event.type === 'ended')).toMatchObject([ - { cause: 'unexpected-exit', fence: 7, acquisitionGeneration: 'generation-1' } - ]) - }) - - it('force-closes after a responded server request is not durably admitted', async () => { - const codex = fakeCodex() - const events: CodexStructuredSessionEvent[] = [] - const sink: StructuredAgentSessionEventSink = { - appendItem: vi.fn(), - appendTombstone: vi.fn(), - publish: vi.fn(), - tryAppendItem: vi.fn(() => ({ accepted: false as const, reason: 'failed' as const })) - } - const adapter = adapterFor(codex, {}, events) - await adapter.acquire({ - identity: identityFor('session-1'), - fence: 7, - spawnToken: 'spawn-9', - events: sink - }) - - codex.connections[0].handlers.onServerRequest?.({ - id: 17, - method: 'item/permissions/requestApproval', - params: { threadId: THREAD_ID } - }) - - await vi.waitFor(() => expect(codex.connections[0].closeCount).toBe(1)) - expect(events.filter((event) => event.type === 'ended')).toMatchObject([ - { cause: 'unexpected-exit', fence: 7, acquisitionGeneration: 'generation-1' } - ]) - }) - - it('answers each approval a tool item asks for separately', async () => { - const codex = fakeCodex() - const events: CodexStructuredSessionEvent[] = [] - const adapter = await acquired(codex, {}, events) - // A shell bridge re-asks per command under one parent tool item, so only the - // approval id tells the two requests apart. - const ask = (id: number, approvalId: string): void => { - codex.connections[0].handlers.onServerRequest?.({ - id, - method: 'item/commandExecution/requestApproval', - params: { itemId: 'codex-item-1', approvalId, threadId: THREAD_ID, turnId: 'turn-1' } - }) - } - - ask(11, 'approval-a') - ask(12, 'approval-b') - adapter.bindPromptItemId('session-1', 'journal-a', 'approval-a') - adapter.bindPromptItemId('session-1', 'journal-b', 'approval-b') - for (const [itemId, optionId] of [ - ['journal-b', 'decline'], - ['journal-a', 'accept'] - ]) { - await adapter.answerPrompt({ - sessionId: 'session-1', - itemId, - kind: 'approval', - response: { kind: 'option', optionId }, - fence: 7, - commit: async () => undefined - }) - } - - expect(codex.connections[0].replies).toEqual([ - { id: 12, result: { decision: 'decline' } }, - { id: 11, result: { decision: 'accept' } } - ]) - expect(events.map((event) => (event.type === 'prompt' ? event.promptKey : null))).toEqual([ - 'approval-a', - 'approval-b' - ]) - }) - - it('rejects an option id that is not a Codex decision', async () => { - const codex = fakeCodex() - const adapter = await acquired(codex) - - askApproval(codex) - const commit = vi.fn(async () => undefined) - - await expect( - adapter.answerPrompt({ - sessionId: 'session-1', - itemId: 'codex-item-1', - kind: 'approval', - response: { kind: 'option', optionId: 'yolo' }, - fence: 7, - commit - }) - ).rejects.toThrow(AgentSessionPromptAnswerRejectedError) - // Refused before the journal records an answer the agent never receives. - expect(commit).not.toHaveBeenCalled() - expect(codex.connections[0].replies).toEqual([]) - }) - - it('holds a multi-question request until every question is answered', async () => { - const codex = fakeCodex() - const adapter = await acquired(codex) - codex.connections[0].handlers.onServerRequest?.({ - id: 12, - method: 'item/tool/requestUserInput', - params: { - itemId: 'codex-item-2', - threadId: THREAD_ID, - turnId: 'turn-1', - questions: [ - { id: 'q1', question: 'Use this answer?' }, - { id: 'q2', question: 'Use that answer?' } - ] - } - }) - - await adapter.answerPrompt({ - sessionId: 'session-1', - itemId: 'codex-item-2', - kind: 'question', - response: { kind: 'answers', answers: [{ questionId: 'q1', optionIds: [], other: 'yes' }] }, - fence: 7, - commit: async () => undefined - }) - expect(codex.connections[0].replies).toEqual([]) - - await adapter.answerPrompt({ - sessionId: 'session-1', - itemId: 'codex-item-2', - kind: 'question', - response: { kind: 'answers', answers: [{ questionId: 'q2', optionIds: [], other: 'no' }] }, - fence: 7, - commit: async () => undefined - }) - - expect(codex.connections[0].replies).toEqual([ - { id: 12, result: { answers: { q1: { answers: ['yes'] }, q2: { answers: ['no'] } } } } - ]) - }) - - it('declines MCP elicitation and journals the explicit disposition', async () => { - const codex = fakeCodex() - const events: CodexStructuredSessionEvent[] = [] - await acquired(codex, {}, events) - - codex.connections[0].handlers.onServerRequest?.({ - id: 13, - method: 'mcpServer/elicitation/request', - params: { itemId: 'codex-item-3', threadId: THREAD_ID } - }) - - expect(codex.connections[0].replies).toEqual([ - { id: 13, result: { action: 'decline', content: null, _meta: null } } - ]) - expect(events.some((event) => event.type === 'prompt')).toBe(false) - }) - - it('surfaces an answer to a prompt Codex already forgot', async () => { - const codex = fakeCodex() - const adapter = await acquired(codex) - - await expect( - adapter.answerPrompt({ - sessionId: 'session-1', - itemId: 'codex-item-gone', - kind: 'approval', - response: { kind: 'option', optionId: 'accept' }, - fence: 7, - commit: async () => undefined - }) - ).rejects.toThrow('no longer waiting on codex-item-gone') - }) -}) diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 11d9a2ffca31..31d67ef0d88f 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -16,6 +16,7 @@ import { supportsCodexStructuredLocation } from './codex-structured-location-sup import { CodexStructuredSessionTeardown } from './codex-structured-session-teardown' import { applyCodexStructuredSessionOption, + prepareLiveCodexSessionOptions, readLiveCodexSessionOptions } from './codex-structured-session-options' import { @@ -268,17 +269,42 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap if (!isCodexTurnOptionKey(input.key)) { throw new Error(`codex app-server has no thread option named ${input.key}`) } - return applyCodexStructuredSessionOption( - this.session(input.sessionId), - input.key, - input.value, - this.deps.requestTimeoutMs - ) + return applyCodexStructuredSessionOption(this.session(input.sessionId), input.key, input.value) } readOptions = (input: { sessionId: string; fence: number }) => readLiveCodexSessionOptions(this.session(input.sessionId), this.deps.requestTimeoutMs) + prepareReadOptions = (input: { sessionId: string; fence: number }) => + prepareLiveCodexSessionOptions(this.session(input.sessionId), this.deps.requestTimeoutMs) + + readAcquisitionOptions = (input: { + sessionId: string + fence: number + priorOptions?: Readonly> + }) => { + const session = this.session(input.sessionId) + const options = { + ...Object.fromEntries( + Object.entries(input.priorOptions ?? {}).filter(([key]) => !isCodexTurnOptionKey(key)) + ), + ...Object.fromEntries(session.options) + } + const reported = session.reportedOptions + // The thread's effort belongs to the thread's model, not to a different saved one. + if ( + options.effort === undefined && + reported.effort && + (options.model === undefined || options.model === reported.model) + ) { + options.effort = reported.effort + } + if (options.model === undefined && reported.model) { + options.model = reported.model + } + return Object.keys(options).length > 0 ? options : undefined + } + closeSession = (sessionId: string): Promise => this.teardown.close(sessionId) forceCloseSession = (sessionId: string): Promise => this.teardown.forceClose(sessionId) disposeSession = (sessionId: string): Promise => this.teardown.close(sessionId) diff --git a/src/main/codex/codex-structured-session-close.test.ts b/src/main/codex/codex-structured-session-close.test.ts index 6d0d9ad74195..e0004a0f7373 100644 --- a/src/main/codex/codex-structured-session-close.test.ts +++ b/src/main/codex/codex-structured-session-close.test.ts @@ -104,7 +104,6 @@ describe('Codex structured session close lifecycle', () => { prompts, options: new Map(), reportedOptions: {}, - fastModeTierByModel: new Map(), dispatchEchoes: createCodexDispatchEchoes(), turnOpenWaits: createCodexTurnOpenWaits(), translator diff --git a/src/main/codex/codex-structured-session-options-catalog.test.ts b/src/main/codex/codex-structured-session-options-catalog.test.ts index b51b4e44d5b8..2bd83ae8b1da 100644 --- a/src/main/codex/codex-structured-session-options-catalog.test.ts +++ b/src/main/codex/codex-structured-session-options-catalog.test.ts @@ -7,11 +7,11 @@ import { applyCodexStructuredSessionOption, readLiveCodexSessionOptions } from './codex-structured-session-options' +import { fetchCodexModelCatalogListing } from './codex-structured-model-catalog' import { CodexBackgroundTaskTracker } from './codex-background-task-tracker' import type { CodexSession } from './codex-structured-session-state' import { AGENT_MODEL_CATALOG_FRESH_MS, - AGENT_MODEL_CATALOG_VALIDATION_MIN_AGE_MS, AgentModelCatalogStore, type AgentModelCatalogProbe } from '../native-chat/agent-model-catalog/agent-model-catalog-store' @@ -62,7 +62,6 @@ function storeSession( prompts: new CodexAcquisitionWindow().prompts, options: new Map(), reportedOptions: { model: 'gpt-live', effort: 'high' }, - fastModeTierByModel: new Map(), dispatchEchoes: createCodexDispatchEchoes(), turnOpenWaits: createCodexTurnOpenWaits(), translator: null, @@ -227,14 +226,26 @@ describe('Codex session options through the host catalog store', () => { expect(modelListCalls(request)).toBe(1) }) - it('waits for one refresh when the picked model is missing from a stale entry', async () => { + it('accepts a new model after the picker refreshes a stale entry', async () => { let at = 1_000 const store = new AgentModelCatalogStore({ now: () => at }) seedEntry(store, 'gpt-live') - at += AGENT_MODEL_CATALOG_VALIDATION_MIN_AGE_MS + at += AGENT_MODEL_CATALOG_FRESH_MS const request = vi.fn(async () => listAnswer('gpt-live', 'gpt-next')) const session = storeSession(request, store) - const committed = await applyCodexStructuredSessionOption(session, 'model', 'gpt-next', 5_000) + expect((await readLiveCodexSessionOptions(session, undefined)).models.map((m) => m.id)).toEqual( + ['gpt-live'] + ) + await vi.waitFor(() => + expect(store.get(FINGERPRINT)!.models.map((model) => model.id)).toEqual([ + 'gpt-live', + 'gpt-next' + ]) + ) + expect((await readLiveCodexSessionOptions(session, undefined)).models.map((m) => m.id)).toEqual( + ['gpt-live', 'gpt-next'] + ) + const committed = await applyCodexStructuredSessionOption(session, 'model', 'gpt-next') expect(committed.model).toBe('gpt-next') expect(modelListCalls(request)).toBe(1) }) @@ -244,26 +255,58 @@ describe('Codex session options through the host catalog store', () => { seedEntry(store, 'gpt-live') const request = vi.fn(async () => listAnswer('gpt-live', 'gpt-next')) const session = storeSession(request, store) - await expect( - applyCodexStructuredSessionOption(session, 'model', 'gpt-next', 5_000) - ).rejects.toThrow(/does not offer model gpt-next/) + await expect(applyCodexStructuredSessionOption(session, 'model', 'gpt-next')).rejects.toThrow( + /does not offer model gpt-next/ + ) expect(request).not.toHaveBeenCalled() }) - it('falls back to the stored entry when the validation refresh fails', async () => { + it('keeps the stored entry when a picker refresh fails', async () => { let at = 1_000 const store = new AgentModelCatalogStore({ now: () => at }) seedEntry(store, 'gpt-live') - at += AGENT_MODEL_CATALOG_VALIDATION_MIN_AGE_MS + at += AGENT_MODEL_CATALOG_FRESH_MS const request = vi.fn(async () => { throw new Error('provider gone') }) const session = storeSession(request, store) - await expect( - applyCodexStructuredSessionOption(session, 'model', 'gpt-next', 5_000) - ).rejects.toThrow(/does not offer model gpt-next/) + await readLiveCodexSessionOptions(session, undefined) + await expect(applyCodexStructuredSessionOption(session, 'model', 'gpt-next')).rejects.toThrow( + /does not offer model gpt-next/ + ) expect(modelListCalls(request)).toBe(1) // The failed refresh never displaced the last good listing. expect(store.get(FINGERPRINT)!.models.map((model) => model.id)).toEqual(['gpt-live']) }) }) + +describe('first-turn catalog deadline', () => { + it('budgets all pages and config/read against one 30-second deadline', async () => { + let now = 1_000 + const clock = vi.spyOn(Date, 'now').mockImplementation(() => now) + const timeouts: number[] = [] + const request = vi.fn( + async ( + method: string, + _params?: Record, + options?: { timeoutMs?: number } + ) => { + timeouts.push(options?.timeoutMs ?? 0) + if (method === 'model/list') { + now += 15_001 + return { data: [modelRow('gpt-live')], nextCursor: timeouts.length === 1 ? 'more' : null } + } + return { config: {} } + } + ) + try { + await expect( + fetchCodexModelCatalogListing({ connection: { request }, deadlineMs: 30_000 }) + ).rejects.toThrow('deadline exceeded') + expect(request.mock.calls.map(([method]) => method)).toEqual(['model/list', 'model/list']) + expect(timeouts).toEqual([30_000, 14_999]) + } finally { + clock.mockRestore() + } + }) +}) diff --git a/src/main/codex/codex-structured-session-options.test.ts b/src/main/codex/codex-structured-session-options.test.ts index fe30a2f42f9a..5176744195b9 100644 --- a/src/main/codex/codex-structured-session-options.test.ts +++ b/src/main/codex/codex-structured-session-options.test.ts @@ -13,6 +13,17 @@ import { reportedCodexThreadOptions } from './codex-structured-fast-mode' import { CodexBackgroundTaskTracker } from './codex-background-task-tracker' import type { CodexSession } from './codex-structured-session-state' import { startCodexTurn } from './codex-structured-turn-start' +import { + USER_MESSAGE, + adapterFor, + answerWithOpenedTurn, + fakeCodex, + identityFor +} from './codex-structured-session-adapter-fixture' +import { + AGENT_MODEL_CATALOG_FRESH_MS, + AgentModelCatalogStore +} from '../native-chat/agent-model-catalog/agent-model-catalog-store' function optionSession(request: CodexAppServerConnection['request']): CodexSession { return { @@ -33,14 +44,63 @@ function optionSession(request: CodexAppServerConnection['request']): CodexSessi prompts: new CodexAcquisitionWindow().prompts, options: new Map(), reportedOptions: { model: 'gpt-live', effort: 'high' }, - fastModeTierByModel: new Map(), dispatchEchoes: createCodexDispatchEchoes(), turnOpenWaits: createCodexTurnOpenWaits(), translator: null } } +async function primePicker(session: CodexSession): Promise { + session.catalogAccess = { + store: new AgentModelCatalogStore(), + fingerprint: 'codex-test-account', + accountHomePath: '/test-account' + } + await readLiveCodexSessionOptions(session, undefined) +} + describe('structured Codex session options', () => { + it.each([ + ['model', 'gpt-next'], + ['effort', 'high'], + ['fastMode', 'true'] + ])('keeps a cold %s pick as intent without starting a model request', async (key, value) => { + const request = vi.fn(async () => ({ data: [] })) + const session = optionSession(request) + session.options.set('serviceTier', 'priority-old') + + await expect(applyCodexStructuredSessionOption(session, key, value)).resolves.toMatchObject({ + [key]: value + }) + expect(session.options.has('serviceTier')).toBe(key !== 'fastMode') + expect(request).not.toHaveBeenCalled() + }) + + it('drops an effort saved under another model on a cold model change', async () => { + const session = optionSession(vi.fn(async () => ({ data: [] }))) + session.options.set('model', 'gpt-live') + session.options.set('effort', 'xhigh') + + await expect(applyCodexStructuredSessionOption(session, 'model', 'gpt-live')).resolves.toEqual({ + model: 'gpt-live', + effort: 'xhigh' + }) + await expect(applyCodexStructuredSessionOption(session, 'model', 'gpt-next')).resolves.toEqual({ + model: 'gpt-next' + }) + }) + + it('turns Fast off immediately even without a catalog', async () => { + const request = vi.fn(async () => ({ data: [] })) + const session = optionSession(request) + session.options.set('serviceTier', 'priority-old') + + await expect(applyCodexStructuredSessionOption(session, 'fastMode', 'false')).resolves.toEqual({ + fastMode: 'false' + }) + expect(request).not.toHaveBeenCalled() + }) + it('filters restored records to recognized turn options', () => { expect( Object.fromEntries( @@ -160,9 +220,11 @@ describe('structured Codex session options', () => { })) ) - await expect( - applyCodexStructuredSessionOption(session, 'model', 'gpt-fast', undefined) - ).resolves.toEqual({ model: 'gpt-fast', effort: 'low' }) + await primePicker(session) + await expect(applyCodexStructuredSessionOption(session, 'model', 'gpt-fast')).resolves.toEqual({ + model: 'gpt-fast', + effort: 'low' + }) }) it('rejects values absent from the provider catalog', async () => { @@ -173,12 +235,13 @@ describe('structured Codex session options', () => { })) ) + await primePicker(session) await expect( - applyCodexStructuredSessionOption(session, 'model', 'not-entitled', undefined) + applyCodexStructuredSessionOption(session, 'model', 'not-entitled') ).rejects.toThrow('does not offer model not-entitled') - await expect( - applyCodexStructuredSessionOption(session, 'effort', 'high', undefined) - ).rejects.toThrow('does not support high') + await expect(applyCodexStructuredSessionOption(session, 'effort', 'high')).rejects.toThrow( + 'does not support high' + ) }) it('maps canonical Fast on and off to the exact advertised tier and Standard', async () => { @@ -201,9 +264,10 @@ describe('structured Codex session options', () => { : { turn: { id: `turn-${requests.length}` } } }) const session = optionSession(request) + await primePicker(session) await expect( - applyCodexStructuredSessionOption(session, 'fastMode', 'true', undefined) + applyCodexStructuredSessionOption(session, 'fastMode', 'true') ).resolves.toMatchObject({ fastMode: 'true' }) await startCodexTurn(session, { clientMessageId: 'message-on', @@ -213,7 +277,7 @@ describe('structured Codex session options', () => { serviceTier: 'rush-v7' }) - await applyCodexStructuredSessionOption(session, 'fastMode', 'false', undefined) + await applyCodexStructuredSessionOption(session, 'fastMode', 'false') await startCodexTurn(session, { clientMessageId: 'message-off', body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'off' }] } @@ -268,18 +332,14 @@ describe('structured Codex session options', () => { })) ) - await expect( - readCodexStructuredSessionOptions({ - connection: session.connection, - current: { model: 'gpt-live' } - }) - ).resolves.toMatchObject({ + await primePicker(session) + await expect(readLiveCodexSessionOptions(session, undefined)).resolves.toMatchObject({ models: [expect.objectContaining({ supportsFastMode: false })], fastModeSupport: { supported: false } }) - await expect( - applyCodexStructuredSessionOption(session, 'fastMode', 'true', undefined) - ).rejects.toThrow('does not support Fast mode') + await expect(applyCodexStructuredSessionOption(session, 'fastMode', 'true')).rejects.toThrow( + 'does not support Fast mode' + ) }) it('reconciles restored Fast on to explicit Standard when the selected model lost support', async () => { @@ -339,6 +399,9 @@ describe('structured Codex session options', () => { } }) const session = optionSession(request) + let now = 1_000 + const store = new AgentModelCatalogStore({ now: () => now }) + session.catalogAccess = { store, fingerprint: 'codex-test-account', accountHomePath: '/test' } session.options.set('fastMode', 'true') const unknown = await readLiveCodexSessionOptions(session, undefined) @@ -358,6 +421,14 @@ describe('structured Codex session options', () => { expect(session.options.get('fastMode')).toBe('true') catalogRecovered = true + now += AGENT_MODEL_CATALOG_FRESH_MS + // Served from the aged entry; the refresh behind it names the tier. + await readLiveCodexSessionOptions(session, undefined) + await vi.waitFor(() => + expect(store.get('codex-test-account')?.fastModeTierByModel['gpt-live']).toBe( + 'priority-recovered' + ) + ) await expect(readLiveCodexSessionOptions(session, undefined)).resolves.toMatchObject({ models: [expect.objectContaining({ supportsFastMode: true })], fastModeSupport: { supported: true }, @@ -387,7 +458,7 @@ describe('structured Codex session options', () => { ) await expect( - applyCodexStructuredSessionOption(session, 'fastMode', 'false', undefined) + applyCodexStructuredSessionOption(session, 'fastMode', 'false') ).resolves.toMatchObject({ fastMode: 'false' }) await startCodexTurn(session, { clientMessageId: 'message-standard', @@ -459,9 +530,10 @@ describe('structured Codex session options', () => { })) ) session.options.set('fastMode', 'true') + await primePicker(session) await expect( - applyCodexStructuredSessionOption(session, 'model', 'gpt-standard', undefined) + applyCodexStructuredSessionOption(session, 'model', 'gpt-standard') ).resolves.toMatchObject({ model: 'gpt-standard', fastMode: 'false' }) }) }) @@ -474,7 +546,7 @@ describe('Codex service tier is not a settable option', () => { const session = optionSession(async () => ({ data: [] })) await expect( - applyCodexStructuredSessionOption(session, 'serviceTier', 'priority', undefined) + applyCodexStructuredSessionOption(session, 'serviceTier', 'priority') ).rejects.toThrow('cannot be set directly') expect(session.options.has('serviceTier')).toBe(false) @@ -483,3 +555,46 @@ describe('Codex service tier is not a settable option', () => { }) }) }) + +describe('Codex option picks before the model list arrives', () => { + it('accepts launch-held model, effort and Fast picks while its own listing is held', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-1') + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { + modelCatalog: new AgentModelCatalogStore() + }) + await adapter.acquire({ identity: identityFor('session-1'), fence: 7, spawnToken: 'spawn-1' }) + await vi.waitFor(() => + expect(codex.connections[0].calls.some((call) => call.method === 'model/list')).toBe(true) + ) + const pick = (key: string, value: string) => + adapter.setOption({ sessionId: 'session-1', key, value, fence: 7 }) + + await expect(pick('model', 'gpt-next')).resolves.toMatchObject({ model: 'gpt-next' }) + await expect(pick('effort', 'low')).resolves.toMatchObject({ model: 'gpt-next', effort: 'low' }) + await expect(pick('fastMode', 'true')).resolves.toEqual({ + model: 'gpt-next', + effort: 'low', + fastMode: 'true' + }) + expect(adapter.readAcquisitionOptions({ sessionId: 'session-1', fence: 7 })).toEqual({ + model: 'gpt-next', + effort: 'low', + fastMode: 'true' + }) + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'first', + body: USER_MESSAGE, + fence: 7 + }) + expect( + codex.connections[0].calls.find((call) => call.method === 'turn/start')?.params + ).toMatchObject({ model: 'gpt-next', effort: 'low', serviceTier: 'default' }) + expect(codex.connections[0].calls.filter((call) => call.method === 'model/list')).toHaveLength( + 1 + ) + pending.resolve({ data: [], nextCursor: null }) + }) +}) diff --git a/src/main/codex/codex-structured-session-options.ts b/src/main/codex/codex-structured-session-options.ts index 3002be7b673b..0a21d17a2e72 100644 --- a/src/main/codex/codex-structured-session-options.ts +++ b/src/main/codex/codex-structured-session-options.ts @@ -1,6 +1,6 @@ import type { AgentSessionOptionsResult } from '../../shared/agent-session-wire' import type { CodexAppServerConnection } from './codex-app-server-connection' -import type { CodexSession, CodexSessionCatalogAccess } from './codex-structured-session-state' +import type { CodexSession } from './codex-structured-session-state' import { isCodexTurnOptionKey } from './codex-structured-turn-start' import { AgentSessionOptionRejectedError } from '../native-chat/agent-session-wire/structured-agent-session-option-error' import { decodeStructuredAgentSessionOptionValue } from '../../shared/structured-agent-session-option-codec' @@ -12,7 +12,8 @@ import { type CodexModelCatalogListing, type CodexSessionOptionCatalog } from './codex-structured-model-catalog' -import type { AgentModelCatalogEntry } from '../native-chat/agent-model-catalog/agent-model-catalog-store' +import { codexAcquireCatalogListing, listingFromEntry } from './codex-structured-catalog-entry' +export { codexAcquireCatalogListing } from './codex-structured-catalog-entry' export function restoredCodexSessionOptions( options: Readonly> | undefined @@ -37,13 +38,6 @@ export type { CodexSessionOptionCatalog } from './codex-structured-model-catalog export { readCodexStructuredSessionOptionCatalog } from './codex-structured-model-catalog' -function listingFromEntry(entry: AgentModelCatalogEntry): CodexModelCatalogListing { - return { - models: entry.models.map((model) => ({ ...model })), - fastModeTierByModel: new Map(Object.entries(entry.fastModeTierByModel)) - } -} - async function fetchCodexListingThroughStore( session: CodexSession, timeoutMs: number | undefined @@ -90,63 +84,6 @@ export async function codexSessionCatalogListingForPicker( return fetchCodexListingThroughStore(session, timeoutMs) } -/** - * The listing an option write validates against. A stored entry that already - * names the required model answers outright; one old enough to have missed a - * newly granted model waits for one bounded refresh before a refusal — and a - * refresh that fails falls back to the stored entry rather than refusing to - * answer at all. - */ -async function codexSessionCatalogListingForValidation( - session: CodexSession, - timeoutMs: number | undefined, - requiredModel: string | null -): Promise { - const access = session.catalogAccess - const entry = access?.store.get(access.fingerprint) - if (!access || !entry) { - return fetchCodexListingThroughStore(session, timeoutMs) - } - const hasRequired = - requiredModel === null || entry.models.some((model) => model.id === requiredModel) - const youngEnough = - access.store.withinValidationMinAge(entry) || access.store.hasActiveFailure(access.fingerprint) - if (hasRequired || youngEnough) { - return listingFromEntry(entry) - } - try { - return await fetchCodexListingThroughStore(session, timeoutMs) - } catch { - return listingFromEntry(entry) - } -} - -/** Get-or-fetch for acquire time, before the session object exists. Null on a - * failed fetch — fast-mode restore degrades exactly as a failed listing did. */ -export async function codexAcquireCatalogListing( - connection: Pick, - catalogAccess: CodexSessionCatalogAccess | undefined, - timeoutMs: number | undefined -): Promise { - const entry = catalogAccess?.store.get(catalogAccess.fingerprint) - if (entry) { - return listingFromEntry(entry) - } - try { - const listing = await fetchCodexModelCatalogListing({ connection, timeoutMs }) - if (catalogAccess && listing.models.length > 0) { - catalogAccess.store.recordSuccess(catalogAccess.fingerprint, 'codex', { - models: listing.models, - fastModeTierByModel: listing.fastModeTierByModel, - origin: 'live-session' - }) - } - return listing - } catch { - return null - } -} - export async function readCodexStructuredSessionOptions(input: { connection: Pick current: { model?: string; effort?: string; fastMode?: boolean } @@ -180,11 +117,10 @@ function composeLiveCodexCatalog( }) } -export async function readLiveCodexSessionOptions( +function applyLiveCodexCatalog( session: CodexSession, - timeoutMs: number | undefined -): Promise { - const listing = await codexSessionCatalogListingForPicker(session, timeoutMs) + listing: CodexModelCatalogListing +): AgentSessionOptionsResult { const catalog = composeLiveCodexCatalog(session, listing) reconcileCodexFastModeOption(session, { fastModeTierByModel: catalog.fastModeTierByModel, @@ -200,25 +136,63 @@ export async function readLiveCodexSessionOptions( : { ...catalog.result, current: { ...catalog.result.current, fastMode } } } +export async function readLiveCodexSessionOptions( + session: CodexSession, + timeoutMs: number | undefined +): Promise { + const listing = await codexSessionCatalogListingForPicker(session, timeoutMs) + return applyLiveCodexCatalog(session, listing) +} + +/** Fetch outside the session lane; apply against current options inside it. */ +export async function prepareLiveCodexSessionOptions( + session: CodexSession, + timeoutMs: number | undefined +): Promise<() => AgentSessionOptionsResult> { + const listing = await codexSessionCatalogListingForPicker(session, timeoutMs) + return () => applyLiveCodexCatalog(session, listing) +} + export async function applyCodexStructuredSessionOption( session: CodexSession, key: string, - value: string, - timeoutMs: number | undefined + value: string ): Promise>> { try { - return await applyValidatedCodexStructuredSessionOption(session, key, value, timeoutMs) + return applyValidatedCodexStructuredSessionOption(session, key, value) } catch (error) { throw new AgentSessionOptionRejectedError(error) } } -async function applyValidatedCodexStructuredSessionOption( +/** + * With no listing known, a pick is kept as the next turn's intent rather than refused: catalog + * bookkeeping must not gate it. Codex judges the model on `turn/start`, and Fast without a + * known exact tier sends Standard there. + */ +function applyUnlistedCodexSessionOption( + session: CodexSession, + key: 'model' | 'effort' | 'fastMode', + value: string +): Readonly> { + if (key === 'fastMode') { + session.options.delete('serviceTier') + } else if ( + key === 'model' && + value !== (session.options.get('model') ?? session.reportedOptions.model) + ) { + // An effort saved under another model is unverified for this one; its default applies. + session.options.delete('effort') + } + session.options.set(key, value) + return Object.fromEntries(session.options) +} + +function applyValidatedCodexStructuredSessionOption( session: CodexSession, key: string, - value: string, - timeoutMs: number | undefined -): Promise>> { + value: string +): Readonly> { // `serviceTier` still restores, so a session persisted before Fast existed migrates, // but the turn now derives the tier from `fastMode`. Accepting a direct write would // report success for a value the next turn discards. @@ -229,13 +203,22 @@ async function applyValidatedCodexStructuredSessionOption( session.options.set(key, value) return Object.fromEntries(session.options) } + const requestedFastMode = + key === 'fastMode' ? decodeStructuredAgentSessionOptionValue('fastMode', value) : undefined + if (key === 'fastMode' && typeof requestedFastMode !== 'boolean') { + throw new Error('codex fast mode must be encoded as true or false') + } + if (requestedFastMode === false) { + session.options.delete('serviceTier') + session.options.set('fastMode', 'false') + return Object.fromEntries(session.options) + } + const listing = codexAcquireCatalogListing(session.catalogAccess) + if (!listing) { + return applyUnlistedCodexSessionOption(session, key, value) + } const priorModel = session.options.get('model') ?? session.reportedOptions.model const priorEffort = session.options.get('effort') ?? session.reportedOptions.effort - const listing = await codexSessionCatalogListingForValidation( - session, - timeoutMs, - key === 'model' ? value : (priorModel ?? null) - ) const catalog = composeCodexSessionOptionCatalog(listing, { current: { ...(priorModel ? { model: priorModel } : {}), @@ -256,12 +239,8 @@ async function applyValidatedCodexStructuredSessionOption( const modelId = key === 'model' ? value : catalog.result.current.model const model = catalog.result.models.find((entry) => entry.id === modelId) if (key === 'fastMode') { - const requested = decodeStructuredAgentSessionOptionValue('fastMode', value) - if (typeof requested !== 'boolean') { - throw new Error('codex fast mode must be encoded as true or false') - } if ( - requested && + requestedFastMode && (model?.supportsFastMode !== true || !catalog.fastModeTierByModel.has(modelId)) ) { throw new Error(`codex app-server model ${modelId} does not support Fast mode`) diff --git a/src/main/codex/codex-structured-session-prompts.test.ts b/src/main/codex/codex-structured-session-prompts.test.ts new file mode 100644 index 000000000000..4a61f101a7d7 --- /dev/null +++ b/src/main/codex/codex-structured-session-prompts.test.ts @@ -0,0 +1,285 @@ +import { describe, expect, it, vi } from 'vitest' +import { AgentSessionPromptAnswerRejectedError } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter' +import { + THREAD_ID, + acquired, + adapterFor, + fakeCodex, + identityFor +} from './codex-structured-session-adapter-fixture' + +describe('CodexStructuredSessionAdapter prompts', () => { + function askApproval(codex: ReturnType): void { + codex.connections[0].handlers.onServerRequest?.({ + id: 11, + method: 'item/commandExecution/requestApproval', + params: { itemId: 'codex-item-1', threadId: THREAD_ID, turnId: 'turn-1' } + }) + } + + it('surfaces an approval request and answers it exactly once', async () => { + const codex = fakeCodex() + const events: CodexStructuredSessionEvent[] = [] + const adapter = await acquired(codex, {}, events) + + askApproval(codex) + adapter.bindPromptItemId('session-1', 'codex:thread-abc:turn-1:3', 'codex-item-1') + await adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'codex:thread-abc:turn-1:3', + kind: 'approval', + response: { kind: 'option', optionId: 'accept' }, + fence: 7, + commit: async () => undefined + }) + + expect(events.at(-1)).toMatchObject({ type: 'prompt', codexItemId: 'codex-item-1' }) + expect(codex.connections[0].replies).toEqual([{ id: 11, result: { decision: 'accept' } }]) + + await expect( + adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'codex:thread-abc:turn-1:3', + kind: 'approval', + response: { kind: 'option', optionId: 'decline' }, + fence: 7, + commit: async () => undefined + }) + ).rejects.toThrow('no longer waiting on') + expect(codex.connections[0].replies).toHaveLength(1) + }) + + it('responds with an error when a prompt cannot be admitted to the journal sink', async () => { + const codex = fakeCodex() + const events: CodexStructuredSessionEvent[] = [] + const sink: StructuredAgentSessionEventSink = { + appendItem: vi.fn(), + appendTombstone: vi.fn(), + publish: vi.fn(), + tryAppendItem: vi.fn(() => ({ accepted: false as const, reason: 'closed' as const })) + } + const adapter = adapterFor(codex, {}, events) + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + events: sink + }) + + askApproval(codex) + + expect(events.filter((event) => event.type === 'prompt')).toEqual([]) + expect(codex.connections[0].replies).toEqual([ + { + id: 11, + code: -32001, + message: + 'Orca could not durably record item/commandExecution/requestApproval prompt (closed)' + } + ]) + await expect( + adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'codex-item-1', + kind: 'approval', + response: { kind: 'option', optionId: 'accept' }, + fence: 7, + commit: async () => undefined + }) + ).rejects.toThrow('no longer waiting on') + }) + + it('force-closes when an unhandled provider frame cannot be admitted', async () => { + const codex = fakeCodex() + const events: CodexStructuredSessionEvent[] = [] + const sink: StructuredAgentSessionEventSink = { + appendItem: vi.fn(), + appendTombstone: vi.fn(), + publish: vi.fn(), + tryAppendItem: vi.fn(() => ({ accepted: false as const, reason: 'backpressure' as const })) + } + const adapter = adapterFor(codex, {}, events) + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + events: sink + }) + + codex.connections[0].handlers.onUnhandledFrame?.('frame:invalid-json', '{') + + await vi.waitFor(() => expect(codex.connections[0].closeCount).toBe(1)) + expect(events.filter((event) => event.type === 'ended')).toMatchObject([ + { cause: 'unexpected-exit', fence: 7, acquisitionGeneration: 'generation-1' } + ]) + }) + + it('force-closes after a responded server request is not durably admitted', async () => { + const codex = fakeCodex() + const events: CodexStructuredSessionEvent[] = [] + const sink: StructuredAgentSessionEventSink = { + appendItem: vi.fn(), + appendTombstone: vi.fn(), + publish: vi.fn(), + tryAppendItem: vi.fn(() => ({ accepted: false as const, reason: 'failed' as const })) + } + const adapter = adapterFor(codex, {}, events) + await adapter.acquire({ + identity: identityFor('session-1'), + fence: 7, + spawnToken: 'spawn-9', + events: sink + }) + + codex.connections[0].handlers.onServerRequest?.({ + id: 17, + method: 'item/permissions/requestApproval', + params: { threadId: THREAD_ID } + }) + + await vi.waitFor(() => expect(codex.connections[0].closeCount).toBe(1)) + expect(events.filter((event) => event.type === 'ended')).toMatchObject([ + { cause: 'unexpected-exit', fence: 7, acquisitionGeneration: 'generation-1' } + ]) + }) + + it('answers each approval a tool item asks for separately', async () => { + const codex = fakeCodex() + const events: CodexStructuredSessionEvent[] = [] + const adapter = await acquired(codex, {}, events) + // A shell bridge re-asks per command under one parent tool item, so only the + // approval id tells the two requests apart. + const ask = (id: number, approvalId: string): void => { + codex.connections[0].handlers.onServerRequest?.({ + id, + method: 'item/commandExecution/requestApproval', + params: { itemId: 'codex-item-1', approvalId, threadId: THREAD_ID, turnId: 'turn-1' } + }) + } + + ask(11, 'approval-a') + ask(12, 'approval-b') + adapter.bindPromptItemId('session-1', 'journal-a', 'approval-a') + adapter.bindPromptItemId('session-1', 'journal-b', 'approval-b') + for (const [itemId, optionId] of [ + ['journal-b', 'decline'], + ['journal-a', 'accept'] + ]) { + await adapter.answerPrompt({ + sessionId: 'session-1', + itemId, + kind: 'approval', + response: { kind: 'option', optionId }, + fence: 7, + commit: async () => undefined + }) + } + + expect(codex.connections[0].replies).toEqual([ + { id: 12, result: { decision: 'decline' } }, + { id: 11, result: { decision: 'accept' } } + ]) + expect(events.map((event) => (event.type === 'prompt' ? event.promptKey : null))).toEqual([ + 'approval-a', + 'approval-b' + ]) + }) + + it('rejects an option id that is not a Codex decision', async () => { + const codex = fakeCodex() + const adapter = await acquired(codex) + + askApproval(codex) + const commit = vi.fn(async () => undefined) + + await expect( + adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'codex-item-1', + kind: 'approval', + response: { kind: 'option', optionId: 'yolo' }, + fence: 7, + commit + }) + ).rejects.toThrow(AgentSessionPromptAnswerRejectedError) + // Refused before the journal records an answer the agent never receives. + expect(commit).not.toHaveBeenCalled() + expect(codex.connections[0].replies).toEqual([]) + }) + + it('holds a multi-question request until every question is answered', async () => { + const codex = fakeCodex() + const adapter = await acquired(codex) + codex.connections[0].handlers.onServerRequest?.({ + id: 12, + method: 'item/tool/requestUserInput', + params: { + itemId: 'codex-item-2', + threadId: THREAD_ID, + turnId: 'turn-1', + questions: [ + { id: 'q1', question: 'Use this answer?' }, + { id: 'q2', question: 'Use that answer?' } + ] + } + }) + + await adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'codex-item-2', + kind: 'question', + response: { kind: 'answers', answers: [{ questionId: 'q1', optionIds: [], other: 'yes' }] }, + fence: 7, + commit: async () => undefined + }) + expect(codex.connections[0].replies).toEqual([]) + + await adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'codex-item-2', + kind: 'question', + response: { kind: 'answers', answers: [{ questionId: 'q2', optionIds: [], other: 'no' }] }, + fence: 7, + commit: async () => undefined + }) + + expect(codex.connections[0].replies).toEqual([ + { id: 12, result: { answers: { q1: { answers: ['yes'] }, q2: { answers: ['no'] } } } } + ]) + }) + + it('declines MCP elicitation and journals the explicit disposition', async () => { + const codex = fakeCodex() + const events: CodexStructuredSessionEvent[] = [] + await acquired(codex, {}, events) + + codex.connections[0].handlers.onServerRequest?.({ + id: 13, + method: 'mcpServer/elicitation/request', + params: { itemId: 'codex-item-3', threadId: THREAD_ID } + }) + + expect(codex.connections[0].replies).toEqual([ + { id: 13, result: { action: 'decline', content: null, _meta: null } } + ]) + expect(events.some((event) => event.type === 'prompt')).toBe(false) + }) + + it('surfaces an answer to a prompt Codex already forgot', async () => { + const codex = fakeCodex() + const adapter = await acquired(codex) + + await expect( + adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'codex-item-gone', + kind: 'approval', + response: { kind: 'option', optionId: 'accept' }, + fence: 7, + commit: async () => undefined + }) + ).rejects.toThrow('no longer waiting on codex-item-gone') + }) +}) diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index 7669895937d5..7025014d87a0 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -143,8 +143,6 @@ export type CodexSession = { serviceTier?: string | null serviceTierKnown?: true } - /** Exact provider-advertised Fast request value for each discovered model. */ - fastModeTierByModel: Map /** Absent when the adapter runs without a host catalog store (tests). */ catalogAccess?: CodexSessionCatalogAccess /** Sends whose identity is still to be settled by the provider echo. */ diff --git a/src/main/codex/codex-structured-turn-start.ts b/src/main/codex/codex-structured-turn-start.ts index 271def876076..c180eb045e46 100644 --- a/src/main/codex/codex-structured-turn-start.ts +++ b/src/main/codex/codex-structured-turn-start.ts @@ -22,6 +22,8 @@ import { codexTurnEndRejection } from './codex-structured-turn-end-settlement' import { decodeStructuredAgentSessionOptionValue } from '../../shared/structured-agent-session-option-codec' +import { codexKnownFastModeTier } from './codex-structured-catalog-entry' +import type { CodexSessionCatalogAccess } from './codex-structured-session-state' // Writing a Codex turn and learning which message landed where, which are not // the same event. The answer proves admission and nothing about identity, which @@ -55,7 +57,8 @@ export type CodexTurnHost = { threadId: string options: Map reportedOptions?: { model?: string } - fastModeTierByModel: ReadonlyMap + /** The account's catalog, read at send so any writer's newer tier applies. */ + catalogAccess?: CodexSessionCatalogAccess dispatchEchoes: CodexDispatchEchoes activeTurnIds?: ReadonlySet turnOpenWaits: Pick @@ -81,7 +84,7 @@ function codexTurnOptions(host: CodexTurnHost): Record { ) const encodedFastMode = host.options.get('fastMode') if (encodedFastMode === undefined) { - return options + return host.options.has('serviceTier') ? { ...options, serviceTier: 'default' } : options } const fastMode = decodeStructuredAgentSessionOptionValue('fastMode', encodedFastMode) if (typeof fastMode !== 'boolean') { @@ -91,7 +94,7 @@ function codexTurnOptions(host: CodexTurnHost): Record { return { ...options, serviceTier: 'default' } } const model = host.options.get('model') ?? host.reportedOptions?.model - const tierId = model ? host.fastModeTierByModel.get(model) : undefined + const tierId = model ? codexKnownFastModeTier(host.catalogAccess, model) : undefined // Fast is on but nothing has named the tier for this model yet, so there is no // value to route to. Deliberately Standard rather than an omission: the tier // persists on the thread, so omitting would silently keep routing a paid tier we diff --git a/src/main/codex/codex-structured-turn-steer.test.ts b/src/main/codex/codex-structured-turn-steer.test.ts index 15f95aa29275..3e82c80b81c3 100644 --- a/src/main/codex/codex-structured-turn-steer.test.ts +++ b/src/main/codex/codex-structured-turn-steer.test.ts @@ -35,7 +35,10 @@ async function rig(routes: Record) { body: CODEX_TEST_USER_MESSAGE, fence: 7 }) - const methods = () => connection.calls.map(({ method }) => method) + const methods = () => + connection.calls + .map(({ method }) => method) + .filter((method) => method !== 'model/list' && method !== 'config/read') const endTurn = (turnId: string, status: 'completed' | 'interrupted') => connection.handlers.onNotification?.('turn/completed', { threadId: CODEX_TEST_THREAD_ID, @@ -210,7 +213,10 @@ describe('a Codex send made after Codex answered an earlier one, before it opene expect(await rig.send('client-1')).toEqual({ state: 'admitted' }) const sending = rig.send('client-2') expect(await settledWithin(sending)).toBe('held') - const methods = () => rig.codex.connections[0]!.calls.map(({ method }) => method) + const methods = () => + rig.codex.connections[0]!.calls + .map(({ method }) => method) + .filter((method) => method !== 'model/list' && method !== 'config/read') expect(methods()).toEqual(['thread/start', 'turn/start']) return { ...rig, sending, methods } } diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-store.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-store.ts index b16a29bff357..f444ef36474b 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-store.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-store.ts @@ -14,9 +14,6 @@ import type { AgentModelCatalogPersistence } from './agent-model-catalog-persist export const AGENT_MODEL_CATALOG_FRESH_MS = 10 * 60_000 export const AGENT_MODEL_CATALOG_FAILURE_TTL_MS = 30_000 export const AGENT_MODEL_CATALOG_PICKER_WAIT_MS = 30_000 -/** A validation read younger than this trusts the entry even when the picked - * model is missing; older, it waits for one bounded refresh before refusing. */ -export const AGENT_MODEL_CATALOG_VALIDATION_MIN_AGE_MS = 60_000 export const AGENT_MODEL_CATALOG_MAX_ENTRIES = 256 export type AgentModelCatalogEntry = { @@ -128,11 +125,6 @@ export class AgentModelCatalogStore { return this.now() - entry.fetchedAt >= AGENT_MODEL_CATALOG_FRESH_MS } - /** Young enough for a validation read to trust even without the picked model. */ - withinValidationMinAge(entry: AgentModelCatalogEntry): boolean { - return this.now() - entry.fetchedAt < AGENT_MODEL_CATALOG_VALIDATION_MIN_AGE_MS - } - failureDetail(fingerprint: string): string | null { return this.hasActiveFailure(fingerprint) ? (this.failures.get(fingerprint)?.detail ?? null) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts index 18eff805a63b..ffb50f0ef99b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts @@ -389,6 +389,76 @@ describe('structured session acquisition options', () => { expect(store.getRecord(SESSION)?.options).toEqual({ model: 'gpt-standard' }) }) + it('proves an opened Codex owner from effective options without reading its catalog', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-acquisition-known-options-')) + const store = await openTestAgentSessionRecordStore(root) + const sessionAdapter: StructuredAgentSessionAdapter = { + ...adapter({ origin: 'created' }), + readOptions: vi.fn(async () => { + throw new Error('model list unavailable') + }), + readAcquisitionOptions: vi.fn(() => ({ model: 'gpt-reported', effort: 'high' })) + } + + const created = await performAttach({ + agents: NO_STRUCTURED_AGENTS, + logger: createStructuredAgentSessionLogger(), + store, + adapter: sessionAdapter, + openConversation: openTestAttachConversation(openTestJournalHostDatabase(root!)), + authority: { + spawnToken: 'spawn-a', + claimKeyId: 'key-1', + handoffOperationId: CREATE_OPERATION, + probe: { outcome: 'reservation-unused' } + }, + callerKey: 'client-1', + params: attachParams(CREATE_OPERATION, null, { model: 'gpt-saved', effort: 'medium' }), + now: () => NOW, + onAttached: () => {} + }) + + expect(created).toMatchObject({ ok: true }) + expect(sessionAdapter.readOptions).not.toHaveBeenCalled() + expect(store.getRecord(SESSION)?.options).toEqual({ model: 'gpt-reported', effort: 'high' }) + expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('live') + }) + + it('retains saved options when an opened Codex thread reports no effective model', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-acquisition-saved-options-')) + const store = await openTestAgentSessionRecordStore(root) + const sessionAdapter: StructuredAgentSessionAdapter = { + ...adapter({ origin: 'created' }), + readOptions: vi.fn(async () => { + throw new Error('model list unavailable') + }), + readAcquisitionOptions: vi.fn(() => undefined) + } + const options = { model: 'gpt-saved', effort: 'medium' } + + const created = await performAttach({ + agents: NO_STRUCTURED_AGENTS, + logger: createStructuredAgentSessionLogger(), + store, + adapter: sessionAdapter, + openConversation: openTestAttachConversation(openTestJournalHostDatabase(root!)), + authority: { + spawnToken: 'spawn-a', + claimKeyId: 'key-1', + handoffOperationId: CREATE_OPERATION, + probe: { outcome: 'reservation-unused' } + }, + callerKey: 'client-1', + params: attachParams(CREATE_OPERATION, null, options), + now: () => NOW, + onAttached: () => {} + }) + + expect(created).toMatchObject({ ok: true }) + expect(store.getRecord(SESSION)?.options).toEqual(options) + expect(sessionAdapter.readOptions).not.toHaveBeenCalled() + }) + it('releases an acquisition when provider options cannot be read', async () => { root = await mkdtemp(join(tmpdir(), 'orca-acquisition-options-failure-')) const store = await openTestAgentSessionRecordStore(root) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts index 6d6589521df7..af986ae7e442 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts @@ -55,13 +55,19 @@ export async function acquireOwner( const options = providerChildPhase === 'starting' ? undefined - : await withAgentSessionCreatePhase('restore_options', input.recordPhase, () => - readNativeSessionOptions({ - adapter: input.adapter, - sessionId: record.sessionId, - fence, - ...(record.options ? { priorOptions: record.options } : {}) - }) + : await withAgentSessionCreatePhase('restore_options', input.recordPhase, async () => + input.adapter.readAcquisitionOptions + ? input.adapter.readAcquisitionOptions({ + sessionId: record.sessionId, + fence, + ...(record.options ? { priorOptions: record.options } : {}) + }) + : readNativeSessionOptions({ + adapter: input.adapter, + sessionId: record.sessionId, + fence, + ...(record.options ? { priorOptions: record.options } : {}) + }) ) if (record.lease.ownerProcess === null) { await input.store.commitProcessIdentity({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts index 7f8ad5c91067..27c4a323506d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts @@ -63,6 +63,30 @@ describe('StructuredAgentSessionAdapterRouter.releaseAcquisition', () => { }) }) +describe('StructuredAgentSessionAdapterRouter.readAcquisitionOptions', () => { + it('keeps the existing live read for providers without acquisition options', async () => { + const claude = adapterOf(vi.fn(async () => true)) + claude.readOptions = vi.fn(async () => ({ + current: { model: 'reported-model', effort: 'high' }, + models: [] + })) + const router = claudeAndCodexRouter( + { claude, codex: adapterOf(vi.fn(async () => false)) }, + async () => {} + ) + await router.acquire({ identity: claudeIdentity('session-1'), fence: 1, spawnToken: 'spawn-1' }) + + await expect( + router.readAcquisitionOptions({ + sessionId: 'session-1', + fence: 1, + priorOptions: { model: 'saved-model', personality: 'concise' } + }) + ).resolves.toEqual({ model: 'reported-model', effort: 'high', personality: 'concise' }) + expect(claude.readOptions).toHaveBeenCalledOnce() + }) +}) + describe('StructuredAgentSessionAdapterRouter.closeSession', () => { it('retains the owner after an unproven close so a later retry reaches the same adapter', async () => { const claude = adapterOf(vi.fn(async () => true)) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts index e7d7ba2cf4c0..7156d4976063 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts @@ -6,6 +6,7 @@ import type { AgentSessionExecutionLocation } from '../../../shared/agent-session-record' import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { readNativeSessionOptions } from './structured-agent-session-option-restoration' import type { StructuredAgentRegistration, StructuredAgentRegistry @@ -161,6 +162,9 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi awaitStarted = (sessionId: string): Promise => this.liveOwnerOrNull(sessionId)?.awaitStarted?.(sessionId) ?? Promise.resolve() + prepareReadOptions = (input: { sessionId: string; fence: number }) => + this.liveOwnerOrNull(input.sessionId)?.prepareReadOptions?.(input) + readOptions = (input: { sessionId: string; fence: number }) => { const reader = this.owner(input.sessionId).readOptions if (!reader) { @@ -169,6 +173,17 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi return reader(input) } + readAcquisitionOptions = (input: { + sessionId: string + fence: number + priorOptions?: Readonly> + }) => { + const adapter = this.owner(input.sessionId) + return adapter.readAcquisitionOptions + ? adapter.readAcquisitionOptions(input) + : readNativeSessionOptions({ adapter, ...input }) + } + readOptionRestoreFailures = (sessionId: string): readonly string[] => this.owner(sessionId).readOptionRestoreFailures?.(sessionId) ?? [] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index f94a3b134201..bf47e0f068ad 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -362,7 +362,21 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & * closed; at once for any other. A start that did not land resolves with the chat's words for * why. Never rejects. */ awaitStarted?(sessionId: string): Promise + /** Fetch off the lane, then apply the result under the same child's fence. */ + prepareReadOptions?(input: { + sessionId: string + fence: number + }): Promise<() => AgentSessionOptionsResult> | undefined readOptions?(input: { sessionId: string; fence: number }): Promise + /** Effective options already known after acquisition, without discovering picker choices. */ + readAcquisitionOptions?(input: { + sessionId: string + fence: number + priorOptions?: Readonly> + }): + | Promise> | undefined> + | Readonly> + | undefined /** Option keys skipped after a provider rejected their persisted restore value. */ readOptionRestoreFailures?(sessionId: string): readonly string[] /** Provider history for restart reconciliation, bounded to what the provider diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index 52e0d8e46c0c..cb3a1bbe5838 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -29,6 +29,15 @@ import { openTestJournalHostDatabase } from '../agent-session-journal/journal-ho import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support' import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' import { NO_STRUCTURED_AGENTS } from './structured-agent-session-adapter-router-test-support' +import { + THREAD_ID, + adapterFor, + answerWithOpenedTurn, + fakeCodex, + identityFor +} from '../../codex/codex-structured-session-adapter-fixture' +import { AgentModelCatalogStore } from '../agent-model-catalog/agent-model-catalog-store' +import { CodexAppServerRequestError } from '../../codex/codex-app-server-request-error' const CALLER = { callerKey: 'client-1' } @@ -39,6 +48,7 @@ let dispatch: Mock let cancelTurn: Mock let awaitStarted: Mock> let closeSession: Mock> +let setOption: Mock let acknowledgeSessionRelease: Mock< NonNullable > @@ -59,6 +69,7 @@ beforeEach(async () => { cancelTurn = vi.fn(async () => ({ cancelled: true })) awaitStarted = vi.fn(async () => undefined) closeSession = vi.fn(async () => true) + setOption = vi.fn(async () => undefined) acknowledgeSessionRelease = vi.fn() stopEndsSession = false log = recordingStructuredAgentSessionLogger() @@ -95,7 +106,7 @@ beforeEach(async () => { cancelTurn, stopEndsSession: () => stopEndsSession, answerPrompt: vi.fn(async () => undefined), - setOption: vi.fn(async () => undefined) + setOption }, journalDatabase: openTestJournalHostDatabase(root), claimKeyId: 'key-1', @@ -146,6 +157,23 @@ function stop(turnId?: string, clientOperationId = hostTestOperationId()) { }) } +function pickOption(key: string, value: string) { + const fields = { key, value } + return host.setOption(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: 1, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.setOption', + sessionId: SESSION, + fields + }) + }, + ...fields + }) +} + async function submission(id: string): Promise { return (await host.journalSnapshot(SESSION)).submissions.find( (entry) => entry.clientMessageId === id @@ -159,6 +187,123 @@ async function statusRows(): Promise { } describe('a Stop that names no turn', () => { + it('does not let a stalled catalog hold an option pick, steer, Stop, or replacement send', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-1') + codex.routes['turn/steer'] = () => ({ turn: { id: 'turn-1' } }) + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { + modelCatalog: new AgentModelCatalogStore() + }) + await adapter.acquire({ + identity: identityFor(SESSION), + fence: 1, + spawnToken: 'spawn-catalog', + options: { fastMode: 'true' } + }) + dispatch.mockImplementation((input) => adapter.dispatch(input)) + cancelTurn.mockImplementation((input) => adapter.cancelTurn(input)) + setOption.mockImplementation((input) => adapter.setOption(input)) + + const first = send('first send') + await first.result + await eventually(() => + expect(codex.connections[0].calls.some((call) => call.method === 'turn/start')).toBe(true) + ) + expect(await pickOption('model', 'gpt-next')).toMatchObject({ + ok: true, + value: { options: { model: 'gpt-next', fastMode: 'true' } } + }) + expect(codex.connections[0].calls.filter((call) => call.method === 'model/list')).toHaveLength( + 1 + ) + + const followUp = send('follow-up') + await followUp.result + await eventually(() => + expect(codex.connections[0].calls.some((call) => call.method === 'turn/steer')).toBe(true) + ) + expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + codex.connections[0].handlers.onNotification?.('turn/completed', { + threadId: THREAD_ID, + turn: { id: 'turn-1', status: 'interrupted' } + }) + codex.routes['turn/start'] = answerWithOpenedTurn(codex, 'turn-2') + const replacement = send('replacement') + await replacement.result + await eventually(() => + expect( + codex.connections[0].calls.filter((call) => call.method === 'turn/start') + ).toHaveLength(2) + ) + expect( + codex.connections[0].calls.filter((call) => call.method === 'turn/start')[1]?.params + ).toMatchObject({ model: 'gpt-next', serviceTier: 'default' }) + pending.resolve({ data: [], nextCursor: null }) + }) + + it('fails a turn Codex refuses for a model picked before the list, then sends again', async () => { + const pending = Promise.withResolvers() + const codex = fakeCodex({ 'model/list': () => pending.promise }) + codex.routes['turn/start'] = (params) => { + if (params?.model === 'gpt-missing') { + throw new CodexAppServerRequestError( + 'turn/start', + -32602, + 'codex app-server turn/start failed: unknown model gpt-missing', + 'unknown model gpt-missing' + ) + } + return answerWithOpenedTurn(codex, 'turn-ok')(params) + } + const adapter = adapterFor(codex, { codexHome: '/codex/home' }, [], { + modelCatalog: new AgentModelCatalogStore() + }) + await adapter.acquire({ identity: identityFor(SESSION), fence: 1, spawnToken: 'spawn-bad' }) + dispatch.mockImplementation((input) => adapter.dispatch(input)) + setOption.mockImplementation((input) => adapter.setOption(input)) + + expect(await pickOption('model', 'gpt-missing')).toMatchObject({ ok: true }) + const refused = send('first send') + await refused.result + await eventually(async () => + expect(await submission(refused.id)).toMatchObject({ + dispatchState: 'rejected', + reason: expect.stringContaining('unknown model gpt-missing'), + rejection: { kind: 'providerRejected' } + }) + ) + + expect(await pickOption('model', 'gpt-live')).toMatchObject({ ok: true }) + const retried = send('second send') + await retried.result + await eventually(() => + expect( + codex.connections[0].calls.filter((call) => call.method === 'turn/start') + ).toHaveLength(2) + ) + expect( + codex.connections[0].calls.filter((call) => call.method === 'turn/start')[1]?.params + ).toMatchObject({ model: 'gpt-live' }) + await eventually(async () => + expect((await submission(retried.id))?.dispatchState).not.toBe('rejected') + ) + pending.resolve({ data: [], nextCursor: null }) + }) + + it('delivers a newly accepted message after Stop withdraws the prior send', async () => { + const first = send('first send') + await first.result + await eventually(() => expect(dispatch).toHaveBeenCalledOnce()) + + expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + const replacement = send('replacement send') + await replacement.result + + await eventually(() => expect(dispatch).toHaveBeenCalledTimes(2)) + expect(dispatch.mock.calls[1]![0].clientMessageId).toBe(replacement.id) + }) + it('is a valid cancel, and only a plain Stop may omit the turn', () => { const envelope = { sessionId: SESSION, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.test.ts index 5a458e6f2e9f..475f7bf091bd 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.test.ts @@ -9,6 +9,7 @@ import { } from '../agent-model-catalog/agent-model-catalog-store' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' import { readStructuredAgentSessionOptions } from './structured-agent-session-options-read' +import { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue' import { NO_STRUCTURED_AGENTS } from './structured-agent-session-adapter-router-test-support' const SESSION = 'session-1' @@ -53,3 +54,106 @@ describe('options at rest', () => { expect(result.models).toEqual([]) }) }) + +describe('live Codex option reads', () => { + it('leaves the session lane free while a cold listing waits, then applies it under the child fence', async () => { + const listing = Promise.withResolvers() + const queue = new StructuredAgentSessionTaskQueue() + const child = { fence: 7 } + const apply = vi.fn(() => ({ + models: [{ id: 'gpt-live', label: 'GPT Live', efforts: [] }], + current: { model: 'gpt-live' } + })) + const prepareReadOptions = vi.fn(async () => { + await listing.promise + return apply + }) + const live = { + child, + params: { provider: 'codex' }, + journal: { threadGoal: () => null, contextUsage: () => null } + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: this read touches only the declared context fields. + const context = { + deps: { + adapter: { prepareReadOptions }, + agents: NO_STRUCTURED_AGENTS, + store: { getRecord: () => restingRecord() } + }, + serialize: (sessionId: string, task: () => Promise) => + queue.serialize(sessionId, task), + openConversation: async () => live, + conversation: async () => live + } as unknown as StructuredAgentSessionMutationContext + + const reading = readStructuredAgentSessionOptions(context, SESSION) + await vi.waitFor(() => + expect(prepareReadOptions).toHaveBeenCalledWith({ sessionId: SESSION, fence: 7 }) + ) + await expect(queue.serialize(SESSION, async () => 'delivery can proceed')).resolves.toBe( + 'delivery can proceed' + ) + expect(apply).not.toHaveBeenCalled() + listing.resolve() + await expect(reading).resolves.toMatchObject({ current: { model: 'gpt-live' } }) + expect(apply).toHaveBeenCalledOnce() + }) + + it('keeps a listing failure visible to the picker', async () => { + const queue = new StructuredAgentSessionTaskQueue() + const live = { child: { fence: 7 }, params: { provider: 'codex' } } + const failure = new Error('model/list unavailable') + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the failed read ends before touching the remaining context fields. + const context = { + deps: { + adapter: { + prepareReadOptions: async () => { + throw failure + } + }, + agents: NO_STRUCTURED_AGENTS, + store: { getRecord: () => restingRecord() } + }, + serialize: (sessionId: string, task: () => Promise) => + queue.serialize(sessionId, task), + openConversation: async () => live + } as unknown as StructuredAgentSessionMutationContext + + await expect(readStructuredAgentSessionOptions(context, SESSION)).rejects.toBe(failure) + await expect(queue.serialize(SESSION, async () => 'Stop can proceed')).resolves.toBe( + 'Stop can proceed' + ) + }) + + it('does not apply a listing to a replaced child', async () => { + const listing = Promise.withResolvers() + const queue = new StructuredAgentSessionTaskQueue() + const apply = vi.fn(() => ({ models: [], current: { model: 'stale' } })) + const prepareReadOptions = vi.fn(async () => { + await listing.promise + return apply + }) + const live = { child: { fence: 7 }, params: { provider: 'codex' } } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: this read touches only the declared context fields. + const context = { + deps: { + adapter: { prepareReadOptions }, + agents: NO_STRUCTURED_AGENTS, + store: { getRecord: () => restingRecord() } + }, + serialize: (sessionId: string, task: () => Promise) => + queue.serialize(sessionId, task), + openConversation: async () => live, + conversation: async () => live + } as unknown as StructuredAgentSessionMutationContext + + const reading = readStructuredAgentSessionOptions(context, SESSION) + await vi.waitFor(() => expect(prepareReadOptions).toHaveBeenCalledOnce()) + await queue.serialize(SESSION, async () => { + live.child = { fence: 8 } + }) + listing.resolve() + await expect(reading).resolves.toMatchObject({ current: { model: '' } }) + expect(apply).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.ts index 314f436f1d25..48fb03b44c13 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-options-read.ts @@ -111,19 +111,37 @@ export async function readStructuredAgentSessionOptions( sessionId: string ): Promise { const { adapter, agents, store } = context.deps - const live = await context.serialize(sessionId, async () => { + const started = await context.serialize(sessionId, async () => { const session = await context.openConversation(sessionId).catch((error: unknown) => { throw journalOpenReadRefusal(error, context.deps.logger, sessionId) }) const child = session?.child if (!child) { - return null + return { kind: 'rest' as const } + } + const prepared = adapter.prepareReadOptions?.({ sessionId, fence: child.fence }) + if (prepared) { + return { kind: 'prepared' as const, child, prepared } } if (!adapter.readOptions) { throw new Error('structured_agent_session_options_unsupported') } - return adapter.readOptions({ sessionId, fence: child.fence }) + return { + kind: 'live' as const, + options: await adapter.readOptions({ sessionId, fence: child.fence }) + } }) + const live = + started.kind === 'live' + ? started.options + : started.kind === 'prepared' + ? await started.prepared.then((apply) => + context.serialize(sessionId, async () => { + const child = (await context.openConversation(sessionId))?.child + return child === started.child ? apply() : null + }) + ) + : null const options = live ?? (await readStructuredAgentSessionOptionsAtRest(context.deps, sessionId)) // Re-acquired after the reads above: the handle they saw may have closed and reopened since. const session = await context.conversation(sessionId)