From 1f9ba767269b059d786c33aecf4bce39c1cf6b33 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 13 Nov 2026 13:16:13 -0800 Subject: [PATCH] Keep slow model discovery alive beyond picker waits --- .../acp-listing-session-models.test.ts | 325 ++++++++++++++---- .../discovery/acp-listing-session-models.ts | 127 ++++--- .../acp/discovery/acp-model-catalog-probe.ts | 36 +- .../acp-model-catalog-selected-demand.test.ts | 92 +++++ .../preflight/agent-cli-version-probe.test.ts | 14 +- .../preflight/agent-cli-version-probe.ts | 20 +- .../claude-model-catalog-observation.ts | 43 +++ .../claude-model-catalog-probe.test.ts | 69 ++++ .../claude-model-catalog-probe.ts | 93 +++-- .../sdk-transport/claude-control-responses.ts | 11 +- .../app-server/codex-app-server-session.ts | 21 +- .../catalog/codex-model-catalog-probe.test.ts | 219 +++++++++++- .../catalog/codex-model-catalog-probe.ts | 124 ++++--- .../catalog/codex-structured-model-catalog.ts | 10 +- .../codex-configured-launch-defaults.ts | 4 +- .../agent-model-catalog-account-resolution.ts | 46 +++ .../agent-model-catalog-discovery.ts | 13 +- .../agent-model-catalog-entry-lifetime.ts | 52 ++- .../agent-model-catalog-lister.ts | 175 +++++----- .../agent-model-catalog-listing-waiters.ts | 41 ++- .../agent-model-catalog-lookup.ts | 1 + .../agent-model-catalog-probe-runner.ts | 52 ++- .../agent-model-catalog-refresh.ts | 170 +++++---- .../agent-model-catalog-scoped-lister.ts | 48 +++ .../agent-model-catalog-service-contract.ts | 2 + .../agent-model-catalog-service.ts | 14 +- ...agent-model-catalog-shared-listing.test.ts | 259 +++++++++++++- .../agent-model-catalog-shared-setup.test.ts | 93 +++-- .../agent-model-catalog-snapshot.ts | 9 +- .../agent-model-catalog-store.ts | 6 +- .../structured-agent-session-catalog-reads.ts | 2 +- .../options/rpc-model-catalog-probe.test.ts | 86 +++++ .../pi/rpc/options/rpc-model-catalog-probe.ts | 75 ++-- src/main/pi/rpc/options/rpc-model-options.ts | 7 +- ...uctured-agent-session-options-read.test.ts | 19 + .../structured-agent-session-options-read.ts | 11 +- .../agent-launch-picker-acquisition.test.ts | 42 +++ .../agent/agent-launch-picker-acquisition.ts | 2 +- .../AgentLaunchChoiceEditor.test.tsx | 20 ++ .../use-agent-settings-model-catalog.ts | 7 +- ...nt-session-renderer-client-capabilities.ts | 2 + ...ured-agent-session-surface-capabilities.ts | 4 + 42 files changed, 1922 insertions(+), 544 deletions(-) create mode 100644 src/main/acp/discovery/acp-model-catalog-selected-demand.test.ts create mode 100644 src/main/claude/model-catalog/claude-model-catalog-observation.ts create mode 100644 src/main/native-chat/agent-model-catalog/agent-model-catalog-account-resolution.ts create mode 100644 src/main/native-chat/agent-model-catalog/agent-model-catalog-scoped-lister.ts create mode 100644 src/main/pi/rpc/options/rpc-model-catalog-probe.test.ts diff --git a/src/main/acp/discovery/acp-listing-session-models.test.ts b/src/main/acp/discovery/acp-listing-session-models.test.ts index 0287f52c637..fce0fb79310 100644 --- a/src/main/acp/discovery/acp-listing-session-models.test.ts +++ b/src/main/acp/discovery/acp-listing-session-models.test.ts @@ -1,3 +1,9 @@ +import { createStructuredAgentSessionCatalogReads } from '../../native-chat/agent-session-wire/options/structured-agent-session-catalog-reads' +import type { StructuredAgentSessionHostDeps } from '../../native-chat/agent-session-wire/host/structured-agent-session-host-types' +import type { createAgentModelCatalogSuccessWriter } from '../../native-chat/agent-model-catalog/agent-model-catalog-entry-lifetime' +import { createAcpModelCatalogProbe } from './acp-model-catalog-probe' +import { AgentModelCatalogStore } from '../../native-chat/agent-model-catalog/agent-model-catalog-store' +import { createAgentModelCatalogService } from '../../native-chat/agent-model-catalog/agent-model-catalog-service' import { NewSessionResponseSchema } from '../generated/acp-protocol.generated' import recorded from '../fixtures/opencode-v2-start-options.json' import { expect, it, vi, onTestFinished } from 'vitest' @@ -6,75 +12,268 @@ import { readSessionModels, type AcpModelCatalogSessionConnection } from './acp-listing-session-models' +import type { runAgentModelCatalogListing } from '../../native-chat/agent-model-catalog/agent-model-catalog-probe-runner' import { runAgentModelCatalogSession } from '../../native-chat/agent-model-catalog/agent-model-catalog-probe-runner' -it.each([6_400, 14_500])( - 'keeps inventory received at %sms when optional model efforts stall', - async (startupMs) => { - vi.useFakeTimers() - onTestFinished(() => { - vi.useRealTimers() - }) - const spec = acpLaunchSpecFor('opencode2')! - const owner = new AbortController() - const started = - Promise.withResolvers>>() - const late = - Promise.withResolvers< - Awaited> - >() - const connection: AcpModelCatalogSessionConnection = { - initialize: async () => ({ - protocolVersion: 1, - agentCapabilities: { sessionCapabilities: { delete: {} } } - }), - start: () => started.promise, - setConfigOption: () => late.promise, - deleteSession: async () => ({}), - close: async () => {} - } - const run = runAgentModelCatalogSession< - AcpModelCatalogSessionConnection, - Awaited> - >( - () => connection, - (connection, ready) => - readSessionModels( - spec, - { kind: 'session', listingNamesConfiguredModel: false }, - connection, - '/workspace', - { signal: owner.signal }, - undefined, - undefined, - ready - ), - { label: 'opencode2', signal: owner.signal } - ) - await vi.advanceTimersByTimeAsync(startupMs) - started.resolve({ +it('publishes inventory before a slow optional effort reply and retains its valid late result', async () => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const late = + Promise.withResolvers< + Awaited> + >() + const observations: Awaited>[] = [] + const connection: AcpModelCatalogSessionConnection = { + initialize: async () => ({ + protocolVersion: 1, + agentCapabilities: { sessionCapabilities: { delete: {} } } + }), + start: async () => ({ kind: 'new', sessionId: 'listing', response: NewSessionResponseSchema.parse(recorded) + }), + setConfigOption: () => late.promise, + deleteSession: async () => ({}), + close: async () => {} + } + const run = runAgentModelCatalogSession( + () => connection, + (connection, ready) => + readSessionModels( + acpLaunchSpecFor('opencode2')!, + { kind: 'session', listingNamesConfiguredModel: false }, + connection, + '/workspace', + undefined, + ['opencode/big-pickle'], + undefined, + ready, + (result) => observations.push(result) + ), + { label: 'opencode2', timeoutMs: null } + ) + await vi.advanceTimersByTimeAsync(45_000) + expect(observations[0]?.models.find((row) => row.id === 'opencode/big-pickle')).toMatchObject({ + efforts: [], + effortsKnown: false + }) + const options = structuredClone(recorded.configOptions) + options.find((option) => option.id === 'model')!.currentValue = 'opencode/big-pickle' + late.resolve({ + configOptions: NewSessionResponseSchema.parse({ ...recorded, configOptions: options }) + .configOptions! + }) + const result = await run + expect(result.models.find((row) => row.id === 'opencode/big-pickle')?.efforts).toEqual([ + { value: 'low', label: 'Low' }, + { value: 'high', label: 'High' }, + { value: 'max', label: 'Max' }, + { value: 'default', label: 'Default' } + ]) + expect(result.attemptedModels).toEqual(['opencode/big-pickle']) + expect( + observations.at(-1)?.models.find((row) => row.id === 'opencode/big-pickle')?.efforts + ).toEqual(result.models.find((row) => row.id === 'opencode/big-pickle')?.efforts) + expect( + observations[0]?.models.find((row) => row.id === 'opencode/big-pickle')?.effortsKnown + ).toBe(false) +}) + +function accumulatedCatalogHarness(count: number) { + const baseSpec = acpLaunchSpecFor('opencode2')! + const spec = { + ...baseSpec, + managedRuntime: undefined, + account: { + ...baseSpec.account, + environment: (_home: unknown, env: Record) => env + } + } + const config = (selected = 'm0') => + NewSessionResponseSchema.parse({ + ...recorded, + configOptions: recorded.configOptions.map((option) => + option.id === 'model' + ? { + ...option, + currentValue: selected, + options: Array.from({ length: count }, (_, index) => ({ + value: `m${index}`, + name: `Model ${index}` + })) + } + : option + ) }) - await vi.advanceTimersByTimeAsync(5_000) - const result = await run - expect(result.models.find((row) => row.id === 'opencode/fledge-alpha-free')).toMatchObject({ - label: 'opencode/Fledge Alpha Free', - efforts: [ - { value: 'low', label: 'Low' }, - { value: 'high', label: 'High' }, - { value: 'max', label: 'Max' }, - { value: 'default', label: 'Default' } - ] - }) - expect(result.models.find((row) => row.id === 'opencode/big-pickle')).toMatchObject({ - efforts: [], - effortsKnown: false - }) - expect(result.attemptedModels).toEqual(['opencode/big-pickle']) - late.resolve({ configOptions: [] }) + const requests: { + model: string + reply: PromiseWithResolvers< + Awaited> + > + }[] = [] + const connect = vi.fn((): AcpModelCatalogSessionConnection => ({ + initialize: async () => ({ + protocolVersion: 1, + agentCapabilities: { sessionCapabilities: { delete: {} } } + }), + start: async () => ({ kind: 'new', sessionId: 'listing', response: config() }), + setConfigOption: (_id, model) => { + const reply = + Promise.withResolvers< + Awaited> + >() + requests.push({ model, reply }) + return reply.promise + }, + deleteSession: async () => ({}), + close: async () => {} + })) + const store = new AgentModelCatalogStore() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: measure the production publication boundary required by the scaling regression. + const writes = vi.spyOn( + store as unknown as { writeSuccess: ReturnType }, + 'writeSuccess' + ) + const probe = createAcpModelCatalogProbe(spec, { + resolveEnvironment: async () => ({}), + resolveCommand: () => process.execPath, + probeVersion: async () => ({ version: '2.1.0' }), + homePath: '/account', + dataDirectory: '/unused', + connectSession: connect + }) + const service = createAgentModelCatalogService({ + store, + getRecord: () => undefined, + getConversation: () => undefined, + drivesRecord: () => true, + resolveAccountHome: async () => ({ variable: 'HOME', path: '/account' }), + probes: { opencode2: probe } + }) + const externalRead = createStructuredAgentSessionCatalogReads(() => ({ + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: sessionless catalog reads only access modelCatalog; no chat or journal is mocked. + deps: { modelCatalog: service } as StructuredAgentSessionHostDeps, + serialize: async (_id, task) => task(), + scopeOwned: async () => null + })) + const read = (waitForListing = true, selectedModel?: string) => + externalRead({ agent: 'opencode2', workspacePath: '/workspace', waitForListing, selectedModel }) + onTestFinished(() => service.stop()) + const answer = (index: number) => { + const request = requests[index]! + request.reply.resolve({ configOptions: config(request.model).configOptions! }) + } + return { store, writes, service, requests, connect, read, answer } +} + +it.each([10, 100])( + 'accumulates %i untargeted menus without publishing the full catalog per reply', + async (count) => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const h = accumulatedCatalogHarness(count) + const first = h.read() await vi.advanceTimersByTimeAsync(0) - expect(result.models[1]?.effortsKnown).toBe(false) + expect((await first).models).toHaveLength(count) + let delivered = false + const observer = h.read().then((answer) => { + delivered = true + return answer + }) + await vi.advanceTimersByTimeAsync(0) + for (let index = 0; index < count - 1; index++) { + h.answer(index) + await vi.advanceTimersByTimeAsync(0) + if (index < count - 2) { + expect(delivered).toBe(false) + } + } + const final = await observer + expect(final.models?.every((model) => model.effortsKnown !== false)).toBe(true) + expect(h.connect).toHaveBeenCalledOnce() + // One initial inventory and one terminal observation, independent of the target count. + expect(h.writes.mock.calls.length).toBeLessThanOrEqual(2) + expect( + h.writes.mock.calls.reduce((rows, call) => rows + call[2].models.length, 0) + ).toBeLessThanOrEqual(count * 2) } ) + +it.each(['snapshot', 'deadline', 'selected'])( + 'exposes accumulated menus at a %s boundary while the next switch is unanswered', + async (boundary) => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const h = accumulatedCatalogHarness(5) + const first = h.read() + await first + const observer = h.read() + h.answer(0) + await vi.advanceTimersByTimeAsync(6_000) + // m1 has answered; m2 is still pending, beyond the old optional sweep cutoff. + if (boundary === 'snapshot') { + expect((await h.read(false)).models?.find((row) => row.id === 'm1')?.effortsKnown).toBe(true) + } else if (boundary === 'selected') { + expect((await h.read(true, 'm1')).models?.find((row) => row.id === 'm1')?.effortsKnown).toBe( + true + ) + } else { + await vi.advanceTimersByTimeAsync(24_000) + } + expect((await observer).models?.find((row) => row.id === 'm1')?.effortsKnown).toBe(true) + let followed = false + const follow = h.read().then((answer) => { + followed = true + return answer + }) + await vi.advanceTimersByTimeAsync(1_000) + expect(followed).toBe(false) + // New selected demand joins the same owner and is promoted after its outstanding switch. + const selected = h.read(true, 'm4') + h.answer(1) + await vi.advanceTimersByTimeAsync(6_000) + expect(h.requests[2]?.model).toBe('m4') + h.answer(2) + await vi.advanceTimersByTimeAsync(0) + expect((await selected).models?.find((row) => row.id === 'm4')?.effortsKnown).toBe(true) + await follow + h.answer(3) + await vi.advanceTimersByTimeAsync(0) + expect(h.connect).toHaveBeenCalledOnce() + } +) + +it('keeps a slow OpenCode 1.x version response and selects its command listing', async () => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const version = Promise.withResolvers<{ version: string }>() + const requested = Promise.withResolvers() + const base = acpLaunchSpecFor('opencode')! + const session = vi.fn() + const command = vi.fn( + async () => 'opencode/slow-valid-model\n' + ) + const probe = createAcpModelCatalogProbe( + { ...base, account: { ...base.account, environment: (_home, env) => env } }, + { + resolveEnvironment: async () => ({}), + resolveCommand: () => process.execPath, + probeVersion: async () => { + requested.resolve() + return version.promise + }, + dataDirectory: '/unused', + homePath: '/account', + runListing: command, + connectSession: session + } + ) + const result = probe({ variable: 'HOME', path: '/account' }, { workspacePath: '/workspace' }) + await requested.promise + await vi.advanceTimersByTimeAsync(45_000) + version.resolve({ version: '1.18.31' }) + expect((await result).models.map((row) => row.id)).toEqual(['opencode/slow-valid-model']) + expect(session).not.toHaveBeenCalled() + expect(command.mock.calls[0]?.[1]).toMatchObject({ timeoutMs: null }) +}) diff --git a/src/main/acp/discovery/acp-listing-session-models.ts b/src/main/acp/discovery/acp-listing-session-models.ts index 68281dcf4af..d3e2b519830 100644 --- a/src/main/acp/discovery/acp-listing-session-models.ts +++ b/src/main/acp/discovery/acp-listing-session-models.ts @@ -13,8 +13,8 @@ import type { InitializeResponse, SetSessionConfigOptionResponse } from '../generated/acp-protocol.generated' -import { withTimeout } from '../../../shared/promise-timeout-fallback' import type { AgentModelCatalogReady } from '../../native-chat/agent-model-catalog/agent-model-catalog-probe-runner' +import type { AgentModelCatalogProgress } from '../../native-chat/agent-model-catalog/agent-model-catalog-discovery' import { AcpAgentError } from '../transport/acp-errors' /** Session-free capability boundary: no session methods are exposed. */ @@ -62,7 +62,15 @@ export async function readSessionModels( operation?: StructuredAgentResolutionOperation, selectedModels?: readonly string[], launchedModel?: string, - ready?: AgentModelCatalogReady<{ models: AgentSessionModelOption[]; attemptedModels: string[] }> + ready?: AgentModelCatalogReady<{ models: AgentSessionModelOption[]; attemptedModels: string[] }>, + publish?: (result: { models: AgentSessionModelOption[]; attemptedModels: string[] }) => void, + deferred?: { + defer: ( + read: () => { models: AgentSessionModelOption[]; attemptedModels: string[] }, + known: (model: string) => boolean + ) => void + demanded: NonNullable + } ): Promise<{ models: AgentSessionModelOption[]; attemptedModels: string[] }> { const initialized = await connection.initialize() assertStructuredAgentResolutionCurrent(operation) @@ -83,13 +91,14 @@ export async function readSessionModels( } const options = new AcpStructuredOptions(spec.dialect) options.adoptSession(started.response, 'new') - let models = options.read().models + let models = [...options.read().models] const attemptedModels: string[] = [] const partial = () => { stopped = true return { models: [...models], attemptedModels: [...attemptedModels] } } ready?.(cleanup, models.length > 0 ? partial : undefined) + const progress = () => publish?.({ models: [...models], attemptedModels: [...attemptedModels] }) try { if (launchedModel !== undefined) { attemptedModels.push(launchedModel) @@ -100,57 +109,75 @@ export async function readSessionModels( row.id === running ? { ...row, efforts: [], effortsKnown: false as const } : row ) } + progress() } else if (spec.dialect.effortChangesModel) { - const byId = new Map(models.map((model) => [model.id, model])) - const targets = selectedModels ? selectedModels.flatMap((id) => byId.get(id) ?? []) : models - const sweep = async () => { - for (const model of targets) { - if (stopped) { - return - } - assertStructuredAgentResolutionCurrent(operation) - if (model.effortsKnown !== false) { - continue - } - attemptedModels.push(model.id) - try { - for (let attempt = 0; attempt < 2 && !stopped; attempt++) { - const write = options.write('model', model.id) - if (write?.method !== 'config') { - break - } - const reply = await connection.setConfigOption(write.configId, write.value) - if (stopped) { - return - } - assertStructuredAgentResolutionCurrent(operation) - options.adoptConfigOptions(reply.configOptions) - const observed = options.read() - if (observed.current.model !== model.id) { - continue - } - const known = observed.models.find((row) => row.id === model.id) - if (known) { - models = models.map((row) => (row.id === model.id ? known : row)) - } - break - } - } catch { - assertStructuredAgentResolutionCurrent(operation) + progress() + const indices = new Map(models.map((model, index) => [model.id, index])) + const remaining = new Set(selectedModels ?? models.map((model) => model.id)) + const visited = new Set() + const known = (id: string) => { + const index = indices.get(id) + return index !== undefined && models[index].effortsKnown !== false + } + const snapshot = () => ({ models: [...models], attemptedModels: [...attemptedModels] }) + while (!stopped) { + const demanded = deferred?.demanded() ?? new Set(selectedModels) + for (const id of demanded) { + if (!visited.has(id) && indices.has(id)) { + remaining.add(id) } } + const id = [...demanded].find((id) => remaining.has(id)) ?? remaining.values().next().value + if (id === undefined) { + break + } + remaining.delete(id) + visited.add(id) + const index = indices.get(id) + if (index === undefined || known(id)) { + continue + } + assertStructuredAgentResolutionCurrent(operation) + let conclusive = false + try { + for (let attempt = 0; attempt < 2 && !stopped; attempt++) { + const write = options.write('model', id) + if (write?.method !== 'config') { + conclusive = true + break + } + const reply = await connection.setConfigOption(write.configId, write.value) + if (stopped) { + break + } + assertStructuredAgentResolutionCurrent(operation) + options.adoptConfigOptions(reply.configOptions) + const observed = options.read() + if (observed.current.model !== id) { + continue + } + const learned = observed.models.find((row) => row.id === id) + if (learned) { + models[index] = learned + } + conclusive = true + break + } + } catch { + assertStructuredAgentResolutionCurrent(operation) + conclusive = true + } + if (conclusive) { + attemptedModels.push(id) + } + if (deferred?.demanded().has(id)) { + progress() + } else if (deferred) { + deferred.defer(snapshot, known) + } else { + progress() + } } - await withTimeout( - sweep(), - Math.max( - 0, - Math.min( - 5_000, - operation?.deadline === undefined ? Infinity : operation.deadline - Date.now() - ) - ), - undefined - ) stopped = true } return { models, attemptedModels } diff --git a/src/main/acp/discovery/acp-model-catalog-probe.ts b/src/main/acp/discovery/acp-model-catalog-probe.ts index 691b6aa2240..62cf3f8b230 100644 --- a/src/main/acp/discovery/acp-model-catalog-probe.ts +++ b/src/main/acp/discovery/acp-model-catalog-probe.ts @@ -1,3 +1,4 @@ +import { readAgentCliVersion } from '../../agent-launch/preflight/agent-cli-version-probe' import { observeAcpForkSupport } from '../capabilities/acp-fork-support' import { assertStructuredAgentResolutionCurrent, @@ -7,6 +8,7 @@ import { homedir } from 'node:os' import { join } from 'node:path' import type { AgentCliVersionReader } from '../../agent-launch/preflight/agent-cli-version-probe' import type { + AgentModelCatalogProgress, AgentModelCatalogProbe, AgentModelCatalogLookup } from '../../native-chat/agent-model-catalog/agent-model-catalog-discovery' @@ -100,7 +102,8 @@ export function createAcpModelCatalogProbe( run: async ( signal?: AbortSignal, runOperation = operation, - selectedModels?: readonly string[] + selectedModels?: readonly string[], + publish?: AgentModelCatalogProgress ) => { const options = { signal, operation: runOperation, selectedModels } const operation = runOperation @@ -109,7 +112,7 @@ export function createAcpModelCatalogProbe( spec, { command, prefixArgs, cwd, env, operation }, acpListingDependsOnRelease(discovery), - deps.probeVersion + deps.probeVersion ?? ((input) => readAgentCliVersion({ ...input, timeoutMs: null })) ) assertStructuredAgentResolutionCurrent(operation) if (release.status === 'tooOld') { @@ -126,7 +129,7 @@ export function createAcpModelCatalogProbe( env, envToDelete } - const run = { label: spec.agent, signal: options?.signal } + const run = { label: spec.agent, signal: options?.signal, timeoutMs: null } // OMP records every model switch as account usage; starting on the picked model records none. const launchedModel = spec.launchesWithModel && usesModelSpecificAcpEfforts(spec.agent) @@ -143,7 +146,7 @@ export function createAcpModelCatalogProbe( env, envToDelete }, - { site: `${spec.agent}-model-catalog-probe`, signal: options?.signal } + { site: `${spec.agent}-model-catalog-probe`, signal: options?.signal, timeoutMs: null } ) models = listing.parse(stdout) } else if (listing.kind === 'session') { @@ -177,7 +180,28 @@ export function createAcpModelCatalogProbe( operation, options?.selectedModels, model, - spec.dialect.effortChangesModel ? ready : undefined + spec.dialect.effortChangesModel ? ready : undefined, + (result) => + publish?.( + { + ...result, + models: [...result.models], + attemptedModels: [...result.attemptedModels], + origin: 'probe', + complete: false + }, + true + ), + publish?.defer && publish.demanded + ? { + defer: (read, known) => + publish.defer?.( + () => ({ ...read(), origin: 'probe', complete: false }), + known + ), + demanded: publish.demanded + } + : undefined ), run ) @@ -221,7 +245,7 @@ export function createAcpModelCatalogProbe( try { return await ( await resolve(home, options?.workspacePath ?? '', 'recover', options?.operation) - ).run(options?.signal, options?.operation, options?.selectedModels) + ).run(options?.signal, options?.operation, options?.selectedModels, options?.publish) } catch (error) { if (error instanceof AcpAgentError && error.code === -32000) { throw new AgentModelCatalogUnavailableError({ reason: 'notSignedIn' }) diff --git a/src/main/acp/discovery/acp-model-catalog-selected-demand.test.ts b/src/main/acp/discovery/acp-model-catalog-selected-demand.test.ts new file mode 100644 index 00000000000..8353e0574ac --- /dev/null +++ b/src/main/acp/discovery/acp-model-catalog-selected-demand.test.ts @@ -0,0 +1,92 @@ +import type { ProviderProcessLaunch } from '../../provider-process/provider-process-launch' +import { expect, it, vi, onTestFinished } from 'vitest' +import { acpLaunchSpecFor } from '../launch/acp-launch-registry' +import { NewSessionResponseSchema } from '../generated/acp-protocol.generated' +import { createAcpModelCatalogProbe } from './acp-model-catalog-probe' +import type { AcpModelCatalogSessionConnection } from './acp-listing-session-models' +import { AgentModelCatalogStore } from '../../native-chat/agent-model-catalog/agent-model-catalog-store' +import { createAgentModelCatalogService } from '../../native-chat/agent-model-catalog/agent-model-catalog-service' + +it('starts the first OMP listing on demand collected before admission, without a second session', async () => { + const base = acpLaunchSpecFor('omp')! + const connect = vi.fn((launch: ProviderProcessLaunch): AcpModelCatalogSessionConnection => { + const modelAt = launch.args.indexOf('--model') + const current = modelAt !== -1 ? launch.args[modelAt + 1] : 'provider/default' + return { + initialize: async () => ({ + protocolVersion: 1, + agentCapabilities: { sessionCapabilities: { delete: {} } } + }), + start: async () => ({ + kind: 'new', + sessionId: 'listing', + response: NewSessionResponseSchema.parse({ + sessionId: 'listing', + configOptions: [ + { + id: 'model', + name: 'Model', + category: 'model', + type: 'select', + currentValue: current, + options: [ + { value: 'provider/default', name: 'Default' }, + { value: 'provider/selected', name: 'Selected' } + ] + }, + { + id: 'thinking', + name: 'Thinking', + category: 'thought_level', + type: 'select', + currentValue: 'high', + options: [{ value: 'high', name: 'High' }] + } + ] + }) + }), + setConfigOption: async () => { + throw new Error('OMP selects on launch') + }, + deleteSession: async () => ({}), + close: async () => {} + } + }) + const probe = createAcpModelCatalogProbe( + { + ...base, + account: { ...base.account, environment: (_home, env) => env }, + modelDiscovery: { kind: 'session', listingNamesConfiguredModel: false } + }, + { + resolveEnvironment: async () => ({}), + resolveCommand: () => process.execPath, + probeVersion: async () => ({ version: '18.0.0' }), + homePath: '/account', + dataDirectory: '/unused', + connectSession: connect + } + ) + const store = new AgentModelCatalogStore() + const service = createAgentModelCatalogService({ + store, + getRecord: () => undefined, + getConversation: () => undefined, + drivesRecord: () => true, + resolveAccountHome: async () => ({ variable: 'PI_CODING_AGENT_DIR', path: '/account' }), + probes: { omp: probe } + }) + onTestFinished(() => service.stop()) + const result = await service.read({ + agent: 'omp', + workspacePath: '/workspace', + selectedModel: 'provider/selected', + waitForListing: true + }) + expect(result.models?.find((row) => row.id === 'provider/selected')).toMatchObject({ + effortsKnown: true, + efforts: [{ value: 'high', label: 'High' }] + }) + expect(connect).toHaveBeenCalledOnce() + expect(connect.mock.calls[0]?.[0].args).toEqual(['acp', '--model', 'provider/selected']) +}) diff --git a/src/main/agent-launch/preflight/agent-cli-version-probe.test.ts b/src/main/agent-launch/preflight/agent-cli-version-probe.test.ts index 4d53c912f93..c53b3149acc 100644 --- a/src/main/agent-launch/preflight/agent-cli-version-probe.test.ts +++ b/src/main/agent-launch/preflight/agent-cli-version-probe.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { checkAgentCliVersion } from './agent-cli-version-probe' +import { checkAgentCliVersion, readAgentCliVersion } from './agent-cli-version-probe' const { runProcess } = vi.hoisted(() => ({ runProcess: vi.fn() })) vi.mock('@orca/process-host', () => ({ runProcess })) @@ -58,3 +58,15 @@ describe('checkAgentCliVersion', () => { }) }) }) + +it('allows a catalog version prerequisite to complete late without changing normal preflight limits', async () => { + const answer = Promise.withResolvers() + runProcess.mockReturnValueOnce(answer.promise) + const catalog = readAgentCliVersion({ ...INPUT, timeoutMs: null }) + expect(runProcess.mock.lastCall?.[0].timeoutMs).toBeNull() + answer.resolve({ code: 0, stdout: '1.18.31', stderr: '', timedOut: false }) + expect(await catalog).toEqual({ version: '1.18.31' }) + runProcess.mockResolvedValueOnce({ code: null, stdout: '', stderr: '', timedOut: true }) + expect(await readAgentCliVersion(INPUT)).toEqual({ version: null }) + expect(runProcess.mock.lastCall?.[0].timeoutMs).toBe(5_000) +}) diff --git a/src/main/agent-launch/preflight/agent-cli-version-probe.ts b/src/main/agent-launch/preflight/agent-cli-version-probe.ts index f7cfeb3fa76..84dc4ba53d8 100644 --- a/src/main/agent-launch/preflight/agent-cli-version-probe.ts +++ b/src/main/agent-launch/preflight/agent-cli-version-probe.ts @@ -19,6 +19,7 @@ export type AgentCliVersionCheck = export type AgentCliVersionInput = Pick & { operation?: StructuredAgentResolutionOperation + timeoutMs?: number | null prefixArgs?: readonly string[] } @@ -65,7 +66,7 @@ export async function readAgentCliVersion( input: AgentCliVersionInput, parseVersion: typeof parseCliVersion = parseCliVersion ): Promise<{ version: string | null; started?: false }> { - const { prefixArgs = [], operation, ...launch } = input + const { prefixArgs = [], operation, timeoutMs = 5_000, ...launch } = input assertStructuredAgentResolutionCurrent(operation) let result try { @@ -74,13 +75,16 @@ export async function readAgentCliVersion( args: [...prefixArgs, '--version'], terminationBarrier: true, signal: operation?.signal, - timeoutMs: Math.max( - 1, - Math.min( - 5_000, - operation?.deadline === undefined ? Infinity : operation.deadline - Date.now() - ) - ), + timeoutMs: + timeoutMs === null + ? null + : Math.max( + 1, + Math.min( + timeoutMs, + operation?.deadline === undefined ? Infinity : operation.deadline - Date.now() + ) + ), maxOutputBytes: 4_096, killOnOutputLimit: true }) diff --git a/src/main/claude/model-catalog/claude-model-catalog-observation.ts b/src/main/claude/model-catalog/claude-model-catalog-observation.ts new file mode 100644 index 00000000000..68032bdf3d2 --- /dev/null +++ b/src/main/claude/model-catalog/claude-model-catalog-observation.ts @@ -0,0 +1,43 @@ +import type { AgentSessionAccountKind } from '../../../shared/agent-session/agent-session-availability' +import type { AgentModelCatalogSuccess } from '../../native-chat/agent-model-catalog/agent-model-catalog-entry' +import { + isCompleteClaudeModelListing, + listedModels, + record +} from '../structured/launch/claude-structured-model-catalog' +import { claudeCatalogSignedOut } from './claude-model-catalog-signed-out' +import { + claudeFastModeSupport, + wireClaudeModels +} from '../structured/launch/claude-structured-catalog-listing' +import { readClaudeSettingsApplied } from '../structured/launch/claude-structured-session-options' + +export function readClaudeCatalogObservation( + replies: ReadonlyMap, + env: Readonly>, + account: AgentSessionAccountKind +): AgentModelCatalogSuccess { + const listing = replies.get('orca-model-discovery') + const models = listedModels(listing) + const settings = replies.get('orca-model-settings') + const applied = record(record(settings)?.applied) + const configured = + applied && + ['model', 'effort'].every((key) => applied[key] == null || typeof applied[key] === 'string') + ? (readClaudeSettingsApplied(settings) ?? undefined) + : undefined + if (models.length === 0) { + throw new Error('claude listed no models') + } + const fastModeSupport = claudeFastModeSupport(models) + return { + models: wireClaudeModels(models), + origin: 'probe', + complete: isCompleteClaudeModelListing(listing) && configured !== undefined, + ...(fastModeSupport ? { fastModeSupport } : {}), + ...(configured ? { configured } : {}), + ...(claudeCatalogSignedOut(replies.get('orca-signin-discovery'), settings, env) + ? { unavailable: { reason: 'notSignedIn', account } } + : {}) + } +} diff --git a/src/main/claude/model-catalog/claude-model-catalog-probe.test.ts b/src/main/claude/model-catalog/claude-model-catalog-probe.test.ts index 667762251d4..4943768337e 100644 --- a/src/main/claude/model-catalog/claude-model-catalog-probe.test.ts +++ b/src/main/claude/model-catalog/claude-model-catalog-probe.test.ts @@ -1,3 +1,4 @@ +import type { AgentModelCatalogSuccess } from '../../native-chat/agent-model-catalog/agent-model-catalog-entry' import { mkdtempSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' @@ -47,3 +48,71 @@ describe('Claude catalog availability', () => { } }) }) + +it('publishes models before settings and preserves a late sign-in response', async () => { + const settings = Promise.withResolvers() + const signin = Promise.withResolvers() + const inventory = Promise.withResolvers() + const observations: { + success: AgentModelCatalogSuccess + accountKnown?: boolean + }[] = [] + const response = (id: string, value: unknown) => + `${JSON.stringify({ + type: 'control_response', + response: { subtype: 'success', request_id: id, response: value } + })}\n` + const probe = createClaudeModelCatalogProbe({ + ...probeDeps(), + resolveEnv: () => ({}), + runListing: async (launch, options) => { + expect( + launch.stdin + ?.split(String.fromCharCode(10)) + .filter(Boolean) + .map((line) => JSON.parse(line).request_id) + ).toEqual(['orca-model-discovery', 'orca-model-settings', 'orca-signin-discovery']) + let stdout = response('orca-model-discovery', { + models: [{ value: 'opus', displayName: 'Opus', supportsEffort: false }] + }) + options.onOutput?.(stdout.slice(0, 17)) + options.onOutput?.(stdout.slice(17)) + inventory.resolve() + await settings.promise + const settingsReply = response('orca-model-settings', { + applied: { model: 'opus' }, + effective: {} + }) + stdout += settingsReply + options.onOutput?.(settingsReply) + await signin.promise + const signinReply = response('orca-signin-discovery', { + account: { tokenSource: 'none', apiProvider: 'firstParty' } + }) + stdout += signinReply + options.onOutput?.(signinReply) + return stdout + } + }) + const done = probe(HOME('/homes/a'), { + publish: (success, accountKnown) => observations.push({ success, accountKnown }) + }) + await inventory.promise + expect(observations[0]).toMatchObject({ + success: { models: [{ id: 'opus' }], complete: false }, + accountKnown: false + }) + settings.resolve() + await Promise.resolve() + signin.resolve() + expect(await done).toMatchObject({ + models: [{ id: 'opus' }], + configured: { model: 'opus' }, + unavailable: { reason: 'notSignedIn' } + }) + expect(observations.at(-1)).toMatchObject({ + accountKnown: true, + success: { unavailable: { reason: 'notSignedIn' } } + }) + expect(observations[0].success.configured).toBeUndefined() +}) diff --git a/src/main/claude/model-catalog/claude-model-catalog-probe.ts b/src/main/claude/model-catalog/claude-model-catalog-probe.ts index 95f19867042..c885cf01b5f 100644 --- a/src/main/claude/model-catalog/claude-model-catalog-probe.ts +++ b/src/main/claude/model-catalog/claude-model-catalog-probe.ts @@ -12,17 +12,8 @@ import { } from '../../../shared/claude-model-list-probe' import { requireLegacyAgentSessionAccountHome } from '../../../shared/agent-session/lifecycle/agent-session-account-home' import { runClaudeBackgroundAttempts } from '../../../shared/claude-background-run-isolation' -import { - isCompleteClaudeModelListing, - listedModels, - record -} from '../structured/launch/claude-structured-model-catalog' -import { claudeCatalogSignedOut } from './claude-model-catalog-signed-out' -import { - claudeFastModeSupport, - wireClaudeModels -} from '../structured/launch/claude-structured-catalog-listing' -import { readClaudeSettingsApplied } from '../structured/launch/claude-structured-session-options' +import { listedModels, record } from '../structured/launch/claude-structured-model-catalog' +import { readClaudeCatalogObservation } from './claude-model-catalog-observation' import { readClaudeControlResponses } from '../sdk-transport/claude-control-responses' import { claudeStructuredLaunchArgs } from '../structured/launch/claude-structured-launch-args' import { AgentModelCatalogUnavailableError } from '../../native-chat/agent-model-catalog/agent-model-catalog-unavailable' @@ -41,6 +32,7 @@ import { type ClaudeEnvDeps } from '../structured/launch/claude-structured-child-env' import type { + AgentModelCatalogProgress, AgentModelCatalogProbe, AgentModelCatalogLookup } from '../../native-chat/agent-model-catalog/agent-model-catalog-discovery' @@ -195,7 +187,9 @@ export function createClaudeModelCatalogProbe( }, run: async ( signal?: AbortSignal, - runOperation = operation + runOperation = operation, + _selectedModels?: readonly string[], + publish?: AgentModelCatalogProgress ): Promise => { assertStructuredAgentResolutionCurrent(runOperation) signal?.throwIfAborted() @@ -204,6 +198,14 @@ export function createClaudeModelCatalogProbe( run: (strictMcp) => { assertStructuredAgentResolutionCurrent(runOperation) signal?.throwIfAborted() + const requests = [ + 'orca-model-discovery', + 'orca-model-settings', + 'orca-signin-discovery' + ] + const settled = new Set() + const replies = new Map() + let incomplete: string[] = [] return (deps.runListing ?? runAgentModelCatalogListing)( { command, @@ -219,16 +221,37 @@ export function createClaudeModelCatalogProbe( process.platform ), signal, - partialOutput: { - timeoutMs: 2_000, - ready: (stdout) => { - const replies = readClaudeControlResponses(stdout) - return ( - listedModels(replies.get('orca-model-discovery')).length > 0 && - replies.has('orca-model-settings') + timeoutMs: null, + onOutput: (chunk) => { + const lines = chunk.split('\n') + incomplete.push(lines[0]) + if (lines.length === 1) { + return + } + const stdout = [incomplete.join(''), ...lines.slice(1, -1)].join('\n') + incomplete = [lines.at(-1)!] + const previous = settled.size + for (const [id, value] of readClaudeControlResponses(stdout, (id) => { + if (requests.includes(id)) { + settled.add(id) + } + })) { + if (requests.includes(id)) { + replies.set(id, value) + } + } + if ( + settled.size > previous && + listedModels(replies.get('orca-model-discovery')).length > 0 + ) { + publish?.( + readClaudeCatalogObservation(replies, env, account), + record(record(replies.get('orca-signin-discovery'))?.account) !== null && + record(record(replies.get('orca-model-settings'))?.effective) !== null ) } - } + }, + completeOutput: () => requests.every((id) => settled.has(id)) } ).catch((error: unknown) => { if (error instanceof AgentModelCatalogProbeError) { @@ -248,33 +271,7 @@ export function createClaudeModelCatalogProbe( } throw outcome } - const stdout = outcome - const replies = readClaudeControlResponses(stdout) - const listing = replies.get('orca-model-discovery') - const models = listedModels(listing) - const settings = replies.get('orca-model-settings') - const applied = record(record(settings)?.applied) - const configured = - applied && - ['model', 'effort'].every( - (key) => applied[key] == null || typeof applied[key] === 'string' - ) - ? (readClaudeSettingsApplied(settings) ?? undefined) - : undefined - if (models.length === 0) { - throw new Error('claude listed no models') - } - const fastModeSupport = claudeFastModeSupport(models) - return { - models: wireClaudeModels(models), - origin: 'probe', - complete: isCompleteClaudeModelListing(listing) && configured !== undefined, - ...(fastModeSupport ? { fastModeSupport } : {}), - ...(configured ? { configured } : {}), - ...(claudeCatalogSignedOut(replies.get('orca-signin-discovery'), settings, env) - ? { unavailable: { reason: 'notSignedIn', account } } - : {}) - } + return readClaudeCatalogObservation(readClaudeControlResponses(outcome), env, account) } } } @@ -290,7 +287,7 @@ export function createClaudeModelCatalogProbe( 'recover', options?.operation ) - ).run(options?.signal), + ).run(options?.signal, options?.operation, options?.selectedModels, options?.publish), { resolve, resolveLegacy: (home, operation, lookup) => resolve(home, undefined, lookup, operation) diff --git a/src/main/claude/sdk-transport/claude-control-responses.ts b/src/main/claude/sdk-transport/claude-control-responses.ts index df1f7ab318c..4c5818b6f1d 100644 --- a/src/main/claude/sdk-transport/claude-control-responses.ts +++ b/src/main/claude/sdk-transport/claude-control-responses.ts @@ -6,7 +6,10 @@ function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value) } -export function readClaudeControlResponses(stdout: string): Map { +export function readClaudeControlResponses( + stdout: string, + settled?: (requestId: string) => void +): Map { const responses = new Map() for (const line of stdout.split(/\r?\n/)) { if (!line.trimStart().startsWith('{') || !line.includes('request_id')) { @@ -19,6 +22,12 @@ export function readClaudeControlResponses(stdout: string): Map continue } const response = frame.response + if ( + (response.subtype === 'success' || response.subtype === 'error') && + typeof response.request_id === 'string' + ) { + settled?.(response.request_id) + } if (response.subtype === 'success' && typeof response.request_id === 'string') { responses.set(response.request_id, response.response) } diff --git a/src/main/codex/app-server/codex-app-server-session.ts b/src/main/codex/app-server/codex-app-server-session.ts index 29668ce02cb..03579f0a2b1 100644 --- a/src/main/codex/app-server/codex-app-server-session.ts +++ b/src/main/codex/app-server/codex-app-server-session.ts @@ -25,7 +25,7 @@ export type CodexAppServerInvocation = { env?: Record /** Keys stripped after the overlay, e.g. CODEX_HOME when targeting the real ~/.codex. */ envToDelete?: readonly string[] - timeoutMs: number + timeoutMs: number | null signal?: AbortSignal experimentalApi?: boolean /** Kill unsupervised plugin trees before stdin EOF; root exit leaves children behind and blocks tree killing. */ @@ -186,15 +186,18 @@ export async function runCodexAppServerSession( failPending(error) rejectDeadline(error) } - const deadline = setTimeout( - () => - endSession( - new CodexAppServerTimeoutError( - `codex app-server session exceeded ${invocation.timeoutMs}ms (${invocation.command})` + const deadline = + invocation.timeoutMs === null + ? undefined + : setTimeout( + () => + endSession( + new CodexAppServerTimeoutError( + `codex app-server session exceeded ${invocation.timeoutMs}ms (${invocation.command})` + ) + ), + invocation.timeoutMs ) - ), - invocation.timeoutMs - ) const onAbort = (): void => endSession(new Error('codex app-server session stopped')) if (invocation.signal?.aborted) { onAbort() diff --git a/src/main/codex/structured/catalog/codex-model-catalog-probe.test.ts b/src/main/codex/structured/catalog/codex-model-catalog-probe.test.ts index c2b920e16fb..3d3d8d1125c 100644 --- a/src/main/codex/structured/catalog/codex-model-catalog-probe.test.ts +++ b/src/main/codex/structured/catalog/codex-model-catalog-probe.test.ts @@ -1,3 +1,6 @@ +import { runAgentModelCatalogSession } from '../../../native-chat/agent-model-catalog/agent-model-catalog-probe-runner' +import { PrioritySemaphore } from '../../../../shared/priority-semaphore' +import type { AgentModelCatalogSuccess } from '../../../native-chat/agent-model-catalog/agent-model-catalog-entry' import { mkdtemp, mkdir, writeFile, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' @@ -7,8 +10,9 @@ import { } from '../../configuration/homes/codex-home-paths' import { createAgentModelCatalogService } from '../../../native-chat/agent-model-catalog/agent-model-catalog-service' import { AgentModelCatalogStore } from '../../../native-chat/agent-model-catalog/agent-model-catalog-store' -import { afterEach, describe, expect, it, vi } from 'vitest' +import { afterEach, describe, expect, it, vi, onTestFinished } from 'vitest' import { createCodexModelCatalogProbe } from './codex-model-catalog-probe' +import { fetchCodexModelCatalogListing } from './codex-structured-model-catalog' import { runCodexAppServerSession, type CodexAppServerInvocation @@ -132,7 +136,7 @@ describe('codex model catalog probe', () => { }) const success = await probe({ variable: 'CODEX_HOME', path: '/homes/a' }) expect(success.complete).toBe(false) - expect(configReadTimeouts).toEqual([3_000]) + expect(configReadTimeouts).toEqual([null]) expect(configReadFailures).toEqual([ expect.objectContaining({ name: 'CodexAppServerTimeoutError', @@ -200,3 +204,214 @@ it('observes the shared source config while ignoring mirror writes, and observes await rm(folder, { recursive: true, force: true }) } }) + +it('delivers inventory while config and account are pending, then keeps both late replies', async () => { + const config = Promise.withResolvers() + const account = Promise.withResolvers() + const inventoried = Promise.withResolvers() + const observations: AgentModelCatalogSuccess[] = [] + const probe = createCodexModelCatalogProbe({ + resolveEnvironment: async () => ({}), + resolveCommand: () => ({ command: process.execPath, prefixArgs: [] }), + runSession: async (_invocation, body) => + body({ + request: async (method, _params, options) => { + if (options?.timeoutMs !== null) { + throw new Error('catalog request retained its response deadline') + } + if (method === 'config/read') { + return config.promise + } + if (method === 'account/read') { + return account.promise + } + return { data: [MODEL_ROW], nextCursor: null } + }, + notify: () => {} + }) + }) + const done = probe( + { variable: 'CODEX_HOME', path: '/account' }, + { + publish: (success) => { + observations.push(success) + inventoried.resolve() + } + } + ) + await inventoried.promise + expect(observations[0]).toMatchObject({ models: [{ id: 'gpt-live' }], complete: false }) + expect(observations[0].configured).toBeUndefined() + config.resolve({ config: { model: 'gpt-live', model_reasoning_effort: 'high' } }) + account.resolve({ requiresOpenaiAuth: true, account: null }) + expect(await done).toMatchObject({ + models: [{ id: 'gpt-live', defaultEffort: 'high' }], + complete: true, + unavailable: { reason: 'notSignedIn' } + }) + expect(observations[0].complete).toBe(false) +}) + +it.each([ + ['empty', { requiresOpenaiAuth: true, account: null }, 'notSignedIn'], + ['rejected', { requiresOpenaiAuth: true, account: null }, 'notSignedIn'], + ['empty', { requiresOpenaiAuth: false, account: { type: 'apiKey' } }, undefined], + ['empty', {}, undefined] +] as const)( + 'classifies %s inventory after delayed account evidence %j without optional config, retaining admission through close', + async (kind, accountResult, reason) => { + const configRead = vi.fn(() => new Promise(() => {})) + const account = Promise.withResolvers() + const inventory = Promise.withResolvers() + const closing = Promise.withResolvers() + const closed = Promise.withResolvers() + onTestFinished(() => { + account.resolve({ requiresOpenaiAuth: true, account: null }) + closed.resolve() + }) + const probe = createCodexModelCatalogProbe({ + resolveEnvironment: async () => ({}), + resolveCommand: () => ({ command: process.execPath, prefixArgs: [] }), + runSession: async (invocation, body) => + runAgentModelCatalogSession( + () => ({ + request: async (method: string) => { + if (method === 'account/read') { + return account.promise + } + if (method === 'config/read') { + return configRead() + } + inventory.resolve() + if (kind === 'rejected') { + throw new Error('model listing refused') + } + return { data: [], nextCursor: null } + }, + notify: () => {}, + close: () => { + closing.resolve() + return closed.promise + } + }), + body, + { label: 'codex', signal: invocation.signal, timeoutMs: null } + ) + }) + const store = new AgentModelCatalogStore() + const admission = new PrioritySemaphore(1) + const first = store.refresh( + 'signed-out', + 'codex', + (_selected, operation) => + probe( + { variable: 'CODEX_HOME', path: '/account' }, + { operation, signal: operation.signal } + ), + { admission } + ) + await inventory.promise + expect(store.failure('signed-out')).toBeNull() + expect(store.isListing('signed-out')).toBe(true) + account.resolve(accountResult) + await closing.promise + let nextStarted = false + const next = store.refresh( + 'next', + 'codex', + async () => { + nextStarted = true + return { + origin: 'probe', + complete: true, + models: [{ id: 'next', label: 'Next', isDefault: true, efforts: [] }] + } + }, + { admission } + ) + await Promise.resolve() + expect(nextStarted).toBe(false) + closed.resolve() + expect(await first).toBeNull() + expect(configRead).not.toHaveBeenCalled() + expect(store.failure('signed-out')).not.toBeNull() + expect(store.failure('signed-out')?.unavailable).toEqual(reason ? { reason } : undefined) + expect((await next)?.models[0]?.id).toBe('next') + } +) + +it('settles a failed inventory and pending sign-in classification when the real transport closes', async () => { + const server = String.raw` + const readline = require('node:readline') + readline.createInterface({ input: process.stdin }).on('line', (line) => { + const message = JSON.parse(line) + if (message.method === 'account/read') return + if (message.method === 'model/list') { + process.stdout.write(JSON.stringify({ id: message.id, error: { code: -32000, message: 'inventory unavailable' } }) + '\n', () => process.exit(1)) + return + } + if (typeof message.id === 'number') process.stdout.write(JSON.stringify({ id: message.id, result: {} }) + '\n') + }) + ` + const store = new AgentModelCatalogStore() + const probe = createCodexModelCatalogProbe({ + resolveEnvironment: async () => ({}), + resolveCommand: () => ({ command: process.execPath, prefixArgs: [] }), + runSession: (invocation, body) => + runCodexAppServerSession( + { ...invocation, command: process.execPath, cliPath: null, args: ['-e', server] }, + body + ) + }) + const home = { variable: 'CODEX_HOME', path: '/account' } + await expect( + store.refresh('closed', 'codex', async (_models, operation) => + probe(home, { operation, signal: operation.signal }) + ) + ).resolves.toBeNull() + expect(store.isListing('closed')).toBe(false) + expect(store.failure('closed')).not.toBeNull() +}, 10_000) + +it('keeps a later nonempty page and late configured model after an empty first page', async () => { + const config = Promise.withResolvers() + const observations: string[][] = [] + const request = vi.fn(async (method: string, params: unknown) => { + if (method === 'config/read') { + return config.promise + } + return (params as { cursor?: string }).cursor + ? { data: [MODEL_ROW], nextCursor: null } + : { data: [], nextCursor: 'next' } + }) + const result = fetchCodexModelCatalogListing({ + connection: { request }, + timeoutMs: null, + publish: (listing) => observations.push(listing.models.map((model) => model.id)) + }) + await vi.waitFor(() => expect(observations).toEqual([['gpt-live']])) + config.resolve({ config: { model: 'gpt-live', model_reasoning_effort: 'high' } }) + expect(await result).toMatchObject({ + complete: true, + models: [{ id: 'gpt-live' }], + configured: { model: 'gpt-live', effort: 'high' } + }) +}) + +it.each(['empty', 'malformed', 'capped'])( + 'returns incomplete %s inventory without optional configuration', + async (kind) => { + const request = vi.fn(async (method: string) => { + if (method === 'config/read') { + throw new Error('configuration cannot provide inventory') + } + return kind === 'malformed' + ? { data: [{}], nextCursor: null } + : { data: [], nextCursor: kind === 'capped' ? 'more' : null } + }) + expect( + await fetchCodexModelCatalogListing({ connection: { request }, timeoutMs: null }) + ).toEqual({ models: [], complete: false }) + expect(request.mock.calls.every(([method]) => method === 'model/list')).toBe(true) + } +) diff --git a/src/main/codex/structured/catalog/codex-model-catalog-probe.ts b/src/main/codex/structured/catalog/codex-model-catalog-probe.ts index 29462b3a6f9..b56dbde9ad7 100644 --- a/src/main/codex/structured/catalog/codex-model-catalog-probe.ts +++ b/src/main/codex/structured/catalog/codex-model-catalog-probe.ts @@ -27,8 +27,8 @@ import { type CodexStructuredInvocationDeps } from '../launch/codex-structured-launch-resolution' import { requireLegacyAgentSessionAccountHome } from '../../../../shared/agent-session/lifecycle/agent-session-account-home' -import { AGENT_MODEL_CATALOG_PROBE_TIMEOUT_MS } from '../../../native-chat/agent-model-catalog/agent-model-catalog-probe-runner' import type { + AgentModelCatalogProgress, AgentModelCatalogProbe, AgentModelCatalogLookup } from '../../../native-chat/agent-model-catalog/agent-model-catalog-discovery' @@ -123,12 +123,31 @@ export function createCodexModelCatalogProbe( identity: { command, args, cwd: listIn, overlay: Object.entries(overlay).sort() }, run: async ( signal?: AbortSignal, - runOperation = operation + runOperation = operation, + _selectedModels?: readonly string[], + publish?: AgentModelCatalogProgress ): Promise => { assertStructuredAgentResolutionCurrent(runOperation) signal?.throwIfAborted() deps.prepareHome?.(accountHomePath) const run = deps.runSession ?? runCodexAppServerSession + const successFromListing = ( + listing: Awaited>, + unavailable: AgentModelCatalogSuccess['unavailable'] + ): AgentModelCatalogSuccess => { + return { + models: cwd + ? listing.models + : applyCodexConfiguredLaunchDefaults(listing.models, { + model: listing.configured?.model ?? null, + effort: listing.configured?.effort ?? null + }), + complete: listing.complete !== false, + ...(cwd && listing.configured ? { configured: listing.configured } : {}), + origin: 'probe', + ...(unavailable ? { unavailable } : {}) + } + } const { listing, unavailable } = await run( { command, @@ -136,44 +155,60 @@ export function createCodexModelCatalogProbe( ...(listIn ? { cwd: listIn } : {}), cliPath: command, env, - timeoutMs: AGENT_MODEL_CATALOG_PROBE_TIMEOUT_MS, + timeoutMs: null, ...(signal ? { signal } : {}) }, async (rpc) => { - assertStructuredAgentResolutionCurrent(runOperation) - // Overlap model and account reads to avoid serial probe latency. - const listed = fetchCodexModelCatalogListing({ + let latest: Awaited> | undefined + let accountKnown = false + let unavailable: AgentModelCatalogSuccess['unavailable'] + const observe = (next: NonNullable) => { + assertStructuredAgentResolutionCurrent(runOperation) + latest = next + publish?.(successFromListing(next, unavailable), accountKnown) + } + const account = rpc + .request('account/read', { refreshToken: false }, { timeoutMs: null }) + .catch(() => undefined) + .then((response: unknown) => { + assertStructuredAgentResolutionCurrent(runOperation) + if ( + typeof response === 'object' && + response !== null && + !Array.isArray(response) && + 'account' in response + ) { + accountKnown = true + if ( + 'requiresOpenaiAuth' in response && + response.requiresOpenaiAuth === true && + response.account === null + ) { + const kind = deps.resolveAccountKind?.(accountHomePath) + unavailable = { reason: 'notSignedIn', ...(kind ? { account: kind } : {}) } + } + } + if (latest) { + observe(latest) + } + }) + void account.catch(() => {}) + const listing = await fetchCodexModelCatalogListing({ connection: rpc, cwd, - operation: runOperation + operation: runOperation, + timeoutMs: null, + publish: observe + }).catch(async (error: unknown) => { + await account + if (unavailable) { + throw new AgentModelCatalogUnavailableError(unavailable) + } + throw error }) - void listed.catch(() => {}) - assertStructuredAgentResolutionCurrent(runOperation) - const response: unknown = await rpc - .request('account/read', { refreshToken: false }, { timeoutMs: 2_000 }) - // Failed account reads do not establish sign-out. - .catch(() => undefined) - assertStructuredAgentResolutionCurrent(runOperation) - if ( - typeof response !== 'object' || - response === null || - Array.isArray(response) || - !('requiresOpenaiAuth' in response) || - response.requiresOpenaiAuth !== true || - !('account' in response) || - response.account !== null - ) { - return { listing: await listed, unavailable: undefined } - } - const account = deps.resolveAccountKind?.(accountHomePath) - const signedOut = { reason: 'notSignedIn' as const, ...(account ? { account } : {}) } - // Gateways may misreport sign-out; keep models that were successfully listed. - const inHand = await listed.catch(() => null) - assertStructuredAgentResolutionCurrent(runOperation) - if (!inHand?.models.length) { - throw new AgentModelCatalogUnavailableError(signedOut) - } - return { listing: inHand, unavailable: signedOut } + observe(listing) + await account + return { listing, unavailable } } ).catch((error: unknown) => { if (isMissingProviderExecutable(error, command)) { @@ -183,20 +218,12 @@ export function createCodexModelCatalogProbe( }) assertStructuredAgentResolutionCurrent(runOperation) if (listing.models.length === 0) { + if (unavailable) { + throw new AgentModelCatalogUnavailableError(unavailable) + } throw new Error('codex app-server listed no models') } - return { - models: cwd - ? listing.models - : applyCodexConfiguredLaunchDefaults(listing.models, { - model: listing.configured?.model ?? null, - effort: listing.configured?.effort ?? null - }), - complete: listing.complete !== false, - ...(cwd && listing.configured ? { configured: listing.configured } : {}), - origin: 'probe', - ...(unavailable ? { unavailable } : {}) - } + return successFromListing(listing, unavailable) } } } @@ -206,7 +233,10 @@ export function createCodexModelCatalogProbe( >( async (home, options) => (await resolve(home, options?.cwd ?? process.cwd(), 'recover', options?.operation)).run( - options?.signal + options?.signal, + options?.operation, + options?.selectedModels, + options?.publish ), { resolve, diff --git a/src/main/codex/structured/catalog/codex-structured-model-catalog.ts b/src/main/codex/structured/catalog/codex-structured-model-catalog.ts index 55f15f01174..e56a3ec12bc 100644 --- a/src/main/codex/structured/catalog/codex-structured-model-catalog.ts +++ b/src/main/codex/structured/catalog/codex-structured-model-catalog.ts @@ -154,16 +154,17 @@ export async function listCodexModelPages( export async function fetchCodexModelCatalogListing(input: { connection: Pick - timeoutMs?: number + timeoutMs?: number | null deadlineMs?: number cwd?: string operation?: StructuredAgentResolutionOperation + publish?: (listing: CodexModelCatalogListing) => void }): Promise { const deadline = Math.min( input.operation?.deadline ?? Infinity, input.deadlineMs === undefined ? Infinity : Date.now() + input.deadlineMs ) - const remainingTimeout = (): number | undefined => { + const remainingTimeout = (): number | null | undefined => { assertStructuredAgentResolutionCurrent(input.operation) if (!Number.isFinite(deadline)) { return input.timeoutMs @@ -182,6 +183,11 @@ export async function fetchCodexModelCatalogListing(input: { return response }) ) + assertStructuredAgentResolutionCurrent(input.operation) + if (parsedModels.length === 0) { + return { models: [], complete: false } + } + input.publish?.({ models: parsedModels, complete: false }) const configured = await readCodexConfiguredLaunchDefaults( input.connection, remainingTimeout(), diff --git a/src/main/codex/structured/launch/codex-configured-launch-defaults.ts b/src/main/codex/structured/launch/codex-configured-launch-defaults.ts index 42a43a73027..458c14b1988 100644 --- a/src/main/codex/structured/launch/codex-configured-launch-defaults.ts +++ b/src/main/codex/structured/launch/codex-configured-launch-defaults.ts @@ -17,12 +17,12 @@ function configuredText(value: unknown): string | null { export async function readCodexConfiguredLaunchDefaults( connection: Pick, - timeoutMs = CONFIG_READ_TIMEOUT_MS, + timeoutMs: number | null = CONFIG_READ_TIMEOUT_MS, cwd?: string ): Promise { try { const response = await connection.request('config/read', cwd ? { cwd } : {}, { - timeoutMs: Math.min(timeoutMs, CONFIG_READ_TIMEOUT_MS) + timeoutMs: timeoutMs === null ? null : Math.min(timeoutMs, CONFIG_READ_TIMEOUT_MS) }) const config = typeof response === 'object' && response !== null && 'config' in response diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-account-resolution.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-account-resolution.ts new file mode 100644 index 00000000000..4c4f8d3a8fe --- /dev/null +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-account-resolution.ts @@ -0,0 +1,46 @@ +import { independentAgentModelCatalog } from '../../../shared/agent-session/options/agent-model-catalog-ownership' +import type { AgentModelCatalogServiceDeps } from './agent-model-catalog-service-contract' +import type { AgentModelCatalogLaunch } from './agent-model-catalog-refresh' +import { recordCatalogKey } from './agent-model-catalog-fingerprint' +import { newChatCatalogKey } from './agent-model-catalog-prewarm' +import { catalogFolder, sessionWorkspacePath } from './agent-model-catalog-workspace' + +export async function resolveAgentModelCatalogAccount( + deps: AgentModelCatalogServiceDeps, + launch: AgentModelCatalogLaunch, + current: () => boolean +) { + const { agent, records } = launch + const { record, scoped } = records + const workspacePath = + launch.inSessionWorkspace || + (record && + launch.workspacePath === undefined && + !launch.legacy && + (independentAgentModelCatalog(agent) || deps.agentReadsProjectModelConfig?.(agent) === true)) + ? record + ? await sessionWorkspacePath(deps, record) + : null + : launch.workspacePath + if (!current()) { + return null + } + const folder = await catalogFolder(deps, agent, workspacePath, scoped) + if (!current()) { + return null + } + const account = scoped + ? recordCatalogKey(scoped, folder) + : await newChatCatalogKey(deps, agent, folder) + if (!current() || !account) { + return null + } + return { + record, + scoped, + workspacePath, + folder, + base: account.fingerprint, + probeHome: account.accountHome + } +} diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-discovery.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-discovery.ts index bcb1cfa24f4..6d7c579c24e 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-discovery.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-discovery.ts @@ -2,6 +2,15 @@ import type { StructuredAgentResolutionOperation } from '../agent-launch/structu import type { AgentSessionAccountHome } from '../../../shared/agent-session/lifecycle/agent-session-account-home' import type { AgentModelCatalogSuccess } from './agent-model-catalog-entry' +export type AgentModelCatalogProgress = (( + success: AgentModelCatalogSuccess, + accountKnown?: boolean +) => void) & { + /** Retain the latest accumulator; only real reads or settlement materialize it. */ + defer?: (read: () => AgentModelCatalogSuccess, known: (model: string) => boolean) => void + demanded?: () => ReadonlySet +} + export type AgentModelCatalogResolvedProbe = { identity: unknown inputStamp?: string @@ -9,7 +18,8 @@ export type AgentModelCatalogResolvedProbe = { run: ( signal?: AbortSignal, operation?: StructuredAgentResolutionOperation, - selectedModels?: readonly string[] + selectedModels?: readonly string[], + publish?: AgentModelCatalogProgress ) => Promise } @@ -24,6 +34,7 @@ export type AgentModelCatalogProbe = (( cwd?: string selectedModels?: readonly string[] operation?: StructuredAgentResolutionOperation + publish?: AgentModelCatalogProgress } ) => Promise) & { resolveLegacy?: ( diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-entry-lifetime.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-entry-lifetime.ts index 572f58f5cf2..ea55d5a5ce0 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-entry-lifetime.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-entry-lifetime.ts @@ -129,11 +129,16 @@ function writeAgentModelCatalogSuccess( fingerprint: string, agent: string, success: AgentModelCatalogSuccess, - source: AgentModelCatalogSource + source: AgentModelCatalogSource, + settled = true, + accountKnown = settled, + unchanged = false ): AgentModelCatalogEntry | null { - const entry = state.entryFromSuccess(fingerprint, agent, success, source) + const entry = + (unchanged ? state.entries.get(fingerprint) : undefined) ?? + state.entryFromSuccess(fingerprint, agent, success, source) if (!entry) { - if (source === 'discovery' && independentAgentModelCatalog(agent)) { + if (settled && source === 'discovery' && independentAgentModelCatalog(agent)) { state.failures.probeFailed(fingerprint, agent, new Error('agent listed no models')) } return null @@ -144,18 +149,22 @@ function writeAgentModelCatalogSuccess( state.entries.set(fingerprint, entry) // Live listings cannot clear discovery failures or account-unavailability verdicts. if (source === 'discovery') { - state.dueEntries.delete(fingerprint) - state.failures.listed(fingerprint, agent, success.origin, success.unavailable) - } - if (source === 'discovery' && independentAgentModelCatalog(agent)) { - state.efforts.catalogAttempted(fingerprint, agent, success.complete === true) - if (success.complete !== true) { + if (settled) { + state.dueEntries.delete(fingerprint) + } + if (accountKnown || success.unavailable) { state.failures.listed(fingerprint, agent, success.origin, success.unavailable) } } + if (settled && source === 'discovery' && independentAgentModelCatalog(agent)) { + state.efforts.catalogAttempted(fingerprint, agent, success.complete === true) + } state.evict() // Unchanged per-turn listings need no disk write. - if (!previous || agentModelCatalogListingKey(previous) !== agentModelCatalogListingKey(entry)) { + if ( + previous !== entry && + (!previous || agentModelCatalogListingKey(previous) !== agentModelCatalogListingKey(entry)) + ) { state.persistence?.save(() => [...state.entries.values()]) } return entry @@ -163,7 +172,24 @@ function writeAgentModelCatalogSuccess( export function createAgentModelCatalogSuccessWriter( readState: () => Parameters[0] -): ReturnType { - return (fingerprint, agent, success, source) => - writeAgentModelCatalogSuccess(readState(), fingerprint, agent, success, source) +) { + return ( + fingerprint: string, + agent: string, + success: AgentModelCatalogSuccess, + source: AgentModelCatalogSource, + settled = true, + accountKnown = settled, + unchanged = false + ) => + writeAgentModelCatalogSuccess( + readState(), + fingerprint, + agent, + success, + source, + settled, + accountKnown, + unchanged + ) } diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-lister.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-lister.ts index 8ad7f3fdca3..237702e6dd7 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-lister.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-lister.ts @@ -1,7 +1,5 @@ import type { StructuredAgentResolutionOperation } from '../agent-launch/structured-agent-command-resolution' -import type { AgentSessionAccountHome } from '../../../shared/agent-session/lifecycle/agent-session-account-home' import type { PrioritySemaphore } from '../../../shared/priority-semaphore' -import { AgentModelCatalogProbeError } from './agent-model-catalog-probe-runner' import { startSpan } from '../../observability/tracer' import { AgentModelCatalogListingStoppedError, @@ -13,10 +11,7 @@ import type { AgentModelCatalogSource, AgentModelCatalogSuccess } from './agent-model-catalog-entry' -import type { - AgentModelCatalogProbe, - AgentModelCatalogResolvedProbe -} from './agent-model-catalog-discovery' +import type { AgentModelCatalogProgress } from './agent-model-catalog-discovery' import type { AgentModelCatalogStore } from './agent-model-catalog-store' /** Pin listings to the spawned child's account so later selections cannot receive its results. */ @@ -26,26 +21,25 @@ export type AgentModelCatalogSessionAccess = { accountHomePath: string } -/** `owner` is the listing's own operation: its fixed budget, never a reader's. */ +/** The owner outlives readers and fences all observations from this listing. */ export type AgentModelCatalogListModels = ( selectedModels: readonly string[] | undefined, - owner: StructuredAgentResolutionOperation + owner: StructuredAgentResolutionOperation, + publish?: AgentModelCatalogProgress ) => Promise /** Covers the gap between one waiting read and the next, so a following reader keeps its place. */ const AGENT_MODEL_CATALOG_RELEASE_GRACE_MS = 5_000 -/** Queued listings need a holder; admitted probes run to their own timeout. */ +/** Queued listings need a holder; admitted probes run until completion or retirement. */ export type AgentModelCatalogListingBudget = { - /** Expiry withdraws it from admission, stops its process and drops its late answer. */ + /** Queue expiry withdraws unadmitted work only. */ deadline?: number /** This caller's interest; a queued listing is withdrawn a grace after no holder is left. */ signal?: AbortSignal /** Service shutdown retires the listing immediately, without the holder release grace. */ lifetime?: AbortSignal admission?: PrioritySemaphore - /** Execution time after admission. */ - timeoutMs?: number } export type InFlightListing = { @@ -54,6 +48,7 @@ export type InFlightListing = { current: () => boolean /** Adds a queued holder until its signal aborts; admitted probes need none. */ hold: (signal: AbortSignal | undefined) => void + flush: (model?: string) => void promise: Promise finish: (entry: AgentModelCatalogEntry | null) => void cleanup: () => void @@ -65,7 +60,11 @@ export function demandAgentModelEfforts( cap: number ): void { if (model && flight && flight.targets.size < cap) { + const fresh = !flight.targets.has(model) flight.targets.add(model) + if (fresh) { + flight.flush(model) + } } } @@ -102,11 +101,14 @@ export function refreshAgentModelCatalogListing(input: { fingerprint: string, agent: string, success: AgentModelCatalogSuccess, - source: AgentModelCatalogSource + source: AgentModelCatalogSource, + settled?: boolean, + accountKnown?: boolean, + unchanged?: boolean ) => AgentModelCatalogEntry | null }): Promise { const { fingerprint, agent, listModels } = input - const { signal, lifetime, admission, timeoutMs, deadline: budgetDeadline } = input.budget ?? {} + const { signal, lifetime, admission, deadline: budgetDeadline } = input.budget ?? {} let deadline = budgetDeadline const existing = input.refreshes.get(fingerprint) if (existing) { @@ -126,20 +128,10 @@ export function refreshAgentModelCatalogListing(input: { deadline = next clearTimeout(expiry) if (deadline !== undefined) { - expiry = setTimeout( - () => { - if (admitted) { - stopped.abort( - new AgentModelCatalogProbeError('Model listing work deadline ended', 'timeout') - ) - } else { - retire() - } - }, - Math.max(0, deadline - Date.now()) - ) + expiry = setTimeout(retire, Math.max(0, deadline - Date.now())) } } + let flush: (model?: string) => void = () => {} const flight: InFlightListing = { agent, targets: new Set(input.selectedModel ? [input.selectedModel] : []), @@ -163,6 +155,7 @@ export function refreshAgentModelCatalogListing(input: { holder.addEventListener('abort', abandon, { once: true }) } }, + flush: (model) => flush(model), promise: result.promise, finish: result.resolve, cleanup: () => { @@ -187,6 +180,55 @@ export function refreshAgentModelCatalogListing(input: { const span = startSpan('agentModelCatalog.discovery', { attributes: { agent, lister: 'probe' } }) void Promise.resolve().then(async () => { let release: (() => void) | undefined + let entry: AgentModelCatalogEntry | null = null + let lastSuccess: AgentModelCatalogSuccess | undefined + let lastAccountKnown = false + let published: string | undefined + let deferred: + | { read: () => AgentModelCatalogSuccess; known: (model: string) => boolean } + | undefined + const publish = (success: AgentModelCatalogSuccess, accountKnown = false, settled = false) => { + if (!current() || stopped.signal.aborted || success.models.length === 0) { + return + } + const key = JSON.stringify([success, accountKnown]) + if (key === published) { + if (settled) { + entry = input.writeSuccess( + fingerprint, + agent, + success, + 'discovery', + true, + accountKnown, + true + ) + } + return + } + deferred = undefined + published = key + lastSuccess = success + lastAccountKnown = accountKnown + entry = input.writeSuccess(fingerprint, agent, success, 'discovery', settled, accountKnown) + input.listingWaiters.notify(fingerprint) + } + const progress: AgentModelCatalogProgress = Object.assign(publish, { + defer: (read: () => AgentModelCatalogSuccess, known: (model: string) => boolean) => { + if (current() && !stopped.signal.aborted) { + deferred = { read, known } + } + }, + demanded: () => flight.targets + }) + flush = (model) => { + if (!current() || stopped.signal.aborted || !deferred || (model && !deferred.known(model))) { + return + } + const observation = deferred + deferred = undefined + publish(observation.read(), true) + } try { release = await admission?.acquire(0, stopped.signal) if (!current()) { @@ -196,35 +238,30 @@ export function refreshAgentModelCatalogListing(input: { admitted = true span.setAttribute('queuedMs', Date.now() - queuedAt) held = true - // Admission replaces the abandonment deadline with the probe's own execution budget. - setDeadline( - timeoutMs === undefined - ? budgetDeadline - : Math.min(budgetDeadline ?? Infinity, Date.now() + timeoutMs) - ) + setDeadline(undefined) const owner: StructuredAgentResolutionOperation = { current: () => current() && !stopped.signal.aborted, - signal: stopped.signal, - deadline + signal: stopped.signal } - let targets = input.selectedModel ? [input.selectedModel] : undefined + let targets = flight.targets.size > 0 ? [...flight.targets] : undefined const tried = new Set() - let entry: AgentModelCatalogEntry | null = null do { for (const target of targets ?? []) { tried.add(target) } - const success = await listModels(targets, owner) + const success = await listModels(targets, owner, progress) if (!current()) { return } + deferred = undefined span.setAttribute('models', success.models.length) - entry = input.writeSuccess(fingerprint, agent, success, 'discovery') - const attempted = new Set(targets ? (success.attemptedModels ?? targets) : []) - for (const target of targets ?? []) { - if (attempted.has(target)) { - input.targetAttempted(target) - } + publish(success, lastSuccess ? lastAccountKnown : true, true) + const attempted = new Set(success.attemptedModels ?? targets ?? []) + for (const target of attempted) { + tried.add(target) + input.targetAttempted(target) + } + for (const target of flight.targets) { if (attempted.has(target) || !input.targetDue(target)) { flight.targets.delete(target) } @@ -239,7 +276,17 @@ export function refreshAgentModelCatalogListing(input: { span.end() } catch (error) { span.fail(error instanceof Error ? error : String(error)) - if (current() && !(error instanceof AgentModelCatalogListingStoppedError)) { + if ( + current() && + !stopped.signal.aborted && + !(error instanceof AgentModelCatalogListingStoppedError) + ) { + flush() + if (entry && lastSuccess) { + publish({ ...lastSuccess, complete: false }, lastAccountKnown, true) + flight.finish(entry) + return + } input.failed() input.failures.probeFailed(fingerprint, agent, error) } @@ -250,43 +297,3 @@ export function refreshAgentModelCatalogListing(input: { }) return flight.promise } - -export function scopedAgentModelCatalogLister(input: { - legacy?: boolean - signal: AbortSignal - resolved: AgentModelCatalogResolvedProbe | undefined - probe: AgentModelCatalogProbe | undefined - probeHome: AgentSessionAccountHome | null - workspacePath: string | undefined -}): AgentModelCatalogListModels { - const { resolved, probe, probeHome, workspacePath } = input - return async (selectedModels, operation) => { - const signal = operation.signal - ? AbortSignal.any([input.signal, operation.signal]) - : input.signal - try { - // A slot that frees as the budget ends never starts the queued listing. - if (operation.current?.() === false) { - throw new AgentModelCatalogListingStoppedError('listing budget ended') - } - if (resolved) { - return await resolved.run(signal, operation, selectedModels) - } - if (!probe || !probeHome) { - throw new AgentModelCatalogListingStoppedError('catalog scope unavailable') - } - return await probe(probeHome, { - signal, - workspacePath, - ...(selectedModels ? { selectedModels } : {}), - operation - }) - } catch (error) { - throw input.signal.aborted - ? new AgentModelCatalogListingStoppedError(String(error)) - : operation.signal?.aborted - ? operation.signal.reason - : error - } - } -} diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-listing-waiters.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-listing-waiters.ts index c88765eacf8..a0f7ef5c45b 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-listing-waiters.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-listing-waiters.ts @@ -1,10 +1,6 @@ import type { AgentModelCatalogStore } from './agent-model-catalog-store' import type { AgentModelCatalogReadParams } from './agent-model-catalog-service-contract' import type { AgentModelCatalogBinding } from './agent-model-catalog-snapshot' -import { - independentAgentModelCatalog, - usesModelSpecificAcpEfforts -} from '../../../shared/agent-session/options/agent-model-catalog-ownership' import { withTimeout } from '../../../shared/promise-timeout-fallback' import type { AgentModelCatalogEntry } from './agent-model-catalog-entry' import type { AgentModelCatalogRefresh } from './agent-model-catalog-refresh' @@ -25,12 +21,13 @@ export class AgentModelCatalogListingWaiters { fingerprint: string, waitForChange = false, terminal = false, - signal?: AbortSignal + signal?: AbortSignal, + baseline?: AgentModelCatalogEntry | null ): Promise | null { if (!this.listing(fingerprint) || signal?.aborted) { return null } - const previous = this.read(fingerprint) + const previous = baseline === undefined ? this.read(fingerprint) : baseline return new Promise((resolve) => { const waiters = this.waiters.get(fingerprint) ?? new Set<() => void>() let settled = false @@ -99,29 +96,35 @@ export async function waitForAgentModelCatalogRead(input: { params: AgentModelCatalogReadParams reader: { deadline: number; signal: AbortSignal } store: AgentModelCatalogStore - refresh: AgentModelCatalogRefresh | undefined + refresh: () => AgentModelCatalogRefresh | undefined binding: () => AgentModelCatalogBinding | undefined }): Promise { - const { params, store, refresh } = input + const { params, store } = input // Once the reader stops waiting, nothing it left behind subscribes to or holds a listing wait. const left = new AbortController() + const original = input.binding() + const baseline = original ? store.get(original.fingerprint) : null + const refresh = input.refresh() const wait = async (): Promise => { await refresh?.listing - await (refresh?.defaults ?? input.binding()?.projecting) const bound = left.signal.aborted ? undefined : input.binding() - if ( - bound && - (!store.get(bound.fingerprint) || - independentAgentModelCatalog(params.agent) || - (usesModelSpecificAcpEfforts(params.agent) && store.isListing(bound.fingerprint)) || - store.heldReasonDue(bound.fingerprint)) - ) { - await store.pendingListing( + if (bound) { + const entry = store.get(bound.fingerprint) + if (params.progress !== false && entry && entry !== baseline) { + return + } + const pending = store.pendingListing( bound.fingerprint, true, - params.legacy || usesModelSpecificAcpEfforts(params.agent), - left.signal + params.progress === false, + left.signal, + baseline ) + const projection = + params.progress === false && store.isListing(bound.fingerprint) + ? undefined + : bound.projecting + await (pending && projection ? Promise.race([pending, projection]) : (pending ?? projection)) } if (bound && !store.isListing(bound.fingerprint)) { await refresh?.done diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-lookup.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-lookup.ts index c887e78f97a..9bbd4c45abb 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-lookup.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-lookup.ts @@ -54,6 +54,7 @@ export function createAgentModelCatalogLookup( return undefined } const refresh = refreshes.join(found.key, found.launch) + refresh.hold(operation) refresh.want( usesModelSpecificAcpEfforts(params.agent) ? (params.selectedModel ?? diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-probe-runner.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-probe-runner.ts index dd2a398f9ca..f8576dfc81d 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-probe-runner.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-probe-runner.ts @@ -1,5 +1,6 @@ import type { ClaudeBackgroundFailedExit } from '../../../shared/claude-background-run-isolation' import type { PipedProcessSpawner } from '@orca/process-host/process-spec' +import { StringDecoder } from 'node:string_decoder' import { withTimeout } from '../../../shared/promise-timeout-fallback' import { createOutputSink } from '@orca/process-host/bounded-output-sink' @@ -53,7 +54,7 @@ export type AgentModelCatalogListingLaunch = ProviderProcessLaunch & { export type AgentModelCatalogRunnerOptions = { site: string - timeoutMs?: number + timeoutMs?: number | null maxOutputBytes?: number signal?: AbortSignal spawnImpl?: PipedProcessSpawner @@ -61,6 +62,8 @@ export type AgentModelCatalogRunnerOptions = { inheritedEnv?: NodeJS.ProcessEnv /** Finish required replies even when an optional control request never completes. */ partialOutput?: { ready: (stdout: string) => boolean; timeoutMs: number } + onOutput?: (chunk: string) => void + completeOutput?: () => boolean } /** Bound the listing and await child teardown before success, failure or cancellation settles. */ @@ -68,7 +71,8 @@ export async function runAgentModelCatalogListing( launch: AgentModelCatalogListingLaunch, options: AgentModelCatalogRunnerOptions ): Promise { - const timeoutMs = options.timeoutMs ?? AGENT_MODEL_CATALOG_PROBE_TIMEOUT_MS + const timeoutMs = + options.timeoutMs === undefined ? AGENT_MODEL_CATALOG_PROBE_TIMEOUT_MS : options.timeoutMs throwIfStopped(options.signal, launch.command) const managed = spawnManagedProviderProcess(launch, { site: options.site, @@ -79,6 +83,7 @@ export async function runAgentModelCatalogListing( ...(options.platform ? { platform: options.platform } : {}), ...(options.inheritedEnv ? { inheritedEnv: options.inheritedEnv } : {}) }) + const progressDecoder = options.onOutput ? new StringDecoder('utf8') : undefined const { child } = managed const output = createOutputSink( options.maxOutputBytes ?? AGENT_MODEL_CATALOG_PROBE_MAX_OUTPUT_BYTES @@ -111,11 +116,17 @@ export async function runAgentModelCatalogListing( ? failUnlessReady(`${launch.command} listing timed out`, 'timeout') : fail(`${launch.command} listing stopped`, 'stopped') options.signal?.addEventListener('abort', onAbort, { once: true }) - timer = setTimeout( - () => - failUnlessReady(`${launch.command} did not list models within ${timeoutMs}ms`, 'timeout'), - timeoutMs - ) + timer = + timeoutMs === null + ? undefined + : setTimeout( + () => + failUnlessReady( + `${launch.command} did not list models within ${timeoutMs}ms`, + 'timeout' + ), + timeoutMs + ) child.stdout.on('data', (chunk: Buffer | string) => { output.write(chunk) if (output.truncated()) { @@ -123,8 +134,24 @@ export async function runAgentModelCatalogListing( `${launch.command} listed more than the model catalog reads`, 'output-overflow' ) - } else if (partialTimer === undefined && options.partialOutput?.ready(output.text())) { - partialTimer = setTimeout(() => resolve(output.text()), options.partialOutput.timeoutMs) + } else { + try { + if (progressDecoder) { + options.onOutput?.( + progressDecoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + ) + } + if (options.completeOutput?.()) { + resolve(output.text()) + } else if (partialTimer === undefined && options.partialOutput?.ready(output.text())) { + partialTimer = setTimeout( + () => resolve(output.text()), + options.partialOutput.timeoutMs + ) + } + } catch (error) { + reject(error) + } } }) child.stdout.on('error', () => {}) @@ -185,9 +212,10 @@ export type AgentModelCatalogProbeConnection = { close(): Promise } export async function runAgentModelCatalogSession( open: () => C, body: (connection: C, ready: AgentModelCatalogReady) => Promise, - options: { label: string; timeoutMs?: number; signal?: AbortSignal } + options: { label: string; timeoutMs?: number | null; signal?: AbortSignal } ): Promise { - const timeoutMs = options.timeoutMs ?? AGENT_MODEL_CATALOG_PROBE_TIMEOUT_MS + const timeoutMs = + options.timeoutMs === undefined ? AGENT_MODEL_CATALOG_PROBE_TIMEOUT_MS : options.timeoutMs throwIfStopped(options.signal, options.label) const connection = open() let timer: ReturnType | undefined @@ -218,7 +246,7 @@ export async function runAgentModelCatalogSession /** Its listing ingested and the account's list seeded, or it ended. */ done: Promise + hold: (reader: AgentModelCatalogReadOperation) => void /** Also lists efforts for a reader's selected model. */ want: (model: string | undefined) => void } @@ -55,7 +50,7 @@ export type AgentModelCatalogRefresh = { type Held = AgentModelCatalogRefresh & { workspaceId?: string; end: () => void } /** Equivalent requests share one refresh that belongs to no reader. It ends when it finishes, at - * its setup budget, on its agent's or workspace's retirement, or with the service; readers only + * on its agent's or workspace's retirement, or with the service; readers only * wait on it, each until its own deadline or cancellation, and never extend or end it. */ export function createAgentModelCatalogRefreshes(context: { deps: AgentModelCatalogServiceDeps @@ -70,16 +65,11 @@ export function createAgentModelCatalogRefreshes(context: { const start = (key: string, launch: AgentModelCatalogLaunch): Held => { const { agent } = launch const retirement = deps.store.retirementRevision(agent) - const deadline = Date.now() + AGENT_MODEL_CATALOG_REFRESH_MS const ended = new AbortController() - // Retiring its agent ends it at once, like its budget, its workspace or shutdown. + // Scope work belongs to the host, independent of reader deadlines. const signal = AbortSignal.any([lifetime, ended.signal, deps.store.retirements.signal(agent)]) const current = (): boolean => - !signal.aborted && - Date.now() < deadline && - deps.store.retirementRevision(agent) === retirement - const timer = setTimeout(() => ended.abort(), AGENT_MODEL_CATALOG_REFRESH_MS) - timer.unref?.() + !signal.aborted && deps.store.retirementRevision(agent) === retirement const gone = new Promise((resolve) => { signal.addEventListener('abort', () => resolve(null), { once: true }) }) @@ -95,10 +85,12 @@ export function createAgentModelCatalogRefreshes(context: { const listing = Promise.withResolvers() const defaults = Promise.withResolvers() const targets = new Set() - let list: ((model: string | undefined) => void) | undefined + const holders = new Set() + let register: ((holder?: AbortSignal) => void) | undefined + let demand: ((model: string) => void) | undefined let projecting: Promise | undefined const run = async (): Promise => { - const facts = await bounded(resolveAccount(deps, launch, current)) + const facts = await bounded(resolveAgentModelCatalogAccount(deps, launch, current)) if (!facts) { return } @@ -135,6 +127,7 @@ export function createAgentModelCatalogRefreshes(context: { if (keepDefault !== null) { bound.keepDefault = keepDefault } + bound.projecting = undefined }) projecting = bound.projecting void projecting.finally(defaults.resolve) @@ -146,7 +139,7 @@ export function createAgentModelCatalogRefreshes(context: { base, probeHome, current, - { deadline, current, signal } + { current, signal } ) ) if (!scope) { @@ -171,42 +164,60 @@ export function createAgentModelCatalogRefreshes(context: { probeHome, workspacePath: cwd ?? folder }) - list = (model) => { - if ( - current() && - (deps.store.probeDue(fingerprint) || - deps.store.efforts.due(fingerprint, model) || - deps.store.isListing(fingerprint)) + const attach = (holder: AbortSignal) => { + if (holder.aborted || !current()) { + return + } + completion = deps.store.refresh( + fingerprint, + agent, + (selected, owner, publish) => { + const task = lister(selected, owner, publish) + tasks.add(task) + void task.finally(() => tasks.delete(task)).catch(() => {}) + return task + }, + { signal: holder, lifetime, admission } + ) + } + register = (holder) => { + if (!current()) { + return + } + if (deps.store.isListing(fingerprint)) { + if (holder) { + attach(holder) + } else { + for (const live of holders) { + attach(live) + } + } + } else if ( + deps.store.probeDue(fingerprint) || + [...targets].some((model) => deps.store.efforts.due(fingerprint, model)) ) { - // This refresh holds the listing while it runs; once admitted, the listing runs on. - completion = deps.store.refresh( - fingerprint, - agent, - (selected, owner) => { - const task = lister(selected, owner) - tasks.add(task) - void task.finally(() => tasks.delete(task)).catch(() => {}) - return task - }, - { - signal, - lifetime, - admission, - timeoutMs: AGENT_MODEL_CATALOG_REFRESH_MS - }, - model - ) + for (const live of holders) { + attach(live) + } + for (const model of targets) { + deps.store.efforts.demand(fingerprint, model) + } } } - for (const model of targets.size > 0 ? [...targets] : [undefined]) { - list(model) + demand = (model) => { + if (!deps.store.isListing(fingerprint)) { + register?.() + } else { + deps.store.efforts.demand(fingerprint, model) + } } + register() } listing.resolve() if (launch.passive) { await bound.projecting } else { - // Discovery may finish after this refresh's setup budget ends. + // Readers can leave while current discovery continues. const seeded = Promise.resolve(completion).then((entry) => { if (entry !== null && bound.current()) { seedAccountList(deps, scope, agent) @@ -218,13 +229,31 @@ export function createAgentModelCatalogRefreshes(context: { const end = (): void => ended.abort() const held: Held = { current: () => current() && refreshes.get(key) === held, + hold: (reader) => { + if (reader.signal.aborted) { + return + } + const released = new AbortController() + const timer = + reader.deadline === undefined + ? undefined + : setTimeout(() => released.abort(), Math.max(0, reader.deadline - Date.now())) + const holder = AbortSignal.any([signal, reader.signal, released.signal]) + const leave = () => { + clearTimeout(timer) + holders.delete(holder) + } + holders.add(holder) + holder.addEventListener('abort', leave, { once: true }) + register?.(holder) + }, listing: listing.promise, defaults: defaults.promise, done: Promise.resolve(), want: (model) => { if (model && !targets.has(model)) { targets.add(model) - list?.(model) + demand?.(model) } }, workspaceId: launch.records.record?.location.workspaceId, @@ -234,11 +263,10 @@ export function createAgentModelCatalogRefreshes(context: { held.done = run() .catch(() => {}) .finally(() => listing.resolve()) - // The folder's settings may still arrive, inside the same budget, after the listing ends. + // The folder's settings may still arrive after the listing ends. const settled = held.done .then(() => projecting) .finally(() => { - clearTimeout(timer) ended.abort() defaults.resolve() if (refreshes.get(key) === held) { @@ -266,43 +294,3 @@ export function createAgentModelCatalogRefreshes(context: { } } } - -async function resolveAccount( - deps: AgentModelCatalogServiceDeps, - launch: AgentModelCatalogLaunch, - current: () => boolean -) { - const { agent, records } = launch - const { record, scoped } = records - const workspacePath = - launch.inSessionWorkspace || - (record && - launch.workspacePath === undefined && - !launch.legacy && - (independentAgentModelCatalog(agent) || deps.agentReadsProjectModelConfig?.(agent) === true)) - ? record - ? await sessionWorkspacePath(deps, record) - : null - : launch.workspacePath - if (!current()) { - return null - } - const folder = await catalogFolder(deps, agent, workspacePath, scoped) - if (!current()) { - return null - } - const account = scoped - ? recordCatalogKey(scoped, folder) - : await newChatCatalogKey(deps, agent, folder) - if (!current() || !account) { - return null - } - return { - record, - scoped, - workspacePath, - folder, - base: account.fingerprint, - probeHome: account.accountHome - } -} diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-scoped-lister.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-scoped-lister.ts new file mode 100644 index 00000000000..424b3a0a33b --- /dev/null +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-scoped-lister.ts @@ -0,0 +1,48 @@ +import type { AgentSessionAccountHome } from '../../../shared/agent-session/lifecycle/agent-session-account-home' +import type { + AgentModelCatalogProbe, + AgentModelCatalogResolvedProbe +} from './agent-model-catalog-discovery' +import type { AgentModelCatalogListModels } from './agent-model-catalog-lister' +import { AgentModelCatalogListingStoppedError } from './agent-model-catalog-failures' + +export function scopedAgentModelCatalogLister(input: { + legacy?: boolean + signal: AbortSignal + resolved: AgentModelCatalogResolvedProbe | undefined + probe: AgentModelCatalogProbe | undefined + probeHome: AgentSessionAccountHome | null + workspacePath: string | undefined +}): AgentModelCatalogListModels { + const { resolved, probe, probeHome, workspacePath } = input + return async (selectedModels, operation, publish) => { + const signal = operation.signal + ? AbortSignal.any([input.signal, operation.signal]) + : input.signal + try { + // A slot that frees as the budget ends never starts the queued listing. + if (operation.current?.() === false) { + throw new AgentModelCatalogListingStoppedError('listing budget ended') + } + if (resolved) { + return await resolved.run(signal, operation, selectedModels, publish) + } + if (!probe || !probeHome) { + throw new AgentModelCatalogListingStoppedError('catalog scope unavailable') + } + return await probe(probeHome, { + signal, + workspacePath, + ...(selectedModels ? { selectedModels } : {}), + operation, + publish + }) + } catch (error) { + throw input.signal.aborted + ? new AgentModelCatalogListingStoppedError(String(error)) + : operation.signal?.aborted + ? operation.signal.reason + : error + } + } +} diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-service-contract.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-service-contract.ts index a790ee48ed5..88ec30489ed 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-service-contract.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-service-contract.ts @@ -62,6 +62,8 @@ export type AgentModelCatalogService = { export type AgentModelCatalogReadParams = { agent: string + /** Older readers follow only actual wait expiry, so retain their terminal wait. */ + progress?: boolean records?: AgentModelCatalogReadRecords | null operation?: AgentModelCatalogReadOperation /** Original explicit operation budget, shared with compatibility saved-scope continuation. */ diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-service.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-service.ts index 2f88d2866e4..cc41d75786f 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-service.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-service.ts @@ -82,9 +82,18 @@ export function createAgentModelCatalogService( const operation = params.operation ?? captureRead(params) await waitWithinAgentModelCatalogRead(operation, lookup.request(params, operation)?.done) } + const flush = (params: AgentModelCatalogReadParams) => { + const bound = memoryBinding(params) + if (!params.savedOnly && bound?.current()) { + deps.store.flushProgress(bound.fingerprint) + } + } return { captureRead, - snapshot, + snapshot: (params) => { + flush(params) + return snapshot(params) + }, demand: async (params) => { if (!params.savedOnly) { await demand(params) @@ -112,7 +121,7 @@ export function createAgentModelCatalogService( params, reader: { deadline, signal: operation.signal }, store: deps.store, - refresh: lookup.request(params, operation), + refresh: () => lookup.request(params, operation), binding: () => bindings.get(requestKey(params)) }) } @@ -122,6 +131,7 @@ export function createAgentModelCatalogService( expired && params.operation?.current({ ignoreDeadline: true }) ? { ...params, operation: undefined } : params + flush(answerParams) const result = snapshot(answerParams) const answer = lifetime.signal.aborted ? result diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-listing.test.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-listing.test.ts index b3655d99e3f..c35f172f2d0 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-listing.test.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-listing.test.ts @@ -7,6 +7,7 @@ import { type StructuredAgentResolutionOperation } from '../agent-launch/structured-agent-command-resolution' import { AgentModelCatalogStore } from './agent-model-catalog-store' +import type { AgentModelCatalogProgress } from './agent-model-catalog-discovery' import type { AgentModelCatalogSuccess } from './agent-model-catalog-entry' function listing(id: string): AgentModelCatalogSuccess { @@ -31,7 +32,12 @@ function sharedListingHarness() { deadline?: number }[] = [] const run = vi.fn( - (identity: string, signal?: AbortSignal, operation?: StructuredAgentResolutionOperation) => { + ( + identity: string, + signal?: AbortSignal, + operation?: StructuredAgentResolutionOperation, + _publish?: AgentModelCatalogProgress + ) => { runs.push({ identity, signal, current: operation?.current, deadline: operation?.deadline }) return new Promise((resolve, reject) => { void listingFor(identity).promise.then(resolve) @@ -62,7 +68,12 @@ function sharedListingHarness() { const identity = folder?.startsWith('/own') ? folder : 'codex' return { identity, - run: (signal?: AbortSignal, owner = operation) => run(identity, signal, owner) + run: ( + signal?: AbortSignal, + owner = operation, + _selected?: readonly string[], + publish?: AgentModelCatalogProgress + ) => run(identity, signal, owner, publish) } } }) @@ -94,7 +105,7 @@ it("never answers a newly picked folder with the previous folder's late listing" }) }) -it('keeps the admitted slot and publication until timeout teardown completes', async () => { +it('keeps physical admission through late success and actual teardown', async () => { vi.useFakeTimers() onTestFinished(() => { vi.useRealTimers() @@ -103,10 +114,11 @@ it('keeps the admitted slot and publication until timeout teardown completes', a const admission = new PrioritySemaphore(1) const close = Promise.withResolvers() const closing = Promise.withResolvers() + const answer = Promise.withResolvers() const first = store.refresh( 'first', 'grok', - async (_selected, owner) => + async (_selected, owner, publish) => runAgentModelCatalogSession( () => ({ close: () => { @@ -114,16 +126,15 @@ it('keeps the admitted slot and publication until timeout teardown completes', a return close.promise } }), - async (_connection, ready) => { - ready( - async () => {}, - () => listing('partial') - ) - return new Promise(() => {}) + async () => { + publish?.(listing('partial')) + const result = await answer.promise + assertStructuredAgentResolutionCurrent(owner) + return result }, - { label: 'grok', signal: owner.signal } + { label: 'grok', signal: owner.signal, timeoutMs: null } ), - { admission, timeoutMs: 30_000 } + { admission } ) let secondStarted = false const second = store.refresh( @@ -133,14 +144,230 @@ it('keeps the admitted slot and publication until timeout teardown completes', a secondStarted = true return listing('second') }, - { admission, timeoutMs: 30_000 } + { admission } ) - await vi.advanceTimersByTimeAsync(30_001) + await vi.advanceTimersByTimeAsync(45_000) + expect(store.get('first')?.models[0]?.id).toBe('partial') + answer.resolve(listing('late')) await closing.promise expect(secondStarted).toBe(false) expect(store.isListing('first')).toBe(true) - expect(store.get('first')).toBeNull() + expect(store.get('first')?.models[0]?.id).toBe('partial') close.resolve() - expect((await first)?.models[0]?.id).toBe('partial') + expect((await first)?.models[0]?.id).toBe('late') expect((await second)?.models[0]?.id).toBe('second') }) + +it.each([false, true])( + 'keeps a slow successful listing after its reader leaves (saved=%s)', + async (saved) => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const h = sharedListingHarness() + if (saved) { + const initial = h.service.read({ agent: 'codex', waitForListing: true }) + await vi.advanceTimersByTimeAsync(0) + h.discovery.resolve({ ...listing('saved'), complete: true }) + await initial + h.store.expireScopes('codex') + } + const answer = Promise.withResolvers() + h.run.mockImplementation((_identity, signal, operation) => + runAgentModelCatalogSession( + () => ({ close: async () => {} }), + async () => { + const result = await answer.promise + assertStructuredAgentResolutionCurrent(operation) + return result + }, + { label: 'cold', signal, timeoutMs: null } + ) + ) + const reader = h.service.read({ agent: 'codex', waitForListing: true }) + await vi.advanceTimersByTimeAsync(30_000) + expect(await reader).toMatchObject({ listingInProgress: true, waitExpired: true }) + expect(h.service.snapshot({ agent: 'codex' }).models?.[0]?.id).toBe(saved ? 'saved' : undefined) + await vi.advanceTimersByTimeAsync(15_000) + answer.resolve({ ...listing('slow-success'), complete: true }) + await vi.advanceTimersByTimeAsync(0) + expect(await h.service.read({ agent: 'codex', waitForListing: true })).toMatchObject({ + models: [{ id: 'slow-success' }] + }) + expect(h.service.snapshot({ agent: 'codex' }).listingInProgress).toBeUndefined() + } +) + +it.each([true, false])( + 'delivers progress and waits for a changed observation (progress=%s)', + async (progress) => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const h = sharedListingHarness() + const answer = Promise.withResolvers() + let publish: AgentModelCatalogProgress | undefined + h.run.mockImplementation((_identity, _signal, _operation, notify) => { + publish = notify + return answer.promise + }) + const first = h.service.read({ agent: 'codex', waitForListing: true, progress }) + await vi.advanceTimersByTimeAsync(0) + publish?.({ ...listing('listed'), complete: false }) + await vi.advanceTimersByTimeAsync(progress ? 0 : 30_000) + expect(await first).toMatchObject({ + models: [{ id: 'listed', isDefault: false }], + listingInProgress: true + }) + expect((await first).waitExpired).toBe(progress ? undefined : true) + let next: unknown + const follow = h.service + .read({ agent: 'codex', waitForListing: true, progress: true }) + .then((result) => { + next = result + }) + await vi.advanceTimersByTimeAsync(1_000) + expect(next).toBeUndefined() + publish?.({ ...listing('listed'), complete: true, configured: { model: 'listed' } }, true) + await follow + expect(next).toMatchObject({ + models: [{ id: 'listed', isDefault: true }], + listingInProgress: true + }) + answer.resolve({ ...listing('listed'), complete: true, configured: { model: 'listed' } }) + await vi.advanceTimersByTimeAsync(0) + expect(h.service.snapshot({ agent: 'codex' }).listingInProgress).toBeUndefined() + } +) + +it('keeps an account verdict until discovery actually answers it, including optional failure', async () => { + const store = new AgentModelCatalogStore() + store.recordSuccess( + 'held', + 'codex', + { ...listing('saved'), complete: true, unavailable: { reason: 'notSignedIn' } }, + 'discovery' + ) + store.expireFailure('held') + await store.refresh('held', 'codex', async (_selected, _owner, publish) => { + publish?.({ ...listing('listed'), complete: true }, false) + return { ...listing('listed'), complete: true } + }) + expect(store.failure('held')?.unavailable).toEqual({ reason: 'notSignedIn' }) + await store.refresh('held', 'codex', async (_selected, _owner, publish) => { + publish?.({ ...listing('listed'), complete: true }, true) + return { ...listing('listed'), complete: true } + }) + expect(store.failure('held')).toBeNull() +}) + +it('fences obsolete progress and final results while retaining admission until cancelled teardown', async () => { + const store = new AgentModelCatalogStore() + const admission = new PrioritySemaphore(1) + const started = Promise.withResolvers() + const closing = Promise.withResolvers() + const closed = Promise.withResolvers() + const answer = Promise.withResolvers() + let notify: AgentModelCatalogProgress | undefined + const first = store.refresh( + 'old', + 'grok', + async (_selected, owner, publish) => + runAgentModelCatalogSession( + () => ({ + close: () => { + closing.resolve() + return closed.promise + } + }), + async () => { + notify = publish + publish?.(listing('usable')) + started.resolve() + return answer.promise + }, + { label: 'grok', timeoutMs: null, signal: owner.signal } + ), + { admission } + ) + await started.promise + notify?.defer?.( + () => listing('obsolete-deferred'), + () => true + ) + store.expireScopes('grok') + store.flushProgress('old') + await closing.promise + notify?.(listing('obsolete-progress')) + notify?.defer?.( + () => listing('obsolete-new-deferred'), + () => true + ) + store.flushProgress('old') + answer.resolve(listing('obsolete-final')) + let entered = false + const next = store.refresh( + 'new', + 'grok', + async () => { + entered = true + return listing('new') + }, + { admission } + ) + await Promise.resolve() + expect(entered).toBe(false) + expect(store.get('old')?.models[0]?.id).toBe('usable') + closed.resolve() + expect(await first).toBeNull() + expect((await next)?.models[0]?.id).toBe('new') + expect(store.get('old')?.models[0]?.id).toBe('usable') +}) + +it('withdraws an abandoned queued scope while the two admitted listings continue', async () => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const h = sharedListingHarness() + void h.service.demand({ agent: 'codex', workspacePath: '/own-a' }) + void h.service.demand({ agent: 'codex', workspacePath: '/own-b' }) + const queued = h.service.read({ agent: 'codex', workspacePath: '/own-c', waitForListing: true }) + await vi.advanceTimersByTimeAsync(30_000) + expect(await queued).toMatchObject({ + origin: 'unknown', + listingInProgress: true, + waitExpired: true + }) + await vi.advanceTimersByTimeAsync(5_001) + expect( + h.service.snapshot({ agent: 'codex', workspacePath: '/own-c' }).listingInProgress + ).toBeUndefined() + h.listingFor('/own-a').resolve({ ...listing('a'), complete: true }) + await vi.advanceTimersByTimeAsync(0) + expect(h.service.snapshot({ agent: 'codex', workspacePath: '/own-c' }).origin).toBe('unknown') +}) + +it.each([false, true])( + 'settles an identical final observation, including eviction (evicted=%s)', + async (evicted) => { + const store = new AgentModelCatalogStore() + let first: unknown + const success = { ...listing('same'), complete: true } + const result = await store.refresh('same', 'codex', async (_selected, _owner, publish) => { + publish?.(success, true) + first = store.get('same') + if (evicted) { + for (let index = 0; index < 256; index++) { + store.recordSuccess(`other-${index}`, 'codex', success, 'discovery') + } + expect(store.get('same')).toBeNull() + } + return success + }) + expect(result?.models[0]?.id).toBe('same') + if (!evicted) { + expect(store.get('same')).toBe(first) + } + expect(store.get('same')?.models[0]?.id).toBe('same') + expect(store.failure('same')).toBeNull() + expect(store.isListing('same')).toBe(false) + expect(store.probeDue('same')).toBe(false) + } +) diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-setup.test.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-setup.test.ts index 19c2f6f17f2..a2a15342d1c 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-setup.test.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-shared-setup.test.ts @@ -137,7 +137,7 @@ it('keeps one listing for a workspace on the same folder when another workspace expect(await a).toMatchObject({ models: [{ id: 'other' }] }) }) -it('ends a shorter reader at its own deadline, and a stalled refresh lists nothing after its budget', async () => { +it('ends a shorter reader while scope resolution remains current', async () => { vi.useFakeTimers() onTestFinished(() => { vi.useRealTimers() @@ -163,9 +163,7 @@ it('ends a shorter reader at its own deadline, and a stalled refresh lists nothi // Its readers are gone, but the refresh runs on its own budget. expect(h.service.snapshot({ agent: 'codex', sessionId: 'b' })).toHaveProperty('listingInProgress') await vi.advanceTimersByTimeAsync(29_000) - expect(h.service.snapshot({ agent: 'codex', sessionId: 'b' })).not.toHaveProperty( - 'listingInProgress' - ) + expect(h.service.snapshot({ agent: 'codex', sessionId: 'b' })).toHaveProperty('listingInProgress') gate.resolve() await vi.advanceTimersByTimeAsync(0) expect(h.run).not.toHaveBeenCalled() @@ -177,7 +175,8 @@ it('holds readers of a stalled refresh only to their own deadlines, and starts n vi.useRealTimers() }) const h = harness() - h.folder.gate = Promise.withResolvers() + const gate = Promise.withResolvers() + h.folder.gate = gate void h.service.demand({ agent: 'codex', sessionId: 'a', deadline: Date.now() + 1_000 }) const b = h.service.read({ agent: 'codex', @@ -187,7 +186,7 @@ it('holds readers of a stalled refresh only to their own deadlines, and starts n }) await vi.advanceTimersByTimeAsync(30_000) // b's deadline ended its wait; a reader leaving never restarts the work. - expect(await b).toEqual({ origin: 'unknown' }) + expect(await b).toMatchObject({ origin: 'unknown', listingInProgress: true, waitExpired: true }) expect(h.recordWorkspacePath).toHaveBeenCalledOnce() const c = h.service.read({ agent: 'codex', @@ -195,8 +194,10 @@ it('holds readers of a stalled refresh only to their own deadlines, and starts n waitForListing: true, deadline: Date.now() + 30_000 }) - await vi.waitFor(() => expect(h.run).toHaveBeenCalledOnce()) - expect(h.recordWorkspacePath).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(15_000) + gate.resolve() + await vi.advanceTimersByTimeAsync(0) + expect(h.recordWorkspacePath).toHaveBeenCalledOnce() h.discovery.resolve(listing('fresh')) await vi.advanceTimersByTimeAsync(0) expect(await c).toMatchObject({ models: [{ id: 'fresh' }] }) @@ -226,26 +227,30 @@ it("keeps the folder's defaults when they arrive after a shorter last reader lef expect(await a).toMatchObject({ models: [{ id: 'listed', isDefault: true }] }) }) -it("keeps the listing when the folder's settings stall, naming no default, within the reader's budget", async () => { +it('delivers inventory before slow folder defaults, then delivers the proven default', async () => { vi.useFakeTimers() - onTestFinished(() => { - vi.useRealTimers() - }) + onTestFinished(() => vi.useRealTimers()) const h = harness(agent) - h.folder.settings = Promise.withResolvers() - void h.service.demand({ agent, sessionId: 'a', deadline: Date.now() + 1_000 }) - let answer: unknown - void h.service - .read({ agent, sessionId: 'b', waitForListing: true, deadline: Date.now() + 30_000 }) - .then((result) => { - answer = result - }) - await vi.waitFor(() => expect(h.run).toHaveBeenCalledOnce()) + const settings = Promise.withResolvers() + h.folder.settings = settings + const first = h.service.read({ agent, sessionId: 'b', waitForListing: true }) + await vi.advanceTimersByTimeAsync(0) h.discovery.resolve(listing('listed')) - await vi.advanceTimersByTimeAsync(30_000) - // Whether the project picks its own model stays unknown, so the account's default isn't named. - expect(answer).toMatchObject({ models: [{ id: 'listed', isDefault: false }] }) - expect(answer).not.toHaveProperty('listingInProgress') + await vi.advanceTimersByTimeAsync(0) + expect(await first).toMatchObject({ + models: [{ id: 'listed', isDefault: false }], + listingInProgress: true + }) + let next: unknown + const follow = h.service.read({ agent, sessionId: 'b', waitForListing: true }).then((result) => { + next = result + }) + await vi.advanceTimersByTimeAsync(1_000) + expect(next).toBeUndefined() + settings.resolve() + await follow + expect(next).toMatchObject({ models: [{ id: 'listed', isDefault: true }] }) + expect(next).not.toHaveProperty('listingInProgress') }) it("never lets a moved folder's late settings describe the folder it moved to", async () => { @@ -326,3 +331,41 @@ it('keeps one folder-agent listing for a workspace on the same folder when anoth expect(await a).toMatchObject({ models: [{ id: 'other' }] }) expect(probe.mock.calls.map((call) => call[1]?.workspacePath)).toEqual(['/repo', '/other']) }) + +it.each([10, 100])( + 'registers %i readers and unique targets once per shared flight', + async (count) => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + for (const delayedScope of [false, true]) { + const h = harness('opencode2') + const scope = Promise.withResolvers() + if (delayedScope) { + h.folder.gate = scope + } + const registrations = vi.spyOn(h.store, 'refresh') + const submissions = vi.spyOn(h.store.efforts, 'demand') + const readers: Promise[] = [] + for (let index = 0; index < count; index++) { + readers.push( + h.service.read({ + agent: 'opencode2', + sessionId: 'a', + selectedModel: `m${index}`, + waitForListing: true + }) + ) + if (!delayedScope) { + await vi.advanceTimersByTimeAsync(0) + } + } + scope.resolve() + await vi.advanceTimersByTimeAsync(0) + expect(h.run).toHaveBeenCalledOnce() + expect(registrations.mock.calls.length).toBe(count) + expect(submissions.mock.calls.length).toBe(count) + h.discovery.resolve(listing('shared')) + await Promise.all(readers) + } + } +) diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-snapshot.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-snapshot.ts index 55cb348482b..d654c868e70 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-snapshot.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-snapshot.ts @@ -8,10 +8,7 @@ import type { AgentSessionConversationView } from '../../../shared/agent-session import type { AgentSessionModelCatalogResult } from '../../../shared/agent-session/agent-session-wire' import { agentModelCatalogResultFromEntry } from './agent-model-catalog-entry' import { agentModelCatalogDefaultFollowsAccount } from './agent-model-catalog-workspace' -import { - independentAgentModelCatalog, - usesModelSpecificAcpEfforts -} from '../../../shared/agent-session/options/agent-model-catalog-ownership' +import { independentAgentModelCatalog } from '../../../shared/agent-session/options/agent-model-catalog-ownership' import type { AgentModelCatalogReadParams, AgentModelCatalogServiceDeps @@ -239,11 +236,11 @@ export function projectAgentModelCatalogSnapshot( ...(!params.savedOnly && !params.verifiedOnly && (deps.store.isListing(bound.fingerprint) || + (resolving && bound.projecting !== undefined) || (resolving && (deps.store.probeDue(bound.fingerprint) || // The picked model's menu is listed by the refresh this read just asked for. - deps.store.efforts.due(bound.fingerprint, params.selectedModel)))) && - (independent || !listed || usesModelSpecificAcpEfforts(params.agent)) + deps.store.efforts.due(bound.fingerprint, params.selectedModel)))) ? { listingInProgress: true as const } : {}), ...(deps.store.failure(bound.fingerprint)?.unavailable 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 221b99de96a..6786abc1b6c 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 @@ -42,9 +42,6 @@ import { export { AGENT_MODEL_CATALOG_FAILURE_TTL_MS, AGENT_MODEL_CATALOG_PICKER_WAIT_MS } export type { AgentModelCatalogFailure } -// Persist successful catalogs only; failures/unavailability are held separately. -// Saved display data may be stale, while launch proof requires fresh discovery in this process. - export const AGENT_MODEL_CATALOG_MAX_ENTRIES = 256 import { @@ -256,7 +253,6 @@ export class AgentModelCatalogStore { evict: () => this.evictOverCap() })) - /** One probe per canonical scope; failures resolve null. Its starter's budget bounds it. */ refresh( fingerprint: string, agent: string, @@ -287,6 +283,8 @@ export class AgentModelCatalogStore { }) } + flushProgress = (key: string): void => this.refreshes.get(key)?.flush() + isListing(fingerprint: string): boolean { const flight = this.refreshes.get(fingerprint) if (flight && !flight.current()) { diff --git a/src/main/native-chat/agent-session-wire/options/structured-agent-session-catalog-reads.ts b/src/main/native-chat/agent-session-wire/options/structured-agent-session-catalog-reads.ts index cbf7642a223..da4f1024c4c 100644 --- a/src/main/native-chat/agent-session-wire/options/structured-agent-session-catalog-reads.ts +++ b/src/main/native-chat/agent-session-wire/options/structured-agent-session-catalog-reads.ts @@ -59,7 +59,7 @@ export function createStructuredAgentSessionCatalogReads( return { origin: 'unknown' } } const request = { ...params, records, operation, deadline: operation.deadline } - if (!params.savedOnly) { + if (!params.savedOnly && !params.waitForListing) { void catalog.demand(request) } return catalog.read(request) diff --git a/src/main/pi/rpc/options/rpc-model-catalog-probe.test.ts b/src/main/pi/rpc/options/rpc-model-catalog-probe.test.ts new file mode 100644 index 00000000000..83731374522 --- /dev/null +++ b/src/main/pi/rpc/options/rpc-model-catalog-probe.test.ts @@ -0,0 +1,86 @@ +import { expect, it, vi, onTestFinished } from 'vitest' +import { createPiModelCatalogProbe } from './rpc-model-catalog-probe' +import type { AgentModelCatalogSuccess } from '../../../native-chat/agent-model-catalog/agent-model-catalog-entry' + +it('publishes a slow inventory independently of state, then keeps the late default', async () => { + vi.useFakeTimers() + onTestFinished(() => vi.useRealTimers()) + const inventory = Promise.withResolvers() + const state = Promise.withResolvers() + const requested = Promise.withResolvers() + const observations: AgentModelCatalogSuccess[] = [] + const probe = createPiModelCatalogProbe({ + resolveEnvironment: async () => ({}), + resolveCommand: () => process.execPath, + probeVersion: async () => ({ version: '0.84.0' }), + homePath: '/account', + connect: () => ({ + request: (command, _params, options) => { + // The real peer's explicit-null contract retains responses past its default 30s wait. + if (options?.timeoutMs !== null) { + return Promise.reject(new Error('catalog request retained its response deadline')) + } + if (command === 'get_available_models') { + requested.resolve() + return inventory.promise + } + return state.promise + }, + close: async () => {} + }) + }) + const result = probe( + { variable: 'PI_CODING_AGENT_DIR', path: '/account' }, + { publish: (success) => observations.push(success) } + ) + await requested.promise + await vi.advanceTimersByTimeAsync(45_000) + inventory.resolve({ + models: [{ id: 'gpt-live', provider: 'openai', name: 'GPT Live', reasoning: false }] + }) + await vi.advanceTimersByTimeAsync(0) + expect(observations[0]).toMatchObject({ models: [{ id: 'openai/gpt-live' }] }) + expect(observations[0].configured).toBeUndefined() + state.resolve({ model: { id: 'gpt-live', provider: 'openai' }, thinkingLevel: 'high' }) + expect(await result).toMatchObject({ + models: [{ id: 'openai/gpt-live' }], + configured: { model: 'openai/gpt-live', effort: 'high' } + }) + expect(observations[0].configured).toBeUndefined() +}) + +it.each(['empty', 'rejected'])( + 'retains delayed sign-out evidence after %s inventory', + async (kind) => { + const state = Promise.withResolvers() + const requested = Promise.withResolvers() + const probe = createPiModelCatalogProbe({ + resolveEnvironment: async () => ({}), + resolveCommand: () => process.execPath, + probeVersion: async () => ({ version: '0.84.0' }), + homePath: '/account', + connect: () => ({ + request: (command) => { + if (command === 'get_state') { + return state.promise + } + requested.resolve() + return kind === 'empty' + ? Promise.resolve({ models: [] }) + : Promise.reject(new Error('inventory unavailable')) + }, + close: async () => {} + }) + }) + let settled = false + const result = probe({ variable: 'PI_CODING_AGENT_DIR', path: '/account' }).finally(() => { + settled = true + }) + const verdict = expect(result).rejects.toMatchObject({ unavailable: { reason: 'notSignedIn' } }) + await requested.promise + await Promise.resolve() + expect(settled).toBe(false) + state.resolve({ model: { id: 'unknown', provider: 'unknown' } }) + await verdict + } +) diff --git a/src/main/pi/rpc/options/rpc-model-catalog-probe.ts b/src/main/pi/rpc/options/rpc-model-catalog-probe.ts index e6e4e2520b6..0d76b60f453 100644 --- a/src/main/pi/rpc/options/rpc-model-catalog-probe.ts +++ b/src/main/pi/rpc/options/rpc-model-catalog-probe.ts @@ -8,9 +8,13 @@ import { homedir } from 'node:os' import { z } from 'zod' import { requireLegacyAgentSessionAccountHome } from '../../../../shared/agent-session/lifecycle/agent-session-account-home' import type { resolveCliCommand } from '../../../../shared/node-cli-command-resolution' -import type { AgentCliVersionReader } from '../../../agent-launch/preflight/agent-cli-version-probe' +import { + readAgentCliVersion, + type AgentCliVersionReader +} from '../../../agent-launch/preflight/agent-cli-version-probe' import { JsonlRpcAgentConnection } from '../../../jsonl-rpc/agent-connection' import type { + AgentModelCatalogProgress, AgentModelCatalogProbe, AgentModelCatalogLookup } from '../../../native-chat/agent-model-catalog/agent-model-catalog-discovery' @@ -87,14 +91,16 @@ export function createPiModelCatalogProbe(deps: PiModelCatalogProbeDeps): AgentM inputStamp: await agentModelCatalogInputStamp({ command, prefixArgs, cwd, env }), run: async ( signal?: AbortSignal, - runOperation = operation + runOperation = operation, + _selectedModels?: readonly string[], + publish?: AgentModelCatalogProgress ): Promise => { const operation = runOperation const options = { signal } assertStructuredAgentResolutionCurrent(operation) const release = await checkPiRpcVersion( { command, prefixArgs, cwd, env, operation }, - deps.probeVersion + deps.probeVersion ?? ((input) => readAgentCliVersion({ ...input, timeoutMs: null })) ) assertStructuredAgentResolutionCurrent(operation) if (release.status === 'tooOld') { @@ -116,35 +122,55 @@ export function createPiModelCatalogProbe(deps: PiModelCatalogProbeDeps): AgentM async (connection): Promise => { let configured: AgentModelCatalogSuccess['configured'] let unknownProvider = false - try { - const state = discoveryStateSchema.safeParse(await connection.request('get_state')) - unknownProvider = state.success && state.data.model?.provider === 'unknown' - if (state.success && state.data.model && !unknownProvider) { - configured = { - model: piModelOptionId(state.data.model.provider, state.data.model.id), - ...(typeof state.data.thinkingLevel === 'string' && state.data.thinkingLevel - ? { effort: state.data.thinkingLevel } - : {}) - } + let stateKnown = false + let models: AgentModelCatalogSuccess['models'] | undefined + const progress = () => { + assertStructuredAgentResolutionCurrent(operation) + if (models?.length) { + publish?.( + { models, origin: 'probe', ...(configured ? { configured } : {}) }, + stateKnown + ) } - } catch { - // Optional default facts cannot prevent a usable inventory from being discovered. } - assertStructuredAgentResolutionCurrent(operation) - options?.signal?.throwIfAborted() - // Pi's own "No models available" condition: no selected provider and nothing listable. - const models = await readPiRpcModelOptions(connection).catch((error: unknown) => { + const state = connection + .request('get_state', undefined, { timeoutMs: null }) + .catch(() => undefined) + .then((response) => { + assertStructuredAgentResolutionCurrent(operation) + const parsed = discoveryStateSchema.safeParse(response) + stateKnown = parsed.success + unknownProvider = parsed.success && parsed.data.model?.provider === 'unknown' + if (parsed.success && parsed.data.model && !unknownProvider) { + configured = { + model: piModelOptionId(parsed.data.model.provider, parsed.data.model.id), + ...(typeof parsed.data.thinkingLevel === 'string' && parsed.data.thinkingLevel + ? { effort: parsed.data.thinkingLevel } + : {}) + } + } + progress() + }) + void state.catch(() => {}) + try { + models = await readPiRpcModelOptions(connection, { timeoutMs: null }) + } catch (error) { + await state if (unknownProvider) { throw new AgentModelCatalogUnavailableError({ reason: 'notSignedIn' }) } throw error - }) + } + progress() + await state + assertStructuredAgentResolutionCurrent(operation) + options.signal?.throwIfAborted() if (unknownProvider && models.length === 0) { throw new AgentModelCatalogUnavailableError({ reason: 'notSignedIn' }) } return { models, origin: 'probe', ...(configured ? { configured } : {}) } }, - { label: 'pi', signal: options?.signal } + { label: 'pi', signal: options?.signal, timeoutMs: null } ) if (success.models.length === 0) { throw new Error('pi listed no models') @@ -155,7 +181,12 @@ export function createPiModelCatalogProbe(deps: PiModelCatalogProbeDeps): AgentM } return Object.assign>( async (home, options) => - (await resolve(home, '', 'recover', options?.operation)).run(options?.signal), + (await resolve(home, '', 'recover', options?.operation)).run( + options?.signal, + options?.operation, + options?.selectedModels, + options?.publish + ), { resolve } ) } diff --git a/src/main/pi/rpc/options/rpc-model-options.ts b/src/main/pi/rpc/options/rpc-model-options.ts index 1c1b0b45c40..4c2936a6201 100644 --- a/src/main/pi/rpc/options/rpc-model-options.ts +++ b/src/main/pi/rpc/options/rpc-model-options.ts @@ -35,9 +35,12 @@ export function parsePiModelOptionId(value: string): { provider: string; modelId /** Native inventory projection shared by running chats and the pre-session catalog. */ export async function readPiRpcModelOptions( - rpc: PiRpcRequester + rpc: PiRpcRequester, + options?: { timeoutMs?: number | null } ): Promise { - const models = modelsSchema.parse(await rpc.request('get_available_models')).models + const models = modelsSchema.parse( + await rpc.request('get_available_models', undefined, options) + ).models return models.map((model) => { const id = piModelOptionId(model.provider, model.id) const efforts = diff --git a/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.test.ts b/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.test.ts index d92cf102a78..33512588f64 100644 --- a/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.test.ts +++ b/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.test.ts @@ -78,6 +78,25 @@ describe('agentSession.modelCatalog', () => { ) }) + it.each([false, true])( + 'negotiates early progress through the existing client capability (enabled=%s)', + async (enabled) => { + record = { ...agentSessionRecordFixture(), provider: 'codex' } + const result = await call( + 'agentSession.modelCatalog', + { agent: 'codex', sessionId: record.sessionId, waitForListing: true }, + { + ...STRUCTURED_CLIENT, + clientCapabilities: DESKTOP_RENDERER_RUNTIME_CLIENT_CAPABILITIES.filter( + (capability) => enabled || capability !== 'agent-session.model-catalog.progress.v1' + ) + } + ) + expect(result.ok, JSON.stringify(result)).toBe(true) + expect(read).toHaveBeenLastCalledWith(expect.objectContaining({ progress: enabled })) + } + ) + it("keeps a moved floating workspace's list apart from the old folder's late answer", async () => { const late = Promise.withResolvers() const listed = (id: string): AgentModelCatalogSuccess => ({ diff --git a/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.ts b/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.ts index df3412bce8d..64de6072066 100644 --- a/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.ts +++ b/src/main/runtime/rpc/methods/structured-sessions/catalog/structured-agent-session-options-read.ts @@ -1,4 +1,7 @@ -import { AGENT_SESSION_COHERENT_MODEL_CATALOG_CAPABILITY } from '../../../../../../shared/agent-session/structured-agent-session-surface-capabilities' +import { + AGENT_SESSION_COHERENT_MODEL_CATALOG_CAPABILITY, + AGENT_SESSION_MODEL_CATALOG_PROGRESS_CAPABILITY +} from '../../../../../../shared/agent-session/structured-agent-session-surface-capabilities' // options reads at-rest records/catalogs or the live provider; modelCatalog reads stored/start-published facts without waiting on attach. // Older hosts return method_not_found (mobile may see forbidden); clients retain static seeds. @@ -113,6 +116,10 @@ export const STRUCTURED_AGENT_SESSION_OPTIONS_READ_METHODS = [ const request = { ...params, ...scope, + progress: + ctx.clientKind === undefined || + ctx.clientCapabilities?.includes(AGENT_SESSION_MODEL_CATALOG_PROGRESS_CAPABILITY) === + true, ...(waits && resolving ? { deadline } : {}), ...(launchDirectory ? { workspacePath: launchDirectory } @@ -140,7 +147,7 @@ export const STRUCTURED_AGENT_SESSION_OPTIONS_READ_METHODS = [ ) } // A floating read lists the folder it resolves to (above), never the unresolved one. - if (!host && !params.savedOnly && !follows) { + if (!waits && !host && !params.savedOnly && !follows) { void catalog.demand(request) } const result = host ? await host.readModelCatalog(request) : await catalog.read(request) diff --git a/src/renderer/src/components/agent/agent-launch-picker-acquisition.test.ts b/src/renderer/src/components/agent/agent-launch-picker-acquisition.test.ts index 61dc3cb3aca..27c29331d57 100644 --- a/src/renderer/src/components/agent/agent-launch-picker-acquisition.test.ts +++ b/src/renderer/src/components/agent/agent-launch-picker-acquisition.test.ts @@ -206,3 +206,45 @@ it('keeps an unflagged saved catalog provisional when its read fails', async () stop() } }) + +it('keeps an open picker subscribed after usable progress until late enrichment arrives', async () => { + const inventory = Promise.withResolvers() + const enriched = Promise.withResolvers() + let next = inventory.promise + let published: AgentPickerEntry[] = [] + const close = observeAgentPickerAcquisition( + 'late-enrichment', + [codex], + { + catalog: async (_entry, wait) => + wait ? next : { origin: 'unknown', listingInProgress: true }, + support: async () => undefined + }, + { + selected: () => 'codex', + publish: (entries) => { + published = entries + } + } + ) + onTestFinished(close) + await vi.waitFor(() => expect(published[0]?.catalog).toMatchObject({ listingInProgress: true })) + next = enriched.promise + inventory.resolve({ origin: 'probe', models: [model('listed')], listingInProgress: true }) + await vi.waitFor(() => + expect(published[0]?.catalog).toMatchObject({ models: [{ id: 'listed' }] }) + ) + expect(published[0]?.catalogRead).toBe('pending') + enriched.resolve({ + origin: 'probe', + models: [ + { ...model('listed'), defaultEffort: 'high', efforts: [{ value: 'high', label: 'High' }] } + ] + }) + await vi.waitFor(() => + expect(published[0]).toMatchObject({ + catalogRead: 'settled', + catalog: { models: [{ id: 'listed', defaultEffort: 'high' }] } + }) + ) +}) diff --git a/src/renderer/src/components/agent/agent-launch-picker-acquisition.ts b/src/renderer/src/components/agent/agent-launch-picker-acquisition.ts index ce07b04ae45..bbc08d9c020 100644 --- a/src/renderer/src/components/agent/agent-launch-picker-acquisition.ts +++ b/src/renderer/src/components/agent/agent-launch-picker-acquisition.ts @@ -190,7 +190,7 @@ class AgentPickerAcquisition { const refresh = !signal.aborted && this.observers.size > 0 && - (waitForListing ? catalog.waitExpired : catalog.listingInProgress) === true + (catalog.waitExpired === true || catalog.listingInProgress === true) if (refresh) { this.queued.refresh.add(index) } diff --git a/src/renderer/src/components/agent/launch-settings/AgentLaunchChoiceEditor.test.tsx b/src/renderer/src/components/agent/launch-settings/AgentLaunchChoiceEditor.test.tsx index f5ae1848217..c59c8e5e215 100644 --- a/src/renderer/src/components/agent/launch-settings/AgentLaunchChoiceEditor.test.tsx +++ b/src/renderer/src/components/agent/launch-settings/AgentLaunchChoiceEditor.test.tsx @@ -78,6 +78,26 @@ describe('AgentLaunchChoiceEditor model list', () => { useAppStore.setState({ settings: initialSettings }) }) + it('shows later catalog updates while Settings stays open after initial progress', async () => { + const inventory = Promise.withResolvers() + const enriched = Promise.withResolvers() + call + .mockResolvedValueOnce({ origin: 'unknown', listingInProgress: true }) + .mockReturnValueOnce(inventory.promise) + .mockReturnValueOnce(enriched.promise) + useAppStore.setState({ settings: getDefaultSettings('') }) + renderEditor() + await act(async () => inventory.resolve({ ...LISTED, listingInProgress: true })) + await waitFor(() => expect(menus.get('Model')).toContain('gpt-6-astra')) + await act(async () => + enriched.resolve({ + ...LISTED, + models: [{ id: 'late-model', label: 'Late model', efforts: [] }] + }) + ) + await waitFor(() => expect(menus.get('Model')).toEqual(['__agent_default__', 'late-model'])) + }) + it('shows the refreshed model after the agent command changes and ignores the older answer', async () => { let answerOlder: (catalog: unknown) => void = () => {} call.mockReturnValueOnce(new Promise((resolve) => (answerOlder = resolve))) diff --git a/src/renderer/src/components/agent/launch-settings/use-agent-settings-model-catalog.ts b/src/renderer/src/components/agent/launch-settings/use-agent-settings-model-catalog.ts index dc38bb37afe..cba609cee1c 100644 --- a/src/renderer/src/components/agent/launch-settings/use-agent-settings-model-catalog.ts +++ b/src/renderer/src/components/agent/launch-settings/use-agent-settings-model-catalog.ts @@ -92,11 +92,10 @@ export function useAgentSettingsModelCatalog( if (stale) { return } - if (wait ? catalog.waitExpired : catalog.listingInProgress) { - read(true) - return - } setAnswer({ key, accountSettings, generation, catalog }) + if (catalog.waitExpired || catalog.listingInProgress) { + read(true) + } }) // Why not the snapshot: an unanswered read can't vouch for an older list. .catch(() => { diff --git a/src/shared/agent-session/structured-agent-session-renderer-client-capabilities.ts b/src/shared/agent-session/structured-agent-session-renderer-client-capabilities.ts index 3c18f0a0f48..2a6ea47beb9 100644 --- a/src/shared/agent-session/structured-agent-session-renderer-client-capabilities.ts +++ b/src/shared/agent-session/structured-agent-session-renderer-client-capabilities.ts @@ -1,6 +1,7 @@ import { AGENT_SESSION_COMMANDS_ON_REQUEST_CAPABILITY, AGENT_SESSION_COHERENT_MODEL_CATALOG_CAPABILITY, + AGENT_SESSION_MODEL_CATALOG_PROGRESS_CAPABILITY, PI_STRUCTURED_DIALOGS_RUNTIME_CAPABILITY, STRUCTURED_AGENT_SESSION_REGISTERED_AGENTS_RUNTIME_CAPABILITY } from './structured-agent-session-surface-capabilities' @@ -36,6 +37,7 @@ export const STRUCTURED_AGENT_SESSION_RENDERER_CLIENT_CAPABILITIES = [ AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, AGENT_SESSION_OPTIONAL_MODEL_CLIENT_CAPABILITY, AGENT_SESSION_COHERENT_MODEL_CATALOG_CAPABILITY, + AGENT_SESSION_MODEL_CATALOG_PROGRESS_CAPABILITY, AGENT_SESSION_CURRENT_REQUESTS_CLIENT_CAPABILITY, // Read cursors settle acknowledged echoes even when snapshots omit their rows. AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, diff --git a/src/shared/agent-session/structured-agent-session-surface-capabilities.ts b/src/shared/agent-session/structured-agent-session-surface-capabilities.ts index c92854e44dc..68c58582555 100644 --- a/src/shared/agent-session/structured-agent-session-surface-capabilities.ts +++ b/src/shared/agent-session/structured-agent-session-surface-capabilities.ts @@ -44,6 +44,10 @@ export const STRUCTURED_AGENT_SESSION_RESUME_WRITER_ADMISSION_RUNTIME_CAPABILITY export const STRUCTURED_AGENT_SESSION_AVAILABILITY_RUNTIME_CAPABILITY = 'agent-session.structured.availability.v1' as const +// Progress-aware readers renew on listingInProgress before their wait expires. +export const AGENT_SESSION_MODEL_CATALOG_PROGRESS_CAPABILITY = + 'agent-session.model-catalog.progress.v1' as const + export const AGENT_SESSION_COHERENT_MODEL_CATALOG_CAPABILITY = 'agent-session.coherent-model-catalog.v1' as const