diff --git a/docs/custom-providers.md b/docs/custom-providers.md index 17bd1ceeb..acf882657 100644 --- a/docs/custom-providers.md +++ b/docs/custom-providers.md @@ -2,7 +2,7 @@ Paseo supports configuring custom agent providers through `config.json` (located at `$PASEO_HOME/config.json`, typically `~/.paseo/config.json`). You can extend built-in providers with different API backends, add ACP-compatible agents, set custom binaries, disable providers, and create multiple profiles for the same underlying provider. -All provider configuration lives under `agents.providers` in config.json: +Provider definitions live under `agents.providers` in config.json: ```json { @@ -17,6 +17,20 @@ All provider configuration lives under `agents.providers` in config.json: Provider IDs must be lowercase alphanumeric with hyphens (`/^[a-z][a-z0-9-]*$/`). +Each provider catalog refresh waits up to 60 seconds. If a provider loads many plugins or a large +agent catalog during startup, raise the limit in milliseconds: + +```json +{ + "agents": { + "catalogRefreshTimeoutMs": 180000 + } +} +``` + +The limit applies independently to every provider refresh and covers availability plus the entire +catalog probe. `PASEO_PROVIDER_REFRESH_TIMEOUT_MS` sets it when the config field is absent. + --- ## Table of Contents diff --git a/docs/providers.md b/docs/providers.md index b76aa3d89..4e199e3b0 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -105,6 +105,13 @@ Daemon bootstrap reconciles that ledger in the background, without blocking star The daemon keeps provider snapshots per resolved working directory, with a separate semantic global scope for settings/provider management and requests that do not carry a cwd. Provider catalog probes receive a discriminated `FetchCatalogOptions`: `{ scope: "global", force }` for global catalog refreshes, or `{ scope: "workspace", cwd, force }` for project-scoped refreshes. Providers decide what global means for their runtime; do not infer global by comparing a cwd to the user's home directory. +`ProviderSnapshotManager` owns one refresh deadline per provider. The deadline starts before the +availability check and covers that check plus the complete catalog probe. Providers that make +multiple catalog requests must not apply this deadline separately to each request. The manager +aborts the shared refresh signal at the deadline. Providers name active catalog operations and +finish subprocess, server, or session cleanup before rejecting. Timeout errors list the operations +that were still active when the deadline expired. + Snapshot reads may probe providers only while the requested cwd scope is cold. Once an entry is warm, its `ready`, `error`, or `unavailable` state stays cached until an explicit refresh. Do not add TTL revalidation, focus-triggered refreshes, selector-open refreshes, or config-reload refreshes. Selector-open refetches may read an already-loading or stale React Query, but they must not force provider probing on their own. Capable clients receive a compact, content-addressed snapshot. Model rows derive their provider from the containing entry and reference snapshot-level thinking sets. The app persists that compact shape per server and cwd, then sends its hash on the next pull; an unchanged response carries no catalog body. Keep the legacy encoding for clients without the capability. The hash covers the complete client-visible compact snapshot, including status and `fetchedAt`, so explicit refreshes invalidate it even when the discovered catalog is otherwise equal. diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index cea5a9f1b..686c1b24a 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -668,15 +668,19 @@ export type FetchCatalogOptions = | { scope: "global"; force: boolean; - timeoutMs?: number; } | { scope: "workspace"; cwd: string; force: boolean; - timeoutMs?: number; }; +export interface ProviderRefreshContext { + readonly signal: AbortSignal; + /** Track an upstream operation so timeout errors identify the work still pending. */ + runActivity(name: string, operation: () => Promise): Promise; +} + export interface ProviderCatalog { models: AgentModelDefinition[]; modes: AgentMode[]; @@ -686,6 +690,7 @@ export interface ProviderCatalog { export interface ResolveAgentDefaultModeInput { config: AgentSessionConfig; env?: Record; + signal?: AbortSignal; } export interface AgentClient { @@ -707,8 +712,13 @@ export interface AgentClient { * process, separate upstream calls, static modes, or private helpers; callers * outside the provider do not get separate runtime model/mode probes. * The registry is responsible for merging configured model overrides. + * ProviderSnapshotManager supplies a shared context. Providers must pass its + * signal downstream and finish resource cleanup before rejecting on abort. */ - fetchCatalog(options: FetchCatalogOptions): Promise; + fetchCatalog( + options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise; /** Apply provider-owned defaults to a model supplied through provider configuration. */ resolveConfiguredModel?(model: AgentModelDefinition): AgentModelDefinition; resolveDefaultModeId?(input: ResolveAgentDefaultModeInput): Promise; @@ -727,7 +737,7 @@ export interface AgentClient { * Check if this provider is available (CLI binary is installed). * Returns true if available, false otherwise. */ - isAvailable(): Promise; + isAvailable(signal?: AbortSignal): Promise; getDiagnostic?(): Promise<{ diagnostic: string }>; /** * Archive a durable native session (best-effort). Runtime release belongs to AgentSession.close(). diff --git a/packages/server/src/server/agent/provider-refresh-deadline.ts b/packages/server/src/server/agent/provider-refresh-deadline.ts new file mode 100644 index 000000000..12528cfe8 --- /dev/null +++ b/packages/server/src/server/agent/provider-refresh-deadline.ts @@ -0,0 +1,80 @@ +import type { ProviderRefreshContext } from "./agent-sdk-types.js"; + +interface RunProviderRefreshOptions { + label: string; + timeoutMs: number; + operation: (context: ProviderRefreshContext) => Promise; +} + +export function runProviderRefreshActivity( + context: ProviderRefreshContext | undefined, + name: string, + operation: () => Promise, +): Promise { + return context ? context.runActivity(name, operation) : operation(); +} + +export async function raceProviderRefreshAbort( + signal: AbortSignal | undefined, + operation: Promise, +): Promise { + if (!signal) return await operation; + signal.throwIfAborted(); + let handleAbort: (() => void) | undefined; + const aborted = new Promise((_resolve, reject) => { + handleAbort = () => reject(signal.reason); + signal.addEventListener("abort", handleAbort, { once: true }); + }); + try { + return await Promise.race([operation, aborted]); + } finally { + if (handleAbort) signal.removeEventListener("abort", handleAbort); + } +} + +export async function runProviderRefreshWithDeadline( + options: RunProviderRefreshOptions, +): Promise { + const controller = new AbortController(); + const activityCounts = new Map(); + let timeoutError: Error | undefined; + + const context: ProviderRefreshContext = { + signal: controller.signal, + async runActivity(name, operation) { + controller.signal.throwIfAborted(); + activityCounts.set(name, (activityCounts.get(name) ?? 0) + 1); + try { + const result = await operation(); + controller.signal.throwIfAborted(); + return result; + } finally { + const remaining = (activityCounts.get(name) ?? 1) - 1; + if (remaining === 0) { + activityCounts.delete(name); + } else { + activityCounts.set(name, remaining); + } + } + }, + }; + + const timer = setTimeout(() => { + const pending = Array.from(activityCounts.keys()); + const suffix = pending.length > 0 ? `; pending: ${pending.join(", ")}` : ""; + timeoutError = new Error( + `Timed out refreshing ${options.label} after ${options.timeoutMs}ms${suffix}`, + ); + controller.abort(timeoutError); + }, options.timeoutMs); + + try { + const result = await options.operation(context); + if (timeoutError) throw timeoutError; + return result; + } catch (error) { + throw timeoutError ?? error; + } finally { + clearTimeout(timer); + } +} diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index b673e2bd7..58e45d37c 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -13,6 +13,7 @@ import type { AgentSession, AgentStreamEvent, FetchCatalogOptions, + ProviderRefreshContext, ProviderCatalog, ResolveAgentCreateConfigInput, ResolveAgentCreateConfigResult, @@ -24,6 +25,7 @@ import { resolveDefaultAgentCreateConfig, } from "./create-agent-mode.js"; import { normalizeAgentModelDefinition } from "./agent-sdk-types.js"; +import { runProviderRefreshActivity } from "./provider-refresh-deadline.js"; import type { WorkspaceGitService } from "../workspace-git-service.js"; import type { ManagedProcessRegistry } from "../managed-processes/managed-processes.js"; import type { @@ -92,7 +94,11 @@ export interface ProviderDefinition extends AgentProviderDefinition { * Single catalog discovery call used by ProviderSnapshotManager. Should spawn * at most one provider runtime process and return both models and modes. */ - fetchCatalog: (options: FetchCatalogOptions, client?: AgentClient) => Promise; + fetchCatalog: ( + options: FetchCatalogOptions, + client?: AgentClient, + context?: ProviderRefreshContext, + ) => Promise; } export interface BuildProviderRegistryOptions { @@ -507,8 +513,8 @@ function wrapClientProvider( options, ), ), - fetchCatalog: async (options) => { - const catalog = await inner.fetchCatalog(options); + fetchCatalog: async (options, context) => { + const catalog = await inner.fetchCatalog(options, context); return { ...catalog, models: mergeModels(provider, profileModels, additionalModels, catalog.models, { @@ -518,10 +524,11 @@ function wrapClientProvider( }; }, resolveDefaultModeId: inner.resolveDefaultModeId - ? async ({ config, env }: ResolveAgentDefaultModeInput) => + ? async ({ config, env, signal }: ResolveAgentDefaultModeInput) => await inner.resolveDefaultModeId?.({ config: { ...config, provider: inner.provider }, env, + signal, }) : undefined, resolveCreateConfig: inner.resolveCreateConfig?.bind(inner), @@ -561,7 +568,7 @@ function wrapClientProvider( }; } : undefined, - isAvailable: () => inner.isAvailable(), + isAvailable: (signal) => inner.isAvailable(signal), getDiagnostic: inner.getDiagnostic?.bind(inner), }; } @@ -621,7 +628,11 @@ function createRegistryEntry( resolveCreateConfig: modelClient.resolveCreateConfig ?? resolveDefaultAgentCreateConfig, isCreateConfigUnattended: modelClient.isCreateConfigUnattended ?? isDefaultAgentCreateConfigUnattended, - fetchCatalog: async (options: FetchCatalogOptions, client?: AgentClient) => { + fetchCatalog: async ( + options: FetchCatalogOptions, + client?: AgentClient, + context?: ProviderRefreshContext, + ) => { const catalogClient = client ?? modelClient; if (hasReplacementModels) { // Replacement models skip runtime model discovery, but additionalModels @@ -629,23 +640,29 @@ function createRegistryEntry( // the single catalog API; otherwise use static/empty modes with no runtime. const models = mergeModelAdditions(provider, replacementModels, additionalModels); if (hasStaticModes) { - const defaultModeId = await catalogClient.resolveDefaultModeId?.({ - config: { - provider, - cwd: options.scope === "workspace" ? options.cwd : process.cwd(), - }, - }); + const defaultModeId = await runProviderRefreshActivity( + context, + "default-mode", + async () => + await catalogClient.resolveDefaultModeId?.({ + config: { + provider, + cwd: options.scope === "workspace" ? options.cwd : process.cwd(), + }, + signal: context?.signal, + }), + ); return { models, modes: decorateModes(resolved.definition.modes), defaultModeId, }; } - const catalog = await catalogClient.fetchCatalog(options); + const catalog = await catalogClient.fetchCatalog(options, context); return { ...catalog, models, modes: decorateModes(catalog.modes) }; } - const catalog = await catalogClient.fetchCatalog(options); + const catalog = await catalogClient.fetchCatalog(options, context); return { ...catalog, models: mergeModels(provider, profileModels, additionalModels, catalog.models, { diff --git a/packages/server/src/server/agent/provider-snapshot-manager.test.ts b/packages/server/src/server/agent/provider-snapshot-manager.test.ts index b2c006278..ed67290b0 100644 --- a/packages/server/src/server/agent/provider-snapshot-manager.test.ts +++ b/packages/server/src/server/agent/provider-snapshot-manager.test.ts @@ -9,6 +9,7 @@ import type { AgentModelDefinition, AgentProvider, FetchCatalogOptions, + ProviderRefreshContext, ResolveAgentCreateConfigInput, } from "./agent-sdk-types.js"; import type { ManagedAgent } from "./agent-manager.js"; @@ -68,6 +69,41 @@ async function withEnv(key: string, value: string, run: () => Promise): Pr } } +function waitUntilAborted(signal?: AbortSignal): Promise { + return new Promise((_resolve, reject) => { + signal?.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); +} + +function waitForDelay(delayMs: number): Promise { + return new Promise((finish) => setTimeout(finish, delayMs)); +} + +function waitForAbortWithCleanup( + signal: AbortSignal, + cleanupState: { cleanedUp: boolean }, +): Promise { + return new Promise((_resolve, reject) => { + const handleAbort = () => { + cleanupState.cleanedUp = true; + reject(signal.reason); + }; + signal.addEventListener("abort", handleAbort, { once: true }); + }); +} + +async function runTestCatalogActivities( + context: ProviderRefreshContext, + cleanupState: { cleanedUp: boolean }, +): Promise { + const waitForAgents = () => waitForDelay(50); + const waitForProviders = () => waitForAbortWithCleanup(context.signal, cleanupState); + await Promise.all([ + context.runActivity("app.agents", waitForAgents), + context.runActivity("provider.list", waitForProviders), + ]); +} + describe("ProviderSnapshotManager public surface", () => { test("validates complete Hub agent configurations through the current provider contract", async () => { const manager = new ProviderSnapshotManager({ @@ -380,7 +416,7 @@ describe("ProviderSnapshotManager public surface", () => { test("refreshTimeoutMs option overrides the default and yields a timeout error", async () => { // never-resolving isAvailable forces the timeout path - const isAvailable = vi.fn(() => new Promise(() => {})); + const isAvailable = vi.fn(waitUntilAborted); const manager = new ProviderSnapshotManager({ logger: createTestLogger(), refreshTimeoutMs: 1, @@ -406,9 +442,158 @@ describe("ProviderSnapshotManager public surface", () => { } }); + test("one refresh timeout covers availability and catalog discovery", async () => { + vi.useFakeTimers(); + const manager = new ProviderSnapshotManager({ + logger: createTestLogger(), + refreshTimeoutMs: 100, + extraClients: { + codex: createExtraClient("codex", { + isAvailable: () => + new Promise((settle) => { + setTimeout(() => settle(true), 60); + }), + fetchCatalog: () => + new Promise((settle) => { + setTimeout(() => settle({ models: [], modes: [] }), 60); + }), + }), + }, + }); + + try { + const entryPromise = manager.getProvider({ + cwd: "/tmp/project", + provider: "codex", + wait: true, + }); + + await vi.advanceTimersByTimeAsync(120); + + await expect(entryPromise).resolves.toMatchObject({ + provider: "codex", + status: "error", + error: "Timed out refreshing Codex after 100ms", + }); + } finally { + manager.destroy(); + vi.useRealTimers(); + } + }); + + test("timeout names pending catalog activities, aborts them, and waits for cleanup", async () => { + vi.useFakeTimers(); + let attempt = 0; + const cleanupState = { cleanedUp: false }; + const fetchCatalog = vi.fn( + async (_options: FetchCatalogOptions, context?: ProviderRefreshContext) => { + attempt += 1; + if (attempt > 1) { + expect(cleanupState.cleanedUp).toBe(true); + return { models: [], modes: [] }; + } + + if (!context) throw new Error("missing refresh context"); + await runTestCatalogActivities(context, cleanupState); + return { models: [], modes: [] }; + }, + ); + const manager = new ProviderSnapshotManager({ + logger: createTestLogger(), + refreshTimeoutMs: 100, + extraClients: { + codex: createExtraClient("codex", { + isAvailable: async () => true, + fetchCatalog, + }), + }, + }); + + try { + const first = manager.getProvider({ + cwd: "/tmp/project", + provider: "codex", + wait: true, + }); + await vi.advanceTimersByTimeAsync(100); + await expect(first).resolves.toMatchObject({ + status: "error", + error: "Timed out refreshing Codex after 100ms; pending: provider.list", + }); + expect(cleanupState.cleanedUp).toBe(true); + + await manager.refreshSnapshotForCwd({ cwd: "/tmp/project", providers: ["codex"] }); + expect(fetchCatalog).toHaveBeenCalledTimes(2); + expect( + manager.getSnapshot("/tmp/project").find((entry) => entry.provider === "codex"), + ).toMatchObject({ provider: "codex", status: "ready" }); + } finally { + manager.destroy(); + vi.useRealTimers(); + } + }); + + test("each provider refresh gets its own timeout", async () => { + vi.useFakeTimers(); + const neverReturnsCatalog = (_options: FetchCatalogOptions, context?: ProviderRefreshContext) => + new Promise((_resolve, reject) => { + context?.signal.addEventListener("abort", () => reject(context.signal.reason), { + once: true, + }); + }); + const manager = new ProviderSnapshotManager({ + logger: createTestLogger(), + refreshTimeoutMs: 100, + extraClients: { + codex: createExtraClient("codex", { + isAvailable: async () => true, + fetchCatalog: neverReturnsCatalog, + }), + claude: createExtraClient("claude", { + isAvailable: async () => true, + fetchCatalog: neverReturnsCatalog, + }), + }, + }); + + try { + const codexPromise = manager.getProvider({ + cwd: "/tmp/project", + provider: "codex", + wait: true, + }); + await vi.advanceTimersByTimeAsync(50); + + const claudePromise = manager.getProvider({ + cwd: "/tmp/project", + provider: "claude", + wait: true, + }); + await vi.advanceTimersByTimeAsync(50); + + await expect(codexPromise).resolves.toMatchObject({ + provider: "codex", + status: "error", + }); + await expect( + manager.getProvider({ cwd: "/tmp/project", provider: "claude", wait: false }), + ).resolves.toMatchObject({ provider: "claude", status: "loading" }); + + await vi.advanceTimersByTimeAsync(50); + + await expect(claudePromise).resolves.toMatchObject({ + provider: "claude", + status: "error", + }); + } finally { + manager.destroy(); + vi.useRealTimers(); + } + }); + test("PASEO_PROVIDER_REFRESH_TIMEOUT_MS env var is honored when no option is given", async () => { vi.stubEnv("PASEO_PROVIDER_REFRESH_TIMEOUT_MS", "1"); - const isAvailable = vi.fn(() => new Promise(() => {})); + const isAvailable = vi.fn(waitUntilAborted); const manager = new ProviderSnapshotManager({ logger: createTestLogger(), providerOverrides: { @@ -435,7 +620,7 @@ describe("ProviderSnapshotManager public surface", () => { test("PASEO_PROVIDER_REFRESH_TIMEOUT_MS env var is ignored when option is provided", async () => { vi.stubEnv("PASEO_PROVIDER_REFRESH_TIMEOUT_MS", "1"); - const isAvailable = vi.fn(() => new Promise(() => {})); + const isAvailable = vi.fn(waitUntilAborted); const manager = new ProviderSnapshotManager({ logger: createTestLogger(), refreshTimeoutMs: 5, @@ -708,7 +893,10 @@ describe("ProviderSnapshotManager public surface", () => { extraClients: { codex: createExtraClient("codex", { isAvailable: async () => true, - fetchCatalog: async () => new Promise(() => {}), + fetchCatalog: async (_options, context) => { + await context?.runActivity("model/list", () => waitUntilAborted(context.signal)); + return { models: [], modes: [] }; + }, getDiagnostic: async () => { diagnosticStarted = true; return { diagnostic: "codex diagnostics available" }; @@ -751,9 +939,10 @@ describe("ProviderSnapshotManager public surface", () => { extraClients: { codex: createExtraClient("codex", { isAvailable: async () => true, - fetchCatalog: async () => { + fetchCatalog: async (_options, context) => { snapshotStarted = true; - return new Promise(() => {}); + await context?.runActivity("model/list", () => waitUntilAborted(context.signal)); + return { models: [], modes: [] }; }, getDiagnostic: async () => { diagnosticStarted = true; diff --git a/packages/server/src/server/agent/provider-snapshot-manager.ts b/packages/server/src/server/agent/provider-snapshot-manager.ts index 7b9033bf2..60f7850a6 100644 --- a/packages/server/src/server/agent/provider-snapshot-manager.ts +++ b/packages/server/src/server/agent/provider-snapshot-manager.ts @@ -16,6 +16,10 @@ import { type FetchCatalogOptions, type ProviderSnapshotEntry, } from "./agent-sdk-types.js"; +import { + raceProviderRefreshAbort, + runProviderRefreshWithDeadline, +} from "./provider-refresh-deadline.js"; import type { ManagedAgent } from "./agent-manager.js"; import type { WorkspaceGitService } from "../workspace-git-service.js"; import type { ManagedProcessRegistry } from "../managed-processes/managed-processes.js"; @@ -42,6 +46,7 @@ import { } from "./agent-configuration-validator.js"; const DEFAULT_REFRESH_TIMEOUT_MS = 60_000; +const MAX_REFRESH_TIMEOUT_MS = 2_147_483_647; const DEFAULT_DIAGNOSTIC_TIMEOUT_MS = 120_000; const REFRESH_TIMEOUT_ENV_VAR = "PASEO_PROVIDER_REFRESH_TIMEOUT_MS"; export const GLOBAL_PROVIDER_SNAPSHOT_KEY = "paseo:global"; @@ -50,14 +55,19 @@ export const GLOBAL_PROVIDER_SNAPSHOT_KEY = "paseo:global"; // `copilot --acp` invocation, OpenCode workspace probes with many MCP servers). // Allow operators to bump the ceiling via env var without rebuilding. function resolveRefreshTimeoutMs(option: number | undefined): number { - if (typeof option === "number" && Number.isFinite(option) && option > 0) { + if ( + typeof option === "number" && + Number.isSafeInteger(option) && + option > 0 && + option <= MAX_REFRESH_TIMEOUT_MS + ) { return option; } const fromEnv = process.env[REFRESH_TIMEOUT_ENV_VAR]; if (fromEnv) { // Number() handles scientific notation (e.g. "6e4") which parseInt would silently truncate. const parsed = Number(fromEnv); - if (Number.isFinite(parsed) && parsed > 0) { + if (Number.isSafeInteger(parsed) && parsed > 0 && parsed <= MAX_REFRESH_TIMEOUT_MS) { return parsed; } } @@ -515,7 +525,7 @@ export class ProviderSnapshotManager { client.resolveCreateConfig?.bind(client) ?? definition.resolveCreateConfig, isCreateConfigUnattended: client.isCreateConfigUnattended?.bind(client) ?? definition.isCreateConfigUnattended, - fetchCatalog: client.fetchCatalog.bind(client), + fetchCatalog: (options, _client, context) => client.fetchCatalog(options, context), }; } @@ -834,23 +844,26 @@ export class ProviderSnapshotManager { } const client = this.ensureClient(provider, definition); - const available = await withTimeout( - client.isAvailable(), - this.refreshTimeoutMs, - `Timed out checking ${definition.label} availability after ${this.refreshTimeoutMs}ms`, - ); - if (!available) { + const catalog = await runProviderRefreshWithDeadline({ + label: definition.label, + timeoutMs: this.refreshTimeoutMs, + operation: async (context) => { + const available = await context.runActivity("availability", () => + raceProviderRefreshAbort(context.signal, client.isAvailable(context.signal)), + ); + if (!available) { + return null; + } + + const catalogOptions = createFetchCatalogOptions(catalogScope, force); + return await definition.fetchCatalog(catalogOptions, client, context); + }, + }); + if (!catalog) { setEntry({ ...base, status: "unavailable", enabled: true }); return; } - const catalogOptions = createFetchCatalogOptions(catalogScope, force); - const catalog = await withTimeout( - definition.fetchCatalog({ ...catalogOptions, timeoutMs: this.refreshTimeoutMs }, client), - this.refreshTimeoutMs, - `Timed out refreshing ${definition.label} after ${this.refreshTimeoutMs}ms`, - ); - setEntry({ ...base, defaultModeId: diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 2296aa396..1a43ed553 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -85,6 +85,7 @@ import { type AgentTimelineItem, type AgentUsage, type FetchCatalogOptions, + type ProviderRefreshContext, type ImportableProviderSession, type ImportProviderSessionContext, type ImportProviderSessionInput, @@ -96,6 +97,10 @@ import { type ToolCallDetail, type ToolCallTimelineItem, } from "../agent-sdk-types.js"; +import { + raceProviderRefreshAbort, + runProviderRefreshActivity, +} from "../provider-refresh-deadline.js"; import { isDefaultAgentCreateConfigUnattended, resolveDefaultAgentCreateConfig, @@ -267,7 +272,6 @@ export function buildACPClientCapabilities( // sign-in URL in the browser) when probing an ACP agent for models/modes. // NO_BROWSER is honored by Gemini CLI; other ACP agents ignore it. const PROBE_ENV: Record = { NO_BROWSER: "true" }; -const ACP_CATALOG_TIMEOUT_MS = 60_000; const ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS = 20_000; function summarizeMalformedACPStdoutError(error: unknown): { type: string; message: string } { @@ -934,66 +938,83 @@ export class ACPAgentClient implements AgentClient { return session; } - async fetchCatalog(options: FetchCatalogOptions): Promise { + async fetchCatalog( + options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise { const cwd = options.scope === "global" ? homedir() : options.cwd; - const timeoutMs = options.timeoutMs ?? ACP_CATALOG_TIMEOUT_MS; let probe: UninitializedACPProcess | null = null; - try { - const catalogProbe = (async () => { - const initializedProbe = await this.spawnProcess(PROBE_ENV, { - initializeTimeoutMs: timeoutMs, - onSpawned: (spawned) => { - probe = spawned; - }, - }); - probe = initializedProbe; - const response = await this.runACPRequest(() => - initializedProbe.connection.newSession({ - cwd, - mcpServers: [], - }), - ); - const transformed = this.transformSessionResponse(response); - const derivedModels = deriveModelDefinitionsFromACP( - this.provider, - transformed.models, - transformed.configOptions, - ); - const models = this.catalogModelResolver - ? await this.catalogModelResolver({ - connection: initializedProbe.connection, - sessionId: response.sessionId, - models: derivedModels, - configOptions: transformed.configOptions, - runRequest: (request) => this.runACPRequest(request), - transformConfigOptions: (configOptions) => - this.configOptionsTransformer - ? this.configOptionsTransformer(configOptions) - : configOptions, - logger: this.logger, - provider: this.provider, - }) - : derivedModels; - const modeInfo = deriveModesFromACP( - this.defaultModes, - transformed.modes, - transformed.configOptions, - ); - return { - models: this.modelTransformer ? this.modelTransformer(models) : models, - modes: modeInfo.modes, - }; - })(); + let closePromise: Promise | null = null; + const closeProbe = (): Promise => { + if (!probe) return Promise.resolve(); + closePromise ??= this.closeProbe(probe); + return closePromise; + }; + const handleAbort = () => void closeProbe().catch(() => undefined); + context?.signal.addEventListener("abort", handleAbort, { once: true }); - return await withTimeout( - catalogProbe, - timeoutMs, - `ACP catalog probe timed out after ${timeoutMs}ms`, + try { + const initializedProbe = await runProviderRefreshActivity(context, "initialize", () => + raceProviderRefreshAbort( + context?.signal, + this.spawnProcess(PROBE_ENV, { + onSpawned: (spawned) => { + probe = spawned; + if (context?.signal.aborted) void closeProbe().catch(() => undefined); + }, + }), + ), ); + probe = initializedProbe; + const response = await runProviderRefreshActivity(context, "session/new", () => + raceProviderRefreshAbort( + context?.signal, + this.runACPRequest(() => + initializedProbe.connection.newSession({ + cwd, + mcpServers: [], + }), + ), + ), + ); + const transformed = this.transformSessionResponse(response); + const derivedModels = deriveModelDefinitionsFromACP( + this.provider, + transformed.models, + transformed.configOptions, + ); + const models = this.catalogModelResolver + ? await runProviderRefreshActivity(context, "catalog.resolve", () => + raceProviderRefreshAbort( + context?.signal, + this.catalogModelResolver?.({ + connection: initializedProbe.connection, + sessionId: response.sessionId, + models: derivedModels, + configOptions: transformed.configOptions, + runRequest: (request) => this.runACPRequest(request), + transformConfigOptions: (configOptions) => + this.configOptionsTransformer + ? this.configOptionsTransformer(configOptions) + : configOptions, + logger: this.logger, + provider: this.provider, + }) ?? Promise.resolve(derivedModels), + ), + ) + : derivedModels; + const modeInfo = deriveModesFromACP( + this.defaultModes, + transformed.modes, + transformed.configOptions, + ); + return { + models: this.modelTransformer ? this.modelTransformer(models) : models, + modes: modeInfo.modes, + }; } finally { - if (probe) { - await this.closeProbe(probe); - } + context?.signal.removeEventListener("abort", handleAbort); + await closeProbe(); } } diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index c02697c8d..22873f407 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -121,9 +121,11 @@ import { type ListImportableSessionsOptions, type McpServerConfig, type ProviderCatalog, + type ProviderRefreshContext, type ResolveAgentDefaultModeInput, } from "../../agent-sdk-types.js"; import { importSessionFromPersistence } from "../../provider-session-import.js"; +import { runProviderRefreshActivity } from "../../provider-refresh-deadline.js"; import { checkProviderLaunchAvailable, createProviderEnv, @@ -392,7 +394,7 @@ interface ClaudeAgentClientOptions { runtimeSettings?: ProviderRuntimeSettings; queryFactory?: ClaudeQueryFactory; resolveBinary?: () => Promise; - resolveVersion?: () => Promise; + resolveVersion?: (signal?: AbortSignal) => Promise; configDir?: string; } @@ -1476,7 +1478,7 @@ export class ClaudeAgentClient implements AgentClient { private readonly runtimeSettings?: ProviderRuntimeSettings; private readonly queryFactory?: ClaudeQueryFactory; private readonly resolveBinary: () => Promise; - private readonly resolveVersion: () => Promise; + private readonly resolveVersion: (signal?: AbortSignal) => Promise; private readonly configDir?: string; constructor(options: ClaudeAgentClientOptions) { @@ -1486,7 +1488,8 @@ export class ClaudeAgentClient implements AgentClient { this.queryFactory = options.queryFactory; this.resolveBinary = options.resolveBinary ?? (() => resolveClaudeBinary(this.runtimeSettings)); this.resolveVersion = - options.resolveVersion ?? (() => resolveClaudeCodeVersion(this.runtimeSettings)); + options.resolveVersion ?? + ((signal) => resolveClaudeCodeVersion(this.runtimeSettings, signal)); this.configDir = options.configDir; } @@ -1540,18 +1543,21 @@ export class ClaudeAgentClient implements AgentClient { }); } - async fetchCatalog(_options: FetchCatalogOptions): Promise { + async fetchCatalog( + _options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise { // Claude exposes a global catalog here; cwd/force are intentionally irrelevant. let claudeCodeVersion: string | undefined; try { - claudeCodeVersion = await this.resolveVersion(); + claudeCodeVersion = await runProviderRefreshActivity(context, "version", () => + this.resolveVersion(context?.signal), + ); } catch (error) { this.logger.warn({ err: error }, "Failed to resolve Claude Code version for model catalog"); } - const models = await getClaudeModelsWithSettings( - this.logger, - this.configDir, - claudeCodeVersion, + const models = await runProviderRefreshActivity(context, "settings", () => + getClaudeModelsWithSettings(this.logger, this.configDir, claudeCodeVersion), ); const modes = detectIneligibleAutoModeTransport( createProviderEnv({ baseEnv: process.env, runtimeSettings: this.runtimeSettings }), @@ -1680,6 +1686,7 @@ async function resolveClaudeBinary(runtimeSettings?: ProviderRuntimeSettings): P export async function resolveClaudeCodeVersion( runtimeSettings?: ProviderRuntimeSettings, + signal?: AbortSignal, ): Promise { const launch = await resolveProviderLaunch({ commandConfig: runtimeSettings?.command, @@ -1693,6 +1700,7 @@ export async function resolveClaudeCodeVersion( const { stdout, stderr } = await execCommand(executable, [...launch.args, "--version"], { ...createProviderEnvSpec({ runtimeSettings }), timeout: 5_000, + signal, }); const version = parseClaudeCodeVersion(`${stdout}\n${stderr}`); if (!version) { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 9680d62a0..1a5143430 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -33,8 +33,11 @@ import { type ImportProviderSessionInput, type ListImportableSessionsOptions, type ProviderCatalog, + type ProviderRefreshContext, + type ResolveAgentDefaultModeInput, } from "../agent-sdk-types.js"; import { importSessionFromPersistence } from "../provider-session-import.js"; +import { runProviderRefreshActivity } from "../provider-refresh-deadline.js"; import type { Logger } from "pino"; import type { ChildProcess, ChildProcessWithoutNullStreams } from "node:child_process"; @@ -6611,31 +6614,33 @@ export class CodexAppServerAgentClient implements AgentClient { return this.goalsEnabledPromise; } - private resolveAutoReviewEnabled(): Promise { + private resolveAutoReviewEnabled(signal?: AbortSignal): Promise { + if (signal) return this.probeAutoReviewEnabled(signal); if (!this.autoReviewEnabledPromise) { - this.autoReviewEnabledPromise = (async () => { - try { - const launchPrefix = await resolveCodexLaunchPrefix(this.runtimeSettings); - const versionOutput = await resolveBinaryVersion(launchPrefix.command); - const enabled = codexVersionAtLeast(versionOutput, CODEX_AUTO_REVIEW_MIN_VERSION); - this.logger.trace( - { - provider: CODEX_PROVIDER, - versionOutput, - enabled, - }, - "provider.codex.config.auto_review_resolved", - ); - return enabled; - } catch (error) { - this.logger.warn({ err: error }, "Failed to probe codex version for auto-review gate"); - return false; - } - })(); + this.autoReviewEnabledPromise = this.probeAutoReviewEnabled(); } return this.autoReviewEnabledPromise; } + private async probeAutoReviewEnabled(signal?: AbortSignal): Promise { + try { + const launchPrefix = await resolveCodexLaunchPrefix(this.runtimeSettings); + signal?.throwIfAborted(); + const versionOutput = await resolveBinaryVersion(launchPrefix.command, signal); + signal?.throwIfAborted(); + const enabled = codexVersionAtLeast(versionOutput, CODEX_AUTO_REVIEW_MIN_VERSION); + this.logger.trace( + { provider: CODEX_PROVIDER, versionOutput, enabled }, + "provider.codex.config.auto_review_resolved", + ); + return enabled; + } catch (error) { + if (signal?.aborted) throw signal.reason; + this.logger.warn({ err: error }, "Failed to probe codex version for auto-review gate"); + return false; + } + } + private async spawnAppServer( launchEnv?: Record, options?: { goalsEnabled?: boolean; agentId?: string }, @@ -6787,10 +6792,15 @@ export class CodexAppServerAgentClient implements AgentClient { }); } - async fetchCatalog(_options: FetchCatalogOptions): Promise { + async fetchCatalog( + _options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise { const [models, autoReviewEnabled] = await Promise.all([ - this.fetchModelsFromAppServer(), - this.resolveAutoReviewEnabled(), + this.fetchModelsFromAppServer(context), + runProviderRefreshActivity(context, "version", () => + this.resolveAutoReviewEnabled(context?.signal), + ), ]); return { models, @@ -6801,23 +6811,46 @@ export class CodexAppServerAgentClient implements AgentClient { }; } - async resolveDefaultModeId(): Promise { - return (await this.resolveAutoReviewEnabled()) ? "auto-review" : DEFAULT_CODEX_MODE_ID; + async resolveDefaultModeId(input: ResolveAgentDefaultModeInput): Promise { + return (await this.resolveAutoReviewEnabled(input.signal)) + ? "auto-review" + : DEFAULT_CODEX_MODE_ID; } - private async fetchModelsFromAppServer(): Promise { + private async fetchModelsFromAppServer( + context?: ProviderRefreshContext, + ): Promise { // Codex model/list is global to the app server in this flow; cwd/force are intentionally ignored. - const child = await this.spawnAppServer(); - const client = new CodexAppServerClient(child, this.logger); + let client: CodexAppServerClient | undefined; + let disposePromise: Promise | undefined; + const dispose = () => { + if (!client) return Promise.resolve(); + disposePromise ??= client.dispose(); + return disposePromise; + }; + const handleAbort = () => void dispose().catch(() => undefined); + context?.signal.addEventListener("abort", handleAbort, { once: true }); try { - await client.request("initialize", buildCodexAppServerInitializeParams()); + await runProviderRefreshActivity(context, "app-server.start", async () => { + const child = await this.spawnAppServer(); + client = new CodexAppServerClient(child, this.logger); + if (context?.signal.aborted) await dispose(); + }); + if (!client) throw new Error("Codex app-server did not start"); + await runProviderRefreshActivity(context, "initialize", () => + client!.request("initialize", buildCodexAppServerInitializeParams()), + ); client.notify("initialized", {}); - const rawResponse = await client.request("model/list", {}); + const rawResponse = await runProviderRefreshActivity(context, "model/list", () => + client!.request("model/list", {}), + ); const parsedResponse = CodexModelListResponseSchema.safeParse(rawResponse); const models = parsedResponse.success ? (parsedResponse.data.data ?? []) : []; - const configuredDefaults = await readCodexConfiguredDefaults(client, this.logger); + const configuredDefaults = await runProviderRefreshActivity(context, "config/read", () => + readCodexConfiguredDefaults(client!, this.logger), + ); const configuredDefaultModelId = configuredDefaults.model; const configuredDefaultThinkingOptionId = configuredDefaults.thinkingOptionId; const hasConfiguredDefaultModel = @@ -6832,7 +6865,8 @@ export class CodexAppServerAgentClient implements AgentClient { }), ); } finally { - await client.dispose(); + context?.signal.removeEventListener("abort", handleAbort); + await dispose(); } } diff --git a/packages/server/src/server/agent/providers/diagnostic-utils.ts b/packages/server/src/server/agent/providers/diagnostic-utils.ts index 0cf2b44b0..670bd8075 100644 --- a/packages/server/src/server/agent/providers/diagnostic-utils.ts +++ b/packages/server/src/server/agent/providers/diagnostic-utils.ts @@ -135,11 +135,15 @@ export function toDiagnosticErrorMessage(error: unknown): string { return formatNonErrorDiagnostic(error); } -export async function resolveBinaryVersion(binaryPath: string): Promise { +export async function resolveBinaryVersion( + binaryPath: string, + signal?: AbortSignal, +): Promise { try { const { stdout } = await execCommand(binaryPath, ["--version"], { ...createProviderEnvSpec(), timeout: 5_000, + signal, }); return stdout.trim() || "unknown"; } catch (error) { diff --git a/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts b/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts index 785838bb6..5bac343fd 100644 --- a/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts +++ b/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts @@ -5,6 +5,7 @@ import path from "node:path"; import { describe, expect, test } from "vitest"; import { createTestLogger } from "../../../test-utils/test-logger.js"; +import { runProviderRefreshWithDeadline } from "../provider-refresh-deadline.js"; import { buildVersionProbeCommand, GenericACPAgentClient } from "./generic-acp-agent.js"; const TEST_ACP_TIMEOUT_MS = 1_000; @@ -91,8 +92,15 @@ describe("GenericACPAgentClient diagnostics", () => { }); await expect( - client.fetchCatalog({ cwd: tmpdir(), force: true, timeoutMs: TEST_ACP_TIMEOUT_MS }), - ).rejects.toThrow(`ACP catalog probe timed out after ${TEST_ACP_TIMEOUT_MS}ms`); + runProviderRefreshWithDeadline({ + label: "Grok", + timeoutMs: TEST_ACP_TIMEOUT_MS, + operation: (context) => + client.fetchCatalog({ scope: "workspace", cwd: tmpdir(), force: true }, context), + }), + ).rejects.toThrow( + `Timed out refreshing Grok after ${TEST_ACP_TIMEOUT_MS}ms; pending: session/new`, + ); const pid = Number(await readFile(pidPath, "utf8")); await expectProcessExit(pid); @@ -116,7 +124,7 @@ describe("GenericACPAgentClient diagnostics", () => { }, }); - await client.fetchCatalog({ cwd: testDir, force: true, timeoutMs: TEST_ACP_TIMEOUT_MS }); + await client.fetchCatalog({ scope: "workspace", cwd: testDir, force: true }); const session = await client.createSession({ provider: "acp", cwd: testDir }); await session.close(); diff --git a/packages/server/src/server/agent/providers/mock-slow-provider.ts b/packages/server/src/server/agent/providers/mock-slow-provider.ts index ef2ad42df..19817d29c 100644 --- a/packages/server/src/server/agent/providers/mock-slow-provider.ts +++ b/packages/server/src/server/agent/providers/mock-slow-provider.ts @@ -7,6 +7,7 @@ import type { AgentSession, AgentSessionConfig, FetchCatalogOptions, + ProviderRefreshContext, ProviderCatalog, } from "../agent-sdk-types.js"; @@ -24,10 +25,6 @@ const CAPABILITIES: AgentCapabilityFlags = { supportsRewindBoth: false, }; -function neverResolves(): Promise { - return new Promise(() => {}); -} - export class MockSlowProviderClient implements AgentClient { readonly provider: AgentProvider = MOCK_SLOW_PROVIDER_ID; readonly capabilities = CAPABILITIES; @@ -36,8 +33,19 @@ export class MockSlowProviderClient implements AgentClient { return process.env.PASEO_ENABLE_MOCK_SLOW === "true"; } - async fetchCatalog(_options: FetchCatalogOptions): Promise { - return neverResolves(); + async fetchCatalog( + _options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise { + return await (context?.runActivity( + "mock.catalog", + () => + new Promise((_resolve, reject) => { + context.signal.addEventListener("abort", () => reject(context.signal.reason), { + once: true, + }); + }), + ) ?? new Promise(() => {})); } async getDiagnostic(): Promise<{ diagnostic: string }> { diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index f3f6c7f97..db996cf70 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -33,10 +33,12 @@ import { type ImportProviderSessionInput, type ListImportableSessionsOptions, type ProviderCatalog, + type ProviderRefreshContext, type ToolCallDetail, } from "../../agent-sdk-types.js"; import type { PaseoToolCatalog } from "../../tools/types.js"; import { importSessionFromPersistence } from "../../provider-session-import.js"; +import { runProviderRefreshActivity } from "../../provider-refresh-deadline.js"; import { runProviderTurn } from "../provider-runner.js"; import { checkProviderLaunchAvailable, @@ -108,7 +110,6 @@ import { import { DEFAULT_OMP_THINKING_LEVEL, mapOmpModel } from "./map-omp-model.js"; const OMP_PROVIDER = "omp"; -const OMP_CATALOG_REQUEST_TIMEOUT_MS = 120_000; const QUESTION_RESPONSE_HEADER = "Response"; const QUESTION_COMMENT_HEADER = "Comment"; const OMP_ASK_USER_FREEFORM_SENTINEL = "✏️ Type custom response..."; @@ -2304,23 +2305,44 @@ export class OmpAgentClient implements AgentClient { } } - async fetchCatalog(options: FetchCatalogOptions): Promise { + async fetchCatalog( + options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise { const launchMode = this.resolveLaunchMode(undefined); - const runtimeSession = await this.runtime.startSession({ - cwd: options.scope === "global" ? homedir() : options.cwd, - protocolMode: "rpc-ui", - modeId: launchMode.modeId, - extraArgs: launchMode.extraArgs, - }); + let runtimeSession: OmpRuntimeSession | undefined; + let closePromise: Promise | undefined; + const closeSession = () => { + if (!runtimeSession) return Promise.resolve(); + closePromise ??= runtimeSession.close(); + return closePromise; + }; + const handleAbort = () => void closeSession().catch(() => undefined); + context?.signal.addEventListener("abort", handleAbort, { once: true }); try { + await runProviderRefreshActivity(context, "runtime.start", async () => { + runtimeSession = await this.runtime.startSession({ + cwd: options.scope === "global" ? homedir() : options.cwd, + protocolMode: "rpc-ui", + modeId: launchMode.modeId, + extraArgs: launchMode.extraArgs, + signal: context?.signal, + }); + if (context?.signal.aborted) await closeSession(); + }); + if (!runtimeSession) throw new Error("OMP catalog runtime did not start"); + const catalogSession = runtimeSession; const models = transformOmpModels( - (await runtimeSession.getAvailableModels(OMP_CATALOG_REQUEST_TIMEOUT_MS)).map((model) => - mapOmpModel(model, this.provider), - ), + ( + await runProviderRefreshActivity(context, "get_available_models", () => + catalogSession.getAvailableModels(null), + ) + ).map((model) => mapOmpModel(model, this.provider)), ); return { models, modes: [...OMP_MODES] }; } finally { - await runtimeSession.close(); + context?.signal.removeEventListener("abort", handleAbort); + await closeSession(); } } diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.ts index 22acf58ce..97e01c937 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.ts @@ -88,13 +88,18 @@ export class OmpCliRuntime implements OmpRuntime { ...(spawn ? { spawn: () => spawn(launch) } : {}), }; const process = new JsonlRpcProcess(processOptions); + const handleAbort = () => void process.close(input.signal?.reason).catch(() => undefined); + input.signal?.addEventListener("abort", handleAbort, { once: true }); try { await negotiateOmpProtocolV2(process, this.options.logger); + input.signal?.throwIfAborted(); return new OmpCliRuntimeSession(process, this.commandsRpcName); } catch (error) { const startupError = error instanceof Error ? error : new Error(String(error)); await process.close(startupError); throw startupError; + } finally { + input.signal?.removeEventListener("abort", handleAbort); } } } @@ -216,7 +221,7 @@ class OmpCliRuntimeSession implements OmpRuntimeSession { return data.messages ?? []; } - async getAvailableModels(timeoutMs?: number): Promise { + async getAvailableModels(timeoutMs?: number | null): Promise { const data = OmpModelsResultSchema.parse( await this.request({ type: "get_available_models" }, timeoutMs), ); @@ -337,7 +342,7 @@ class OmpCliRuntimeSession implements OmpRuntimeSession { await this.process.close(new Error("OMP RPC session is closed")); } - private request(command: OmpRpcCommand, timeoutMs?: number): Promise { + private request(command: OmpRpcCommand, timeoutMs?: number | null): Promise { return this.process.request(OmpRpcCommandSchema.parse(command), timeoutMs); } diff --git a/packages/server/src/server/agent/providers/omp/runtime.ts b/packages/server/src/server/agent/providers/omp/runtime.ts index 3a8d90c61..dc8001e3e 100644 --- a/packages/server/src/server/agent/providers/omp/runtime.ts +++ b/packages/server/src/server/agent/providers/omp/runtime.ts @@ -30,6 +30,7 @@ export interface OmpRuntimeLaunch { export interface OmpStartSessionInput { cwd: string; + signal?: AbortSignal; env?: Record; protocolMode?: "rpc" | "rpc-ui"; model?: string; @@ -52,7 +53,7 @@ export interface OmpRuntimeSession { abort(): Promise; getState(): Promise; getMessages(): Promise; - getAvailableModels(timeoutMs?: number): Promise; + getAvailableModels(timeoutMs?: number | null): Promise; setModel(provider: string, modelId: string): Promise; setThinkingLevel(level: OmpThinkingLevel): Promise; getSessionStats(): Promise; diff --git a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts index 6d46c50f1..0b17fe8c5 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts @@ -274,7 +274,7 @@ export class FakeOmpSession implements OmpRuntimeSession { return this.messages; } - async getAvailableModels(_timeoutMs?: number): Promise { + async getAvailableModels(_timeoutMs?: number | null): Promise { return this.models; } diff --git a/packages/server/src/server/agent/providers/opencode-agent.list-models-timeout.test.ts b/packages/server/src/server/agent/providers/opencode-agent.list-models-timeout.test.ts index ffa85ce38..a3aed018b 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.list-models-timeout.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.list-models-timeout.test.ts @@ -2,6 +2,7 @@ import { afterEach, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { OpenCodeAgentClient } from "./opencode-agent.js"; +import { runProviderRefreshWithDeadline } from "../provider-refresh-deadline.js"; import { TestOpenCodeClient, TestOpenCodeHarness, @@ -11,32 +12,15 @@ afterEach(() => { vi.useRealTimers(); }); -test("allows a slow provider.list call to succeed instead of failing after 10 seconds", async () => { +test("the catalog deadline aborts provider.list and releases the server", async () => { vi.useFakeTimers(); const runtime = new TestOpenCodeHarness(); const openCodeClient = new TestOpenCodeClient(); - openCodeClient.providerListImplementation = () => - new Promise((resolve) => { - setTimeout(() => { - resolve({ - data: { - connected: ["zai"], - all: [ - { - id: "zai", - name: "Z.AI", - models: { - "glm-5.1": { - name: "GLM 5.1", - limit: { context: 128_000 }, - }, - }, - }, - ], - }, - }); - }, 15_000); + openCodeClient.providerListImplementation = (_parameters, options) => + new Promise((_resolve, reject) => { + const signal = (options as { signal: AbortSignal }).signal; + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); }); runtime.enqueueClient(openCodeClient); @@ -44,24 +28,72 @@ test("allows a slow provider.list call to succeed instead of failing after 10 se serverManager: runtime, createClient: runtime.createClient, }); - const modelsPromise = client.fetchCatalog({ - scope: "workspace", - cwd: "/tmp/opencode-models", - force: false, + const modelsPromise = runProviderRefreshWithDeadline({ + label: "OpenCode", + timeoutMs: 100, + operation: (context) => + client.fetchCatalog( + { scope: "workspace", cwd: "/tmp/opencode-models", force: false }, + context, + ), }); + const rejection = expect(modelsPromise).rejects.toThrow( + "Timed out refreshing OpenCode after 100ms; pending: provider.list", + ); - await vi.advanceTimersByTimeAsync(15_000); + await vi.advanceTimersByTimeAsync(100); - await expect(modelsPromise).resolves.toMatchObject({ - models: [ - { - provider: "opencode", - id: "zai/glm-5.1", - label: "GLM 5.1", - }, - ], - }); + await rejection; expect(openCodeClient.calls.providerList).toHaveLength(1); + expect(openCodeClient.calls.providerListOptions[0]).toMatchObject({ + signal: expect.objectContaining({ aborted: true }), + }); + expect(runtime.acquisitions).toEqual([{ kind: "current", releaseCount: 1 }]); +}); + +test("the catalog deadline aborts app.agents and releases the server", async () => { + vi.useFakeTimers(); + + const runtime = new TestOpenCodeHarness(); + const openCodeClient = new TestOpenCodeClient(); + openCodeClient.providerListResponse = { + data: { + connected: ["openai"], + all: [{ id: "openai", name: "OpenAI", models: {} }], + }, + }; + openCodeClient.appAgentsImplementation = (_parameters, options) => + new Promise((_resolve, reject) => { + const signal = (options as { signal: AbortSignal }).signal; + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + runtime.enqueueClient(openCodeClient); + + const client = new OpenCodeAgentClient(createTestLogger(), undefined, { + serverManager: runtime, + createClient: runtime.createClient, + }); + const catalogPromise = runProviderRefreshWithDeadline({ + label: "OpenCode", + timeoutMs: 100, + operation: (context) => + client.fetchCatalog( + { scope: "workspace", cwd: "/tmp/opencode-agents", force: false }, + context, + ), + }); + const rejection = expect(catalogPromise).rejects.toThrow( + "Timed out refreshing OpenCode after 100ms; pending: app.agents", + ); + + await vi.advanceTimersByTimeAsync(100); + + await rejection; + expect(openCodeClient.calls.appAgents).toHaveLength(1); + expect(openCodeClient.calls.appAgentsOptions[0]).toMatchObject({ + signal: expect.objectContaining({ aborted: true }), + }); + expect(runtime.acquisitions).toEqual([{ kind: "current", releaseCount: 1 }]); }); test("uses a new server for explicit catalog refresh", async () => { diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 7dbe5d3b7..d2eb6730f 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -41,6 +41,7 @@ import { type AgentTimelineItem, type AgentUsage, type FetchCatalogOptions, + type ProviderRefreshContext, type ImportableProviderSession, type ImportProviderSessionContext, type ImportProviderSessionInput, @@ -53,6 +54,10 @@ import { type ToolCallTimelineItem, } from "../agent-sdk-types.js"; import { importSessionFromPersistence } from "../provider-session-import.js"; +import { + raceProviderRefreshAbort, + runProviderRefreshActivity, +} from "../provider-refresh-deadline.js"; import { isDefaultAgentCreateConfigUnattended, resolveDefaultAgentCreateConfig, @@ -68,6 +73,7 @@ import { execCommand } from "../../../utils/spawn.js"; import { mapOpencodeToolCall } from "./opencode/tool-call-mapper.js"; import { OpenCodeServerManager, + type OpenCodeServerAcquisition, type OpenCodeServerManagerLike, } from "./opencode/server-manager.js"; import { resolveOpenCodeHomeDir } from "./opencode/paths.js"; @@ -297,9 +303,9 @@ type OpenCodeMcpConfig = }; const MCP_ALREADY_PRESENT_ERROR_TOKENS = ["already", "exists", "connected"] as const; -const OPENCODE_PROVIDER_LIST_TIMEOUT_MS = 30_000; const OPENCODE_METADATA_CONCURRENCY = 4; const openCodeMetadataLimit = pLimit(OPENCODE_METADATA_CONCURRENCY); + const OPENCODE_HANDLED_BUILTIN_SLASH_COMMANDS: AgentSlashCommand[] = [ { name: "compact", @@ -1422,14 +1428,22 @@ export class OpenCodeAgentClient implements AgentClient { } } - async fetchCatalog(options: FetchCatalogOptions): Promise { - const acquisition = options.force - ? await this.serverManager.acquireNew() - : await this.serverManager.acquireCurrent(); - const { url } = acquisition.server; - const isGlobalCatalog = options.scope === "global"; - + async fetchCatalog( + options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise { + let acquisition: OpenCodeServerAcquisition | undefined; try { + await runProviderRefreshActivity(context, "server.acquire", async () => { + acquisition = options.force + ? await this.serverManager.acquireNew(context?.signal) + : await this.serverManager.acquireCurrent(context?.signal); + }); + if (!acquisition) throw new Error("OpenCode server acquisition did not complete"); + context?.signal.throwIfAborted(); + const { url } = acquisition.server; + const isGlobalCatalog = options.scope === "global"; + // OpenCode treats the catalog directory as a workspace. The global catalog // is not a project, so use the neutral OpenCode home instead of user home. const directory = isGlobalCatalog ? this.resolveHomeDir() : options.cwd; @@ -1444,12 +1458,12 @@ export class OpenCodeAgentClient implements AgentClient { const client = this.createOpenCodeClient({ baseUrl: url, directory }); const [models, modes] = await Promise.all([ - this.fetchModelsFromClient(client, directory), - this.fetchModesFromClient(client, directory), + this.fetchModelsFromClient(client, directory, context), + this.fetchModesFromClient(client, directory, context), ]); return { models, modes }; } finally { - await acquisition.release(); + await acquisition?.release(); } } @@ -1637,12 +1651,18 @@ export class OpenCodeAgentClient implements AgentClient { private async fetchModelsFromClient( client: OpencodeClient, directory: string, + context?: ProviderRefreshContext, ): Promise { - const response = await openCodeMetadataLimit(() => - withTimeout( - client.provider.list({ directory }), - OPENCODE_PROVIDER_LIST_TIMEOUT_MS, - `OpenCode provider.list timed out after ${OPENCODE_PROVIDER_LIST_TIMEOUT_MS / 1000}s - server may not be authenticated or connected to any providers`, + const response = await runProviderRefreshActivity(context, "provider.list", () => + raceProviderRefreshAbort( + context?.signal, + openCodeMetadataLimit(() => { + context?.signal.throwIfAborted(); + return client.provider.list( + { directory }, + context ? { signal: context.signal } : undefined, + ); + }), ), ); @@ -1669,7 +1689,7 @@ export class OpenCodeAgentClient implements AgentClient { } const models: AgentModelDefinition[] = []; - this.modelContextWindows.clear(); + const contextWindows = new Map(); for (const provider of providers.all) { if (!isAccessible(provider)) { continue; @@ -1679,7 +1699,7 @@ export class OpenCodeAgentClient implements AgentClient { const definition = buildOpenCodeModelDefinition(provider, modelId, model); const contextWindowMaxTokens = extractOpenCodeModelContextWindow(model); if (contextWindowMaxTokens !== undefined) { - this.modelContextWindows.set( + contextWindows.set( buildOpenCodeModelLookupKey(provider.id, modelId), contextWindowMaxTokens, ); @@ -1688,18 +1708,25 @@ export class OpenCodeAgentClient implements AgentClient { } } + context?.signal.throwIfAborted(); + this.modelContextWindows.clear(); + for (const [key, value] of contextWindows) this.modelContextWindows.set(key, value); + return models; } private async fetchModesFromClient( client: OpencodeClient, directory: string, + context?: ProviderRefreshContext, ): Promise { - const response = await openCodeMetadataLimit(() => - withTimeout( - client.app.agents({ directory }), - 10_000, - "OpenCode app.agents timed out after 10s", + const response = await runProviderRefreshActivity(context, "app.agents", () => + raceProviderRefreshAbort( + context?.signal, + openCodeMetadataLimit(() => { + context?.signal.throwIfAborted(); + return client.app.agents({ directory }, context ? { signal: context.signal } : undefined); + }), ), ); diff --git a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts index 38f698b47..d6a1d5578 100644 --- a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts +++ b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts @@ -138,6 +138,25 @@ describe("OpenCodeServerManager generations", () => { expect(await runtime.managedProcesses.list()).toEqual([]); }); + test("aborted acquisition transfers no reference and leaves startup reusable", async () => { + const { manager, runtime } = createTestManager([4477], { autoAnnounce: false }); + const controller = new AbortController(); + + const abortedAcquisition = manager.acquireCurrent(controller.signal); + await runtime.settle(); + controller.abort(new Error("catalog refresh expired")); + + await expect(abortedAcquisition).rejects.toThrow("catalog refresh expired"); + runtime.processForPort(4477).announceListening(); + + const nextAcquisition = await manager.acquireCurrent(); + expect(nextAcquisition.server.url).toBe("http://127.0.0.1:4477"); + expect(runtime.launchedPorts).toEqual([4477]); + + await nextAcquisition.release(); + expect(runtime.terminatedPorts).toEqual([4477]); + }); + test("shutdown kills a server that is still starting", async () => { const { manager, runtime } = createTestManager([4472], { autoAnnounce: false }); diff --git a/packages/server/src/server/agent/providers/opencode/server-manager.ts b/packages/server/src/server/agent/providers/opencode/server-manager.ts index 2baa07bbd..0c850514d 100644 --- a/packages/server/src/server/agent/providers/opencode/server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/server-manager.ts @@ -25,8 +25,8 @@ export interface OpenCodeServerAcquisition { } export interface OpenCodeServerManagerLike { - acquireCurrent(): Promise; - acquireNew(): Promise; + acquireCurrent(signal?: AbortSignal): Promise; + acquireNew(signal?: AbortSignal): Promise; acquireDedicated(env: Record): Promise; acquireExisting(url: string): OpenCodeServerAcquisition | null; shutdown(): Promise; @@ -137,13 +137,17 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { process.on("SIGINT", cleanup); } - async acquireCurrent(): Promise { - const server = await this.getCurrentServer(); + async acquireCurrent(signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + const server = await waitForServerAcquisition(this.getCurrentServer(), signal); + signal?.throwIfAborted(); return this.acquireServer(server); } - async acquireNew(): Promise { - const server = await this.getNewServer(); + async acquireNew(signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + const server = await waitForServerAcquisition(this.getNewServer(), signal); + signal?.throwIfAborted(); return this.acquireServer(server); } @@ -533,6 +537,23 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { } } +async function waitForServerAcquisition( + operation: Promise, + signal: AbortSignal | undefined, +): Promise { + if (!signal) return await operation; + let handleAbort: (() => void) | undefined; + const aborted = new Promise((_resolve, reject) => { + handleAbort = () => reject(signal.reason); + signal.addEventListener("abort", handleAbort, { once: true }); + }); + try { + return await Promise.race([operation, aborted]); + } finally { + if (handleAbort) signal.removeEventListener("abort", handleAbort); + } +} + async function resolveOpenCodeBinary(): Promise { const found = await findExecutable("opencode"); if (!found) { diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts index 7433532f0..f17ba9da6 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts @@ -71,6 +71,7 @@ export class TestOpenCodeHarness implements OpenCodeServerManagerLike { export class TestOpenCodeClient { readonly calls = { appAgents: [] as unknown[], + appAgentsOptions: [] as unknown[], commandList: [] as unknown[], eventSubscribe: [] as unknown[], experimentalSessionList: [] as unknown[], @@ -79,6 +80,7 @@ export class TestOpenCodeClient { mcpConnect: [] as unknown[], permissionReply: [] as unknown[], providerList: [] as unknown[], + providerListOptions: [] as unknown[], questionReject: [] as unknown[], questionReply: [] as unknown[], sessionAbort: [] as unknown[], @@ -95,6 +97,9 @@ export class TestOpenCodeClient { }; appAgentsResponse: OpenCodeResponse = { data: [] }; + appAgentsImplementation: + | ((parameters: unknown, options: unknown) => Promise) + | null = null; commandListResponse: OpenCodeResponse = { data: [] }; eventStream: AsyncIterable; experimentalSessionListResponse: OpenCodeResponse = { data: [] }; @@ -102,7 +107,9 @@ export class TestOpenCodeClient { mcpConnectResponse: OpenCodeResponse = {}; permissionReplyResponse: OpenCodeResponse = {}; providerListResponse: OpenCodeResponse = { data: { connected: [], all: [] } }; - providerListImplementation: (() => Promise) | null = null; + providerListImplementation: + | ((parameters: unknown, options: unknown) => Promise) + | null = null; globalEventImplementation: | ((options: unknown) => Promise<{ stream: AsyncIterable }>) | null = null; @@ -141,8 +148,12 @@ export class TestOpenCodeClient { asSdkClient(): OpencodeClient { return { app: { - agents: async (parameters: unknown) => { + agents: async (parameters: unknown, options: unknown) => { this.calls.appAgents.push(parameters); + this.calls.appAgentsOptions.push(options); + if (this.appAgentsImplementation) { + return await this.appAgentsImplementation(parameters, options); + } return this.appAgentsResponse; }, }, @@ -195,10 +206,11 @@ export class TestOpenCodeClient { }, }, provider: { - list: async (parameters: unknown) => { + list: async (parameters: unknown, options: unknown) => { this.calls.providerList.push(parameters); + this.calls.providerListOptions.push(options); return this.providerListImplementation - ? await this.providerListImplementation() + ? await this.providerListImplementation(parameters, options) : this.providerListResponse; }, }, diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 04adc29b9..82bd0a823 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -36,9 +36,11 @@ import { type ImportProviderSessionInput, type ListImportableSessionsOptions, type ProviderCatalog, + type ProviderRefreshContext, type ToolCallDetail, } from "../../agent-sdk-types.js"; import { importSessionFromPersistence } from "../../provider-session-import.js"; +import { runProviderRefreshActivity } from "../../provider-refresh-deadline.js"; import { runProviderTurn } from "../provider-runner.js"; import { checkProviderLaunchAvailable, @@ -88,7 +90,6 @@ import { const PI_PROVIDER = "pi"; const DEFAULT_PI_THINKING_LEVEL: PiThinkingLevel = "medium"; const PI_BINARY_COMMAND = process.env.PI_COMMAND ?? process.env.PI_ACP_PI_COMMAND ?? "pi"; -const PI_CATALOG_REQUEST_TIMEOUT_MS = 120_000; const PASEO_PI_TREE_EXTENSION_COMMAND = "paseo_tree"; const PASEO_PI_CAPTURE_EXTENSION_COMMAND = "paseo_capture_entries"; const PASEO_PI_ENTRY_CAPTURE_MARKER = "PASEO_ENTRY_CAPTURE"; @@ -2474,19 +2475,40 @@ export class PiRpcAgentClient implements AgentClient { } } - async fetchCatalog(options: FetchCatalogOptions): Promise { - const runtimeSession = await this.runtime.startSession({ - cwd: options.scope === "global" ? homedir() : options.cwd, - }); + async fetchCatalog( + options: FetchCatalogOptions, + context?: ProviderRefreshContext, + ): Promise { + let runtimeSession: PiRuntimeSession | undefined; + let closePromise: Promise | undefined; + const closeSession = () => { + if (!runtimeSession) return Promise.resolve(); + closePromise ??= runtimeSession.close(); + return closePromise; + }; + const handleAbort = () => void closeSession().catch(() => undefined); + context?.signal.addEventListener("abort", handleAbort, { once: true }); try { + await runProviderRefreshActivity(context, "runtime.start", async () => { + runtimeSession = await this.runtime.startSession({ + cwd: options.scope === "global" ? homedir() : options.cwd, + signal: context?.signal, + }); + if (context?.signal.aborted) await closeSession(); + }); + if (!runtimeSession) throw new Error("Pi catalog runtime did not start"); + const catalogSession = runtimeSession; const models = transformPiModels( - (await runtimeSession.getAvailableModels(PI_CATALOG_REQUEST_TIMEOUT_MS)).map((model) => - mapPiModel(model, PI_PROVIDER), - ), + ( + await runProviderRefreshActivity(context, "get_available_models", () => + catalogSession.getAvailableModels(null), + ) + ).map((model) => mapPiModel(model, PI_PROVIDER)), ); return { models, modes: [] }; } finally { - await runtimeSession.close(); + context?.signal.removeEventListener("abort", handleAbort); + await closeSession(); } } diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.ts index 66cff0daf..5432b913e 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.ts @@ -78,6 +78,10 @@ export class PiCliRuntime implements PiRuntime { ...(spawn ? { spawn: () => spawn(launch) } : {}), }; const process = new JsonlRpcProcess(processOptions); + if (input.signal?.aborted) { + await process.close(input.signal.reason); + input.signal.throwIfAborted(); + } return new PiCliRuntimeSession(process, this.commandsRpcName); } } diff --git a/packages/server/src/server/agent/providers/pi/runtime.ts b/packages/server/src/server/agent/providers/pi/runtime.ts index 2d4800c14..7d29fc159 100644 --- a/packages/server/src/server/agent/providers/pi/runtime.ts +++ b/packages/server/src/server/agent/providers/pi/runtime.ts @@ -26,6 +26,7 @@ export interface PiRuntimeLaunch { export interface PiStartSessionInput { cwd: string; + signal?: AbortSignal; env?: Record; protocolMode?: "rpc" | "rpc-ui"; model?: string; diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index fad665b10..e27021674 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -427,6 +427,7 @@ export interface PaseoDaemonConfig { dictationFinalTimeoutMs?: number; downloadTokenTtlMs?: number; agentProviderSettings?: AgentProviderRuntimeSettingsMap; + providerCatalogRefreshTimeoutMs?: number; metadataGeneration?: { providers?: Array<{ provider: string; @@ -820,6 +821,7 @@ export async function createPaseoDaemon( const providerSnapshotLogger = logger.child({ module: "provider-snapshot-manager" }); const providerSnapshotManager = new ProviderSnapshotManager({ logger: providerSnapshotLogger, + refreshTimeoutMs: config.providerCatalogRefreshTimeoutMs, runtimeSettings: config.agentProviderSettings, providerOverrides: config.providerOverrides, workspaceGitService, diff --git a/packages/server/src/server/config.test.ts b/packages/server/src/server/config.test.ts index f9d4929c6..5cfe41ec4 100644 --- a/packages/server/src/server/config.test.ts +++ b/packages/server/src/server/config.test.ts @@ -1,4 +1,4 @@ -import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { pathToFileURL } from "node:url"; @@ -26,6 +26,19 @@ describe("server config", () => { expect(standaloneConfig.desktopManaged).toBe(false); }); + test("loads the provider catalog refresh timeout", async () => { + const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-config-provider-timeout-")); + roots.push(paseoHome); + await writeFile( + path.join(paseoHome, "config.json"), + JSON.stringify({ agents: { catalogRefreshTimeoutMs: 180_000 } }), + ); + + const config = loadConfig(paseoHome, { env: {} }); + + expect(config.providerCatalogRefreshTimeoutMs).toBe(180_000); + }); + test("resolves bundled web UI path from source-tree modules", () => { const root = path.parse(process.cwd()).root; expect( diff --git a/packages/server/src/server/config.ts b/packages/server/src/server/config.ts index 2930d1f0b..9419b5b15 100644 --- a/packages/server/src/server/config.ts +++ b/packages/server/src/server/config.ts @@ -555,6 +555,7 @@ export function loadConfig( voiceLlmProviderExplicit: voiceLlm.providerExplicit, voiceLlmModel: voiceLlm.model, agentProviderSettings: extractAgentProviderSettings(providerOverrides), + providerCatalogRefreshTimeoutMs: persisted.agents?.catalogRefreshTimeoutMs, metadataGeneration: persisted.agents?.metadataGeneration, providerOverrides, log: resolveLogConfigFromEnv(env, persisted), diff --git a/packages/server/src/server/file-observer/index.test.ts b/packages/server/src/server/file-observer/index.test.ts index 112302321..0427cd823 100644 --- a/packages/server/src/server/file-observer/index.test.ts +++ b/packages/server/src/server/file-observer/index.test.ts @@ -237,7 +237,7 @@ test("observes a thousand concurrent writes and remains healthy after delete and await Promise.all(paths.map((path, index) => writeFile(path, `${index}`))); await expect - .poll(() => paths.filter((path) => !observed.has(path)), { timeout: 10_000 }) + .poll(() => paths.filter((path) => !observed.has(path)), { timeout: 60_000 }) .toEqual([]); const removedPaths = paths.slice(0, 100); @@ -258,7 +258,7 @@ test("observes a thousand concurrent writes and remains healthy after delete and await writeFile(sentinel, "alive"); await expect.poll(() => observed.has(sentinel)).toBe(true); await subscription.unsubscribe(); -}, 30_000); +}, 90_000); test("unsubscribe cancels reconciliation queued by a write burst", async () => { const root = await createRoot(); diff --git a/packages/server/src/server/persisted-config.test.ts b/packages/server/src/server/persisted-config.test.ts index 140bc0f40..8ea1c5c25 100644 --- a/packages/server/src/server/persisted-config.test.ts +++ b/packages/server/src/server/persisted-config.test.ts @@ -251,6 +251,20 @@ describe("PersistedConfigSchema agent provider runtime settings", () => { ], }); }); + + test("accepts a custom provider catalog refresh timeout", () => { + const parsed = PersistedConfigSchema.parse({ + agents: { catalogRefreshTimeoutMs: 180_000 }, + }); + + expect(parsed.agents?.catalogRefreshTimeoutMs).toBe(180_000); + }); + + test("rejects provider catalog refresh timeouts that overflow Node timers", () => { + expect(() => + PersistedConfigSchema.parse({ agents: { catalogRefreshTimeoutMs: 2_147_483_648 } }), + ).toThrow(); + }); }); describe("provider overrides (new format)", () => { diff --git a/packages/server/src/server/persisted-config.ts b/packages/server/src/server/persisted-config.ts index 9a4dbd11b..e83b02603 100644 --- a/packages/server/src/server/persisted-config.ts +++ b/packages/server/src/server/persisted-config.ts @@ -308,6 +308,7 @@ export const PersistedConfigSchema = z agents: z .object({ providers: z.preprocess(normalizeAgentProviders, ProviderOverridesSchema).optional(), + catalogRefreshTimeoutMs: z.number().int().positive().max(2_147_483_647).optional(), metadataGeneration: AgentMetadataGenerationSchema.optional(), }) .strict() diff --git a/packages/server/src/utils/spawn.ts b/packages/server/src/utils/spawn.ts index ac27572c2..8fac115f0 100644 --- a/packages/server/src/utils/spawn.ts +++ b/packages/server/src/utils/spawn.ts @@ -27,6 +27,7 @@ interface ExecCommandOptions extends ExternalEnvOptions { timeout?: number; maxBuffer?: number; shell?: boolean | string; + signal?: AbortSignal; } interface ExecCommandResult { @@ -77,6 +78,7 @@ export function spawnProcess( ...spawnOptions, env: childEnv, shell, + signal: options?.signal, windowsHide: true, }); } diff --git a/packages/website/public/schemas/paseo.config.v1.json b/packages/website/public/schemas/paseo.config.v1.json index 033b49613..c8dc40a75 100644 --- a/packages/website/public/schemas/paseo.config.v1.json +++ b/packages/website/public/schemas/paseo.config.v1.json @@ -335,6 +335,11 @@ }, "additionalProperties": false } + }, + "catalogRefreshTimeoutMs": { + "type": "integer", + "exclusiveMinimum": 0, + "maximum": 2147483647 } }, "additionalProperties": false