Keep slow model discovery alive beyond picker waits

This commit is contained in:
Brennan Benson
2026-11-13 13:16:13 -08:00
parent 5fe7308e99
commit 1f9ba76726
42 changed files with 1922 additions and 544 deletions
@@ -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<Awaited<ReturnType<AcpModelCatalogSessionConnection['start']>>>()
const late =
Promise.withResolvers<
Awaited<ReturnType<AcpModelCatalogSessionConnection['setConfigOption']>>
>()
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<ReturnType<typeof readSessionModels>>
>(
() => 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<ReturnType<AcpModelCatalogSessionConnection['setConfigOption']>>
>()
const observations: Awaited<ReturnType<typeof readSessionModels>>[] = []
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<string, string>) => 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<ReturnType<AcpModelCatalogSessionConnection['setConfigOption']>>
>
}[] = []
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<ReturnType<AcpModelCatalogSessionConnection['setConfigOption']>>
>()
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<typeof createAgentModelCatalogSuccessWriter> },
'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<void>()
const base = acpLaunchSpecFor('opencode')!
const session = vi.fn()
const command = vi.fn<typeof runAgentModelCatalogListing>(
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 })
})
@@ -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<AgentModelCatalogProgress['demanded']>
}
): 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<string>()
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 }
@@ -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' })
@@ -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'])
})
@@ -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<unknown>()
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)
})
@@ -19,6 +19,7 @@ export type AgentCliVersionCheck =
export type AgentCliVersionInput = Pick<ProcessSpec, 'program' | 'cwd' | 'env'> & {
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
})
@@ -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<string, unknown>,
env: Readonly<Record<string, string>>,
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 } }
: {})
}
}
@@ -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<void>()
const signin = Promise.withResolvers<void>()
const inventory = Promise.withResolvers<void>()
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()
})
@@ -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<AgentModelCatalogSuccess> => {
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<string>()
const replies = new Map<string, unknown>()
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)
@@ -6,7 +6,10 @@ function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
}
export function readClaudeControlResponses(stdout: string): Map<string, unknown> {
export function readClaudeControlResponses(
stdout: string,
settled?: (requestId: string) => void
): Map<string, unknown> {
const responses = new Map<string, unknown>()
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<string, unknown>
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)
}
@@ -25,7 +25,7 @@ export type CodexAppServerInvocation = {
env?: Record<string, string>
/** 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<T>(
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()
@@ -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<unknown>()
const account = Promise.withResolvers<unknown>()
const inventoried = Promise.withResolvers<void>()
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<never>(() => {}))
const account = Promise.withResolvers<unknown>()
const inventory = Promise.withResolvers<void>()
const closing = Promise.withResolvers<void>()
const closed = Promise.withResolvers<void>()
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<unknown>()
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)
}
)
@@ -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<AgentModelCatalogSuccess> => {
assertStructuredAgentResolutionCurrent(runOperation)
signal?.throwIfAborted()
deps.prepareHome?.(accountHomePath)
const run = deps.runSession ?? runCodexAppServerSession
const successFromListing = (
listing: Awaited<ReturnType<typeof fetchCodexModelCatalogListing>>,
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<ReturnType<typeof fetchCodexModelCatalogListing>> | undefined
let accountKnown = false
let unavailable: AgentModelCatalogSuccess['unavailable']
const observe = (next: NonNullable<typeof latest>) => {
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,
@@ -154,16 +154,17 @@ export async function listCodexModelPages(
export async function fetchCodexModelCatalogListing(input: {
connection: Pick<CodexAppServerConnection, 'request'>
timeoutMs?: number
timeoutMs?: number | null
deadlineMs?: number
cwd?: string
operation?: StructuredAgentResolutionOperation
publish?: (listing: CodexModelCatalogListing) => void
}): Promise<CodexModelCatalogListing> {
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(),
@@ -17,12 +17,12 @@ function configuredText(value: unknown): string | null {
export async function readCodexConfiguredLaunchDefaults(
connection: Pick<CodexAppServerConnection, 'request'>,
timeoutMs = CONFIG_READ_TIMEOUT_MS,
timeoutMs: number | null = CONFIG_READ_TIMEOUT_MS,
cwd?: string
): Promise<CodexConfiguredLaunchDefaults | null> {
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
@@ -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
}
}
@@ -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<string>
}
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<AgentModelCatalogSuccess>
}
@@ -24,6 +34,7 @@ export type AgentModelCatalogProbe = ((
cwd?: string
selectedModels?: readonly string[]
operation?: StructuredAgentResolutionOperation
publish?: AgentModelCatalogProgress
}
) => Promise<AgentModelCatalogSuccess>) & {
resolveLegacy?: (
@@ -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<typeof writeAgentModelCatalogSuccess>[0]
): ReturnType<typeof createAgentModelCatalogEntryFactory> {
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
)
}
@@ -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<AgentModelCatalogSuccess>
/** 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<AgentModelCatalogEntry | null>
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<AgentModelCatalogEntry | null> {
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<string>()
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
}
}
}
@@ -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<AgentModelCatalogEntry | null> | 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<void> {
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<void> => {
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
@@ -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 ??
@@ -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<string> {
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<unknown> }
export async function runAgentModelCatalogSession<C extends AgentModelCatalogProbeConnection, T>(
open: () => C,
body: (connection: C, ready: AgentModelCatalogReady<T>) => Promise<T>,
options: { label: string; timeoutMs?: number; signal?: AbortSignal }
options: { label: string; timeoutMs?: number | null; signal?: AbortSignal }
): Promise<T> {
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<typeof setTimeout> | undefined
@@ -218,7 +246,7 @@ export async function runAgentModelCatalogSession<C extends AgentModelCatalogPro
}
}
options.signal?.addEventListener('abort', onAbort, { once: true })
timer = setTimeout(expire, timeoutMs)
timer = timeoutMs === null ? undefined : setTimeout(expire, timeoutMs)
})
try {
return await Promise.race([body(connection, ready), deadline])
@@ -1,3 +1,4 @@
import { resolveAgentModelCatalogAccount } from './agent-model-catalog-account-resolution'
import type { PrioritySemaphore } from '../../../shared/priority-semaphore'
import { independentAgentModelCatalog } from '../../../shared/agent-session/options/agent-model-catalog-ownership'
import type {
@@ -5,16 +6,12 @@ import type {
AgentModelCatalogServiceDeps
} from './agent-model-catalog-service-contract'
import type { AgentModelCatalogReadRecords } from './agent-model-catalog-running-account'
import { recordCatalogKey } from './agent-model-catalog-fingerprint'
import { newChatCatalogKey } from './agent-model-catalog-prewarm'
import {
agentModelCatalogDefaultFollowsAccount,
agentModelCatalogListedDefault,
catalogFolder,
sessionWorkspacePath
agentModelCatalogListedDefault
} from './agent-model-catalog-workspace'
import { resolveAgentModelCatalogScope } from './agent-model-catalog-scope'
import { scopedAgentModelCatalogLister } from './agent-model-catalog-lister'
import { scopedAgentModelCatalogLister } from './agent-model-catalog-scoped-lister'
import { seedAccountList } from './agent-model-catalog-account-list'
import type { AgentModelCatalogEntry } from './agent-model-catalog-entry'
import {
@@ -22,10 +19,7 @@ import {
bindAgentModelCatalogScope,
type AgentModelCatalogBinding
} from './agent-model-catalog-snapshot'
import { AGENT_MODEL_CATALOG_PICKER_WAIT_MS } from './agent-model-catalog-listing-waiters'
/** Bounds scope resolution and the folder's settings; listings own their execution budget. */
const AGENT_MODEL_CATALOG_REFRESH_MS = AGENT_MODEL_CATALOG_PICKER_WAIT_MS
import type { AgentModelCatalogReadOperation } from './agent-model-catalog-operation'
/** What a refresh resolves, captured when it starts; no reader's session, deadline or callback. */
export type AgentModelCatalogLaunch = {
@@ -48,6 +42,7 @@ export type AgentModelCatalogRefresh = {
defaults: Promise<void>
/** Its listing ingested and the account's list seeded, or it ended. */
done: Promise<void>
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<null>((resolve) => {
signal.addEventListener('abort', () => resolve(null), { once: true })
})
@@ -95,10 +85,12 @@ export function createAgentModelCatalogRefreshes(context: {
const listing = Promise.withResolvers<void>()
const defaults = Promise.withResolvers<void>()
const targets = new Set<string>()
let list: ((model: string | undefined) => void) | undefined
const holders = new Set<AbortSignal>()
let register: ((holder?: AbortSignal) => void) | undefined
let demand: ((model: string) => void) | undefined
let projecting: Promise<void> | undefined
const run = async (): Promise<void> => {
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
}
}
@@ -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
}
}
}
@@ -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. */
@@ -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
@@ -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<AgentModelCatalogSuccess>((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<void>()
const closing = Promise.withResolvers<void>()
const answer = Promise.withResolvers<AgentModelCatalogSuccess>()
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<AgentModelCatalogSuccess>(() => {})
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<AgentModelCatalogSuccess>()
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<AgentModelCatalogSuccess>()
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<void>()
const closing = Promise.withResolvers<void>()
const closed = Promise.withResolvers<void>()
const answer = Promise.withResolvers<AgentModelCatalogSuccess>()
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)
}
)
@@ -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<void>()
const gate = Promise.withResolvers<void>()
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>()
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<void>()
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<void>()
if (delayedScope) {
h.folder.gate = scope
}
const registrations = vi.spyOn(h.store, 'refresh')
const submissions = vi.spyOn(h.store.efforts, 'demand')
const readers: Promise<unknown>[] = []
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)
}
}
)
@@ -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
@@ -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()) {
@@ -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)
@@ -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<unknown>()
const state = Promise.withResolvers<unknown>()
const requested = Promise.withResolvers<void>()
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<unknown>()
const requested = Promise.withResolvers<void>()
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
}
)
@@ -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<AgentModelCatalogSuccess> => {
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<AgentModelCatalogSuccess> => {
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<AgentModelCatalogProbe, Pick<AgentModelCatalogProbe, 'resolve'>>(
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 }
)
}
+5 -2
View File
@@ -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<AgentSessionOptionsResult['models']> {
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 =
@@ -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<AgentModelCatalogSuccess>()
const listed = (id: string): AgentModelCatalogSuccess => ({
@@ -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)
@@ -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<AgentPickerEntry['catalog']>()
const enriched = Promise.withResolvers<AgentPickerEntry['catalog']>()
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' }] }
})
)
})
@@ -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)
}
@@ -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<unknown>()
const enriched = Promise.withResolvers<unknown>()
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)))
@@ -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(() => {
@@ -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,
@@ -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