Fix provider refresh scope and self-healing

This commit is contained in:
Mohamed Boudra
2026-04-21 12:41:28 +07:00
parent 92e701be8f
commit b01e274cce
54 changed files with 2057 additions and 694 deletions
@@ -2745,7 +2745,7 @@ export class AgentManager {
const client = this.clients.get(normalized.provider);
if (client) {
try {
const models = await client.listModels();
const models = await client.listModels({ cwd: normalized.cwd, force: false });
const defaultModel = models.find((model) => model.isDefault) ?? models[0];
if (defaultModel) {
normalized.model = defaultModel.id;
@@ -8,6 +8,7 @@ import { experimental_createMCPClient } from "ai";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import pino from "pino";
import { withTimeout } from "../../utils/promise-timeout.js";
import { createPaseoDaemon, type PaseoDaemonConfig } from "../bootstrap.js";
import { createTestAgentClients } from "../test-utils/fake-agent-client.js";
@@ -23,26 +24,6 @@ type McpClient = {
close: () => Promise<void>;
};
async function withTimeout<T>(options: {
promise: Promise<T>;
timeoutMs: number;
label: string;
}): Promise<T> {
let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
const timeout = new Promise<never>((_, reject) => {
timeoutHandle = setTimeout(() => {
reject(new Error(`Timed out after ${options.timeoutMs}ms (${options.label})`));
}, options.timeoutMs);
});
try {
return await Promise.race([options.promise, timeout]);
} finally {
if (timeoutHandle) {
clearTimeout(timeoutHandle);
}
}
}
async function waitForPathExists(options: {
targetPath: string;
timeoutMs: number;
@@ -490,11 +490,13 @@ export interface AgentSession {
}
export interface ListModelsOptions {
cwd?: string;
cwd: string;
force: boolean;
}
export interface ListModesOptions {
cwd?: string;
cwd: string;
force: boolean;
}
export interface AgentClient {
@@ -509,8 +511,8 @@ export interface AgentClient {
overrides?: Partial<AgentSessionConfig>,
launchContext?: AgentLaunchContext,
): Promise<AgentSession>;
listModels(options?: ListModelsOptions): Promise<AgentModelDefinition[]>;
listModes?(options?: ListModesOptions): Promise<AgentMode[]>;
listModels(options: ListModelsOptions): Promise<AgentModelDefinition[]>;
listModes?(options: ListModesOptions): Promise<AgentMode[]>;
listPersistedAgents?(options?: ListPersistedAgentsOptions): Promise<PersistedAgentDescriptor[]>;
/**
* Check if this provider is available (CLI binary is installed).
+21 -4
View File
@@ -31,6 +31,7 @@ import { runAsyncWorktreeBootstrap } from "../worktree-bootstrap.js";
import type { ScheduleService } from "../schedule/service.js";
import { ScheduleSummarySchema, StoredScheduleSchema } from "../schedule/types.js";
import type { ProviderDefinition } from "./provider-registry.js";
import { resolveSnapshotCwd } from "./provider-snapshot-manager.js";
import {
AgentModelSchema,
AgentProviderEnum,
@@ -1296,11 +1297,27 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
const scheduleTarget =
target === "self"
? (() => {
if (!callerAgentId) {
const callerAgent = resolveCallerAgent();
if (!callerAgentId || !callerAgent) {
throw new Error("target=self requires a caller agent");
}
if (provider !== undefined || cwd !== undefined) {
throw new Error("provider and cwd can only be used with target=new-agent");
const trimmedCwd = cwd?.trim();
if (trimmedCwd && expandUserPath(trimmedCwd) !== callerAgent.cwd) {
throw new Error("cwd can only differ from the caller agent when target=new-agent");
}
if (provider !== undefined) {
const resolved = resolveProviderAndModel({
provider,
defaultProvider: callerAgent.provider,
});
if (
resolved.provider !== callerAgent.provider ||
(resolved.model !== undefined && resolved.model !== callerAgent.config.model)
) {
throw new Error(
"provider can only differ from the caller agent when target=new-agent",
);
}
}
return { type: "agent" as const, agentId: callerAgentId };
})()
@@ -1498,7 +1515,7 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
throw new Error(`Provider ${provider} is not configured`);
}
const models = await definition.fetchModels();
const models = await definition.fetchModels({ cwd: resolveSnapshotCwd(), force: false });
return {
content: [],
structuredContent: ensureValidJson({
@@ -1,4 +1,5 @@
import { buildProviderRegistry } from "./provider-registry.js";
import { resolveSnapshotCwd } from "./provider-snapshot-manager.js";
import type { AgentProvider } from "./agent-sdk-types.js";
import { expandTilde } from "../../utils/path.js";
import type { Logger } from "pino";
@@ -21,7 +22,8 @@ export async function resolveAgentModel(
try {
const providerRegistry = buildProviderRegistry(options.logger);
const models = await providerRegistry[options.provider].fetchModels({
cwd: options.cwd ? expandTilde(options.cwd) : undefined,
cwd: resolveSnapshotCwd(options.cwd ? expandTilde(options.cwd) : undefined),
force: false,
});
const preferred = models.find((model) => model.isDefault) ?? models[0];
return preferred?.id;
@@ -578,7 +578,10 @@ describe("buildProviderRegistry", () => {
},
});
const models = await registry.claude.fetchModels();
const models = await registry.claude.fetchModels({
cwd: "/tmp/registry-models",
force: false,
});
expect(models.map((model) => model.id)).toEqual(["profile-fast"]);
});
@@ -610,7 +613,10 @@ describe("buildProviderRegistry", () => {
},
});
const models = await registry.claude.fetchModels();
const models = await registry.claude.fetchModels({
cwd: "/tmp/registry-models",
force: false,
});
expect(models).toEqual([
{
@@ -645,7 +651,10 @@ describe("buildProviderRegistry", () => {
},
});
const models = await registry.claude.fetchModels();
const models = await registry.claude.fetchModels({
cwd: "/tmp/registry-models",
force: false,
});
expect(models).toEqual([
{
@@ -668,7 +677,10 @@ describe("buildProviderRegistry", () => {
]);
const registry = buildProviderRegistry(logger);
const models = await registry.claude.fetchModels();
const models = await registry.claude.fetchModels({
cwd: "/tmp/registry-models",
force: false,
});
expect(models).toEqual([
{
@@ -42,8 +42,8 @@ export { AGENT_PROVIDER_DEFINITIONS, getAgentProviderDefinition };
export interface ProviderDefinition extends AgentProviderDefinition {
createClient: (logger: Logger) => AgentClient;
fetchModels: (options?: ListModelsOptions) => Promise<AgentModelDefinition[]>;
fetchModes: (options?: ListModesOptions) => Promise<AgentMode[]>;
fetchModels: (options: ListModelsOptions) => Promise<AgentModelDefinition[]>;
fetchModes: (options: ListModesOptions) => Promise<AgentMode[]>;
}
export type BuildProviderRegistryOptions = {
@@ -329,9 +329,9 @@ function createRegistryEntry(
const inner = resolved.createBaseClient(providerLogger);
return inner.provider === provider ? inner : wrapClientProvider(provider, inner);
},
fetchModels: async (options?: ListModelsOptions) =>
fetchModels: async (options: ListModelsOptions) =>
mergeModels(provider, resolved.profileModels, await modelClient.listModels(options)),
fetchModes: async (options?: ListModesOptions) => {
fetchModes: async (options: ListModesOptions) => {
const modes = modelClient.listModes
? await modelClient.listModes(options)
: resolved.definition.modes;
@@ -1,3 +1,4 @@
import { homedir } from "node:os";
import { resolve } from "node:path";
import { describe, expect, test, vi } from "vitest";
@@ -7,6 +8,8 @@ import type {
AgentMode,
AgentModelDefinition,
AgentProvider,
ListModesOptions,
ListModelsOptions,
ProviderSnapshotEntry,
} from "./agent-sdk-types.js";
import type { ProviderDefinition } from "./provider-registry.js";
@@ -24,8 +27,8 @@ type MockProviderOptions = {
description?: string;
defaultModeId?: string | null;
isAvailable?: () => Promise<boolean>;
fetchModels?: (cwd?: string) => Promise<AgentModelDefinition[]>;
fetchModes?: (cwd?: string) => Promise<AgentMode[]>;
fetchModels?: (cwd: string, force: boolean) => Promise<AgentModelDefinition[]>;
fetchModes?: (cwd: string, force: boolean) => Promise<AgentMode[]>;
};
type MockProviderHandle = {
@@ -441,21 +444,22 @@ describe("ProviderSnapshotManager", () => {
manager.destroy();
});
test("refresh during an in-flight refresh is a no-op", async () => {
const fetchModels = deferred<AgentModelDefinition[]>();
const fetchModes = deferred<AgentMode[]>();
test("explicit refresh bypasses an in-flight background warm-up", async () => {
const initialFetchModels = deferred<AgentModelDefinition[]>();
const explicitFetchModels = deferred<AgentModelDefinition[]>();
const { registry, handles } = createRegistry([
createMockProvider({
provider: "codex",
fetchModels: async () => fetchModels.promise,
fetchModes: async () => fetchModes.promise,
fetchModels: vi
.fn<() => Promise<AgentModelDefinition[]>>()
.mockImplementationOnce(async () => initialFetchModels.promise)
.mockImplementationOnce(async () => explicitFetchModels.promise),
fetchModes: async () => [createMode("auto")],
}),
]);
const manager = new ProviderSnapshotManager(registry, createTestLogger());
const changes: ProviderSnapshotEntry[][] = [];
manager.on("change", (entries) => changes.push(entries));
manager.refresh({ cwd: projectCwd });
manager.getSnapshot(projectCwd);
expect(manager.getSnapshot(projectCwd)).toEqual([
{
@@ -467,15 +471,18 @@ describe("ProviderSnapshotManager", () => {
},
]);
manager.refresh({ cwd: projectCwd });
manager.refresh({ cwd: projectCwd });
manager.refresh({ cwd: projectCwd });
await vi.waitFor(() => {
expect(handles.codex?.fetchModels).toHaveBeenCalledTimes(1);
});
expect(changes).toHaveLength(1);
expect(handles.codex?.isAvailable).toHaveBeenCalledTimes(1);
const refreshPromise = manager.refresh({ cwd: projectCwd, providers: ["codex"] });
fetchModels.resolve([createModel("codex", "gpt-5.2")]);
fetchModes.resolve([createMode("auto")]);
await vi.waitFor(() => {
expect(handles.codex?.fetchModels).toHaveBeenCalledTimes(2);
});
explicitFetchModels.resolve([createModel("codex", "gpt-5.2")]);
await refreshPromise;
await vi.waitFor(() => {
expect(getProviderEntry(manager.getSnapshot(projectCwd), "codex")).toMatchObject({
@@ -486,12 +493,126 @@ describe("ProviderSnapshotManager", () => {
});
});
expect(handles.codex?.fetchModels).toHaveBeenCalledTimes(1);
expect(handles.codex?.fetchModes).toHaveBeenCalledTimes(1);
initialFetchModels.resolve([createModel("codex", "stale-background-model")]);
await Promise.resolve();
expect(getProviderEntry(manager.getSnapshot(projectCwd), "codex")?.models?.[0]?.id).toBe(
"gpt-5.2",
);
expect(handles.codex?.fetchModels).toHaveBeenCalledTimes(2);
expect(handles.codex?.fetchModes).toHaveBeenCalledTimes(2);
manager.destroy();
});
test("warmUpSnapshotForCwd awaits an in-flight loading provider without force", async () => {
const loadingFetchModels = deferred<AgentModelDefinition[]>();
const fetchModels = vi
.fn<(cwd: string, force: boolean) => Promise<AgentModelDefinition[]>>()
.mockImplementation(async (_cwd, _force) => loadingFetchModels.promise);
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
fetchModels: async (cwd, force) => fetchModels(cwd, force),
fetchModes: async () => [createMode("auto")],
}),
]);
const manager = new ProviderSnapshotManager(registry, createTestLogger());
manager.getSnapshot(projectCwd);
await vi.waitFor(() => {
expect(fetchModels).toHaveBeenCalledTimes(1);
});
const warmUpPromise = manager.warmUpSnapshotForCwd({
cwd: projectCwd,
providers: ["codex"],
});
await Promise.resolve();
expect(fetchModels).toHaveBeenCalledTimes(1);
expect(fetchModels).toHaveBeenCalledWith(projectCwd, false);
expect(fetchModels).not.toHaveBeenCalledWith(projectCwd, true);
loadingFetchModels.resolve([createModel("codex", "gpt-5.4")]);
await warmUpPromise;
expect(fetchModels).toHaveBeenCalledTimes(1);
expect(fetchModels).not.toHaveBeenCalledWith(projectCwd, true);
manager.destroy();
});
test("settings refresh refreshes only home cwd and invalidates matching provider elsewhere", async () => {
const fetchModels = vi
.fn<(cwd: string, force: boolean) => Promise<AgentModelDefinition[]>>()
.mockImplementation(async (cwd) => [createModel("codex", cwd)]);
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
fetchModels: async (cwd, force) => fetchModels(cwd, force),
fetchModes: async () => [createMode("auto")],
}),
createMockProvider({
provider: "claude",
fetchModels: async (cwd) => [createModel("claude", cwd)],
fetchModes: async () => [createMode("default")],
}),
]);
const manager = new ProviderSnapshotManager(registry, createTestLogger());
manager.getSnapshot(projectACwd);
manager.getSnapshot(projectBCwd);
await vi.waitFor(() => {
expect(getProviderEntry(manager.getSnapshot(projectACwd), "codex")?.status).toBe("ready");
expect(getProviderEntry(manager.getSnapshot(projectBCwd), "codex")?.status).toBe("ready");
});
await manager.refreshSettingsSnapshot({ providers: ["codex"] });
expect(fetchModels.mock.calls.map(([cwd]) => cwd)).toEqual([
projectACwd,
projectBCwd,
homedir(),
]);
expect(fetchModels.mock.calls.map(([, force]) => force)).toEqual([false, false, true]);
const projectASnapshot = manager.getSnapshot(projectACwd);
expect(getProviderEntry(projectASnapshot, "codex")?.status).toBe("loading");
expect(getProviderEntry(projectASnapshot, "claude")?.status).toBe("ready");
manager.destroy();
});
test("refresh marks a slow provider as error after the timeout", async () => {
const fetchModels = deferred<AgentModelDefinition[]>();
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
fetchModels: async () => fetchModels.promise,
fetchModes: async () => [createMode("auto")],
}),
]);
const manager = new ProviderSnapshotManager(registry, createTestLogger(), {
refreshTimeoutMs: 5,
});
await manager.refresh({ cwd: projectCwd, providers: ["codex"] });
expect(getProviderEntry(manager.getSnapshot(projectCwd), "codex")).toMatchObject({
provider: "codex",
status: "error",
error: "Timed out refreshing codex after 5ms",
});
manager.destroy();
fetchModels.resolve([createModel("codex", "gpt-5.2")]);
});
test("getSnapshot returns stale ready entries and starts background warm-up when snapshot is older than TTL", async () => {
let now = 1_000;
const fetchModels = vi
@@ -799,6 +920,91 @@ describe("ProviderSnapshotManager", () => {
manager.destroy();
});
test("missing cwd resolves to home and shares the explicit home cache entry", async () => {
const seenCwds: string[] = [];
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
fetchModels: async (cwd) => {
seenCwds.push(cwd);
return [createModel("codex", cwd)];
},
}),
]);
const manager = new ProviderSnapshotManager(registry, createTestLogger());
manager.getSnapshot();
await vi.waitFor(() => {
expect(getProviderEntry(manager.getSnapshot(homedir()), "codex")?.status).toBe("ready");
});
manager.getSnapshot(" ");
expect(seenCwds).toEqual([homedir()]);
expect(getProviderEntry(manager.getSnapshot(), "codex")?.models?.[0]?.id).toBe(homedir());
manager.destroy();
});
test("cwd normalization resolves tilde relative paths and trailing slashes before provider calls", async () => {
const seenCwds: string[] = [];
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
fetchModels: async (cwd) => {
seenCwds.push(cwd);
return [createModel("codex", cwd)];
},
}),
]);
const manager = new ProviderSnapshotManager(registry, createTestLogger());
manager.getSnapshot("~/paseo-provider-test/../paseo-provider-test/");
manager.getSnapshot("relative-provider-test/..");
await vi.waitFor(() => {
expect(seenCwds).toHaveLength(2);
});
expect(seenCwds).toEqual([resolve(homedir(), "paseo-provider-test"), resolve(".")]);
manager.destroy();
});
test("workspace refresh refreshes only the requested cwd with force true", async () => {
const fetchModels = vi
.fn<(cwd: string, force: boolean) => Promise<AgentModelDefinition[]>>()
.mockImplementation(async (cwd) => [createModel("codex", cwd)]);
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
fetchModels: async (cwd, force) => fetchModels(cwd, force),
}),
]);
const manager = new ProviderSnapshotManager(registry, createTestLogger());
manager.getSnapshot(projectACwd);
manager.getSnapshot(projectBCwd);
await vi.waitFor(() => {
expect(fetchModels).toHaveBeenCalledTimes(2);
});
await manager.refreshSnapshotForCwd({ cwd: projectACwd, providers: ["codex"] });
expect(fetchModels.mock.calls).toEqual([
[projectACwd, false],
[projectBCwd, false],
[projectACwd, true],
]);
expect(getProviderEntry(manager.getSnapshot(projectBCwd), "codex")?.models?.[0]?.id).toBe(
projectBCwd,
);
manager.destroy();
});
test("snapshot includes user-defined providers from the registry", async () => {
const { registry } = createRegistry([
createMockProvider({ provider: "claude" }),
@@ -911,14 +1117,16 @@ function createRegistry(handles: MockProviderHandle[]): {
function createMockProvider(options: MockProviderOptions): MockProviderHandle {
const isAvailable = vi.fn(async () => options.isAvailable?.() ?? true);
const fetchModels = vi.fn(
async (listOptions?: { cwd?: string }) =>
options.fetchModels?.(listOptions?.cwd) ?? [
async (listOptions: ListModelsOptions) =>
options.fetchModels?.(listOptions.cwd, listOptions.force) ?? [
createModel(options.provider, `${options.provider}-default`),
],
);
const fetchModes = vi.fn(
async (listOptions?: { cwd?: string }) =>
options.fetchModes?.(listOptions?.cwd) ?? [createMode(`${options.provider}-mode`)],
async (listOptions: ListModesOptions) =>
options.fetchModes?.(listOptions.cwd, listOptions.force) ?? [
createMode(`${options.provider}-mode`),
],
);
const definition: ProviderDefinition = {
@@ -937,7 +1145,7 @@ function createMockProvider(options: MockProviderOptions): MockProviderHandle {
async resumeSession() {
throw new Error("not implemented");
},
async listModels() {
async listModels(_options: ListModelsOptions) {
return [];
},
async isAvailable() {
@@ -1,31 +1,43 @@
import { EventEmitter } from "node:events";
import { homedir } from "node:os";
import { resolve } from "node:path";
import type { Logger } from "pino";
import { withTimeout } from "../../utils/promise-timeout.js";
import type { AgentProvider, ProviderSnapshotEntry } from "./agent-sdk-types.js";
import type { ProviderDefinition } from "./provider-registry.js";
const DEFAULT_CWD_KEY = "__default__";
const DEFAULT_SNAPSHOT_TTL_MS = 300_000;
const DEFAULT_REFRESH_TIMEOUT_MS = 30_000;
type ProviderSnapshotChangeListener = (entries: ProviderSnapshotEntry[], cwd?: string) => void;
type ProviderSnapshotChangeListener = (entries: ProviderSnapshotEntry[], cwd: string) => void;
type ProviderSnapshotManagerOptions = {
ttlMs?: number;
refreshTimeoutMs?: number;
now?: () => number;
};
type ProviderSnapshotRefreshOptions = {
cwd?: string;
cwd: string;
providers?: AgentProvider[];
};
type ProviderLoadOptions = {
cwd: string;
providers: AgentProvider[];
force: boolean;
};
type ProviderLoad = {
promise: Promise<void>;
};
export class ProviderSnapshotManager {
private readonly snapshots = new Map<string, Map<AgentProvider, ProviderSnapshotEntry>>();
private readonly lastCheckedAts = new Map<string, number>();
private readonly warmUps = new Map<string, Promise<void>>();
private readonly providerLoads = new Map<string, Map<AgentProvider, ProviderLoad>>();
private readonly events = new EventEmitter();
private destroyed = false;
private readonly ttlMs: number;
private readonly refreshTimeoutMs: number;
private readonly now: () => number;
constructor(
@@ -34,37 +46,81 @@ export class ProviderSnapshotManager {
options: ProviderSnapshotManagerOptions = {},
) {
this.ttlMs = options.ttlMs ?? DEFAULT_SNAPSHOT_TTL_MS;
this.refreshTimeoutMs = options.refreshTimeoutMs ?? DEFAULT_REFRESH_TIMEOUT_MS;
this.now = options.now ?? Date.now;
}
getSnapshot(cwd?: string): ProviderSnapshotEntry[] {
const cwdKey = normalizeCwdKey(cwd);
const entries = this.snapshots.get(cwdKey);
const resolvedCwd = resolveSnapshotCwd(cwd);
const entries = this.snapshots.get(resolvedCwd);
if (!entries) {
const loadingEntries = this.resetSnapshotToLoading(cwdKey);
void this.warmUp(cwd);
const loadingEntries = this.resetSnapshotToLoading(resolvedCwd);
void this.warmUp(resolvedCwd);
return entriesToArray(loadingEntries);
}
if (this.shouldRevalidate(cwdKey)) {
void this.warmUp(cwd);
const missingProviders = this.getProviderIds().filter((provider) => !entries.has(provider));
if (missingProviders.length > 0) {
this.resetSnapshotToLoading(resolvedCwd, missingProviders);
void this.warmUp(resolvedCwd, missingProviders);
}
if (this.shouldRevalidate(resolvedCwd)) {
void this.warmUp(resolvedCwd);
}
return entriesToArray(entries);
}
async refresh(options: ProviderSnapshotRefreshOptions = {}): Promise<void> {
const { cwd } = options;
const cwdKey = normalizeCwdKey(cwd);
const inFlight = this.warmUps.get(cwdKey);
if (inFlight) {
await inFlight;
async refreshSnapshotForCwd(options: ProviderSnapshotRefreshOptions): Promise<void> {
const cwd = resolveSnapshotCwd(options.cwd);
const providers = this.resolveRefreshProviders(options.providers);
this.resetSnapshotToLoading(cwd, providers);
this.emitChange(cwd);
await this.refreshProviders(cwd, providers ?? this.getProviderIds());
if (!providers) {
this.lastCheckedAts.set(cwd, this.now());
}
}
async refreshSettingsSnapshot(
options: Omit<ProviderSnapshotRefreshOptions, "cwd"> = {},
): Promise<void> {
const homeCwd = resolveSnapshotCwd();
const providers = this.resolveRefreshProviders(options.providers);
const providersToRefresh = providers ?? this.getProviderIds();
this.resetSnapshotToLoading(homeCwd, providers);
this.emitChange(homeCwd);
await this.refreshProviders(homeCwd, providersToRefresh);
if (!providers) {
this.lastCheckedAts.set(homeCwd, this.now());
}
this.invalidateNonHomeSnapshots({ homeCwd, providers });
}
async warmUpSnapshotForCwd(options: ProviderSnapshotRefreshOptions): Promise<void> {
const cwd = resolveSnapshotCwd(options.cwd);
const providers = this.resolveRefreshProviders(options.providers);
if (options.providers && providers?.length === 0) {
return;
}
const providers = this.resolveRefreshProviders(options.providers);
this.resetSnapshotToLoading(cwdKey, providers);
this.emitChange(cwdKey);
const snapshot = this.snapshots.get(cwd);
if (!snapshot) {
this.resetSnapshotToLoading(cwd, providers);
} else if (providers) {
const missingProviders = providers.filter((provider) => !snapshot.has(provider));
if (missingProviders.length > 0) {
this.resetSnapshotToLoading(cwd, missingProviders);
}
}
await this.warmUp(cwd, providers);
}
async refresh(options: ProviderSnapshotRefreshOptions): Promise<void> {
await this.refreshSnapshotForCwd(options);
}
on(event: "change", listener: ProviderSnapshotChangeListener): this {
this.events.on(event, listener);
return this;
@@ -80,7 +136,7 @@ export class ProviderSnapshotManager {
this.events.removeAllListeners();
this.snapshots.clear();
this.lastCheckedAts.clear();
this.warmUps.clear();
this.providerLoads.clear();
}
private createLoadingEntries(): Map<AgentProvider, ProviderSnapshotEntry> {
@@ -98,49 +154,87 @@ export class ProviderSnapshotManager {
return entries;
}
private async warmUp(cwd?: string, providers?: AgentProvider[]): Promise<void> {
const cwdKey = normalizeCwdKey(cwd);
const inFlight = this.warmUps.get(cwdKey);
if (inFlight) {
return inFlight;
}
private async warmUp(cwd: string, providers?: AgentProvider[]): Promise<void> {
const providersToRefresh = providers ?? this.getProviderIds();
const warmUpPromise = Promise.allSettled(
providersToRefresh.map((provider) => this.refreshProvider(cwdKey, provider, cwd)),
).then(() => {
if (!providers) {
this.lastCheckedAts.set(cwdKey, this.now());
}
await this.loadProviders({
cwd,
providers: providersToRefresh,
force: false,
});
this.warmUps.set(cwdKey, warmUpPromise);
try {
await warmUpPromise;
} finally {
if (this.warmUps.get(cwdKey) === warmUpPromise) {
this.warmUps.delete(cwdKey);
}
if (!providers) {
this.lastCheckedAts.set(cwd, this.now());
}
}
private async refreshProvider(
cwdKey: string,
provider: AgentProvider,
cwd?: string,
): Promise<void> {
const definition = this.providerRegistry[provider];
private async refreshProviders(cwd: string, providers: AgentProvider[]): Promise<void> {
await this.loadProviders({ cwd, providers, force: true });
}
private async loadProviders(options: ProviderLoadOptions): Promise<void> {
await Promise.allSettled(
options.providers.map((provider) => this.loadProvider({ ...options, provider })),
);
}
private loadProvider(options: ProviderLoadOptions & { provider: AgentProvider }): Promise<void> {
const definition = this.providerRegistry[options.provider];
if (!definition) {
return;
return Promise.resolve();
}
const snapshot = this.getOrCreateSnapshot(cwdKey);
const existingLoad = this.getProviderLoad(options.cwd, options.provider);
if (existingLoad && !options.force) {
return existingLoad.promise;
}
const load: ProviderLoad = {
promise: Promise.resolve(),
};
this.setProviderLoad(options.cwd, options.provider, load);
load.promise = Promise.resolve()
.then(() =>
this.refreshProvider({
cwd: options.cwd,
provider: options.provider,
definition,
load,
force: options.force,
}),
)
.finally(() => {
const providerLoads = this.providerLoads.get(options.cwd);
if (providerLoads?.get(options.provider) === load) {
providerLoads.delete(options.provider);
}
if (providerLoads?.size === 0) {
this.providerLoads.delete(options.cwd);
}
});
return load.promise;
}
private async refreshProvider(options: {
cwd: string;
provider: AgentProvider;
definition: ProviderDefinition;
load: ProviderLoad;
force: boolean;
}): Promise<void> {
const { cwd, provider, definition, load, force } = options;
const snapshot = this.getOrCreateSnapshot(options.cwd);
try {
const client = definition.createClient(this.logger);
const available = await client.isAvailable();
const available = await withTimeout(
client.isAvailable(),
this.refreshTimeoutMs,
`Timed out checking ${definition.label} availability after ${this.refreshTimeoutMs}ms`,
);
if (!available) {
if (!this.isCurrentProviderLoad(cwd, provider, load)) {
return;
}
snapshot.set(provider, {
provider,
status: "unavailable",
@@ -148,15 +242,22 @@ export class ProviderSnapshotManager {
description: definition.description,
defaultModeId: definition.defaultModeId,
});
this.emitChange(cwdKey);
this.emitChange(cwd);
return;
}
const [models, modes] = await Promise.all([
definition.fetchModels({ cwd }),
definition.fetchModes({ cwd }),
]);
const [models, modes] = await withTimeout(
Promise.all([
definition.fetchModels({ cwd, force }),
definition.fetchModes({ cwd, force }),
]),
this.refreshTimeoutMs,
`Timed out refreshing ${definition.label} after ${this.refreshTimeoutMs}ms`,
);
if (!this.isCurrentProviderLoad(cwd, provider, load)) {
return;
}
snapshot.set(provider, {
provider,
status: "ready",
@@ -167,8 +268,11 @@ export class ProviderSnapshotManager {
description: definition.description,
defaultModeId: definition.defaultModeId,
});
this.emitChange(cwdKey);
this.emitChange(cwd);
} catch (error) {
if (!this.isCurrentProviderLoad(cwd, provider, load)) {
return;
}
snapshot.set(provider, {
provider,
status: "error",
@@ -177,14 +281,32 @@ export class ProviderSnapshotManager {
description: definition.description,
defaultModeId: definition.defaultModeId,
});
this.logger.warn(
{ err: error, provider, cwd: cwdKey },
"Failed to refresh provider snapshot",
);
this.emitChange(cwdKey);
this.logger.warn({ err: error, provider, cwd }, "Failed to refresh provider snapshot");
this.emitChange(cwd);
}
}
private getProviderLoad(cwdKey: string, provider: AgentProvider): ProviderLoad | undefined {
return this.providerLoads.get(cwdKey)?.get(provider);
}
private setProviderLoad(cwdKey: string, provider: AgentProvider, load: ProviderLoad): void {
let providerLoads = this.providerLoads.get(cwdKey);
if (!providerLoads) {
providerLoads = new Map<AgentProvider, ProviderLoad>();
this.providerLoads.set(cwdKey, providerLoads);
}
providerLoads.set(provider, load);
}
private isCurrentProviderLoad(
cwdKey: string,
provider: AgentProvider,
load: ProviderLoad,
): boolean {
return this.providerLoads.get(cwdKey)?.get(provider) === load;
}
private emitChange(cwdKey: string): void {
if (this.destroyed) {
return;
@@ -193,11 +315,11 @@ export class ProviderSnapshotManager {
if (!snapshot) {
return;
}
this.events.emit("change", entriesToArray(snapshot), denormalizeCwdKey(cwdKey));
this.events.emit("change", entriesToArray(snapshot), cwdKey);
}
private shouldRevalidate(cwdKey: string): boolean {
if (this.warmUps.has(cwdKey)) {
if (this.providerLoads.has(cwdKey)) {
return false;
}
const lastCheckedAt = this.lastCheckedAts.get(cwdKey);
@@ -259,23 +381,51 @@ export class ProviderSnapshotManager {
const providerIds = new Set(this.getProviderIds());
return Array.from(new Set(providers)).filter((provider) => providerIds.has(provider));
}
private invalidateNonHomeSnapshots(options: {
homeCwd: string;
providers?: AgentProvider[];
}): void {
for (const cwd of Array.from(this.snapshots.keys())) {
if (cwd === options.homeCwd) {
continue;
}
if (!options.providers) {
this.snapshots.delete(cwd);
this.lastCheckedAts.delete(cwd);
this.providerLoads.delete(cwd);
this.events.emit("change", [], cwd);
continue;
}
const snapshot = this.snapshots.get(cwd);
if (!snapshot) {
continue;
}
let changed = false;
for (const provider of options.providers) {
changed = snapshot.delete(provider) || changed;
this.providerLoads.get(cwd)?.delete(provider);
}
if (this.providerLoads.get(cwd)?.size === 0) {
this.providerLoads.delete(cwd);
}
if (changed) {
this.emitChange(cwd);
}
}
}
}
function normalizeCwdKey(cwd?: string): string {
if (!cwd) {
return DEFAULT_CWD_KEY;
}
const trimmed = cwd.trim();
export function resolveSnapshotCwd(cwd?: string | null): string {
const trimmed = cwd?.trim();
if (!trimmed) {
return DEFAULT_CWD_KEY;
return homedir();
}
return resolve(trimmed);
}
function denormalizeCwdKey(cwdKey: string): string | undefined {
return cwdKey === DEFAULT_CWD_KEY ? undefined : cwdKey;
const expanded =
trimmed === "~" || trimmed.startsWith("~/") ? `${homedir()}${trimmed.slice(1)}` : trimmed;
return resolve(expanded);
}
function entriesToArray(
@@ -3,6 +3,7 @@ import { describe, expect, test, vi } from "vitest";
import {
ACPAgentClient,
ACPAgentSession,
type SpawnedACPProcess,
type SessionStateResponse,
deriveModelDefinitionsFromACP,
deriveModesFromACP,
@@ -218,7 +219,7 @@ describe("deriveModelDefinitionsFromACP", () => {
describe("ACPAgentClient modelTransformer", () => {
test("applies modelTransformer after deriving ACP models", async () => {
class TestACPAgentClient extends ACPAgentClient {
protected override async spawnProcess(): Promise<any> {
protected override async spawnProcess(): Promise<SpawnedACPProcess> {
return {
child: { kill: vi.fn(), exitCode: 0, signalCode: null, once: vi.fn() },
connection: {
@@ -237,7 +238,7 @@ describe("ACPAgentClient modelTransformer", () => {
}),
},
initialize: { agentCapabilities: {} },
};
} as unknown as SpawnedACPProcess;
}
protected override async closeProbe(): Promise<void> {}
@@ -250,7 +251,7 @@ describe("ACPAgentClient modelTransformer", () => {
modelTransformer: transformPiModels,
});
await expect(client.listModels()).resolves.toEqual([
await expect(client.listModels({ cwd: "/tmp/acp-models", force: false })).resolves.toEqual([
{
provider: "pi",
id: "openrouter/openai/gpt-4.1-mini",
@@ -266,7 +267,7 @@ describe("ACPAgentClient modelTransformer", () => {
describe("ACPAgentClient sessionResponseTransformer", () => {
class TestACPAgentClient extends ACPAgentClient {
protected override async spawnProcess(): Promise<any> {
protected override async spawnProcess(): Promise<SpawnedACPProcess> {
const response: SessionStateResponse = {
sessionId: "session-1",
modes: {
@@ -283,7 +284,7 @@ describe("ACPAgentClient sessionResponseTransformer", () => {
newSession: vi.fn().mockResolvedValue(response),
},
initialize: { agentCapabilities: {} },
};
} as unknown as SpawnedACPProcess;
}
protected override async closeProbe(): Promise<void> {}
@@ -304,7 +305,7 @@ describe("ACPAgentClient sessionResponseTransformer", () => {
}),
});
await expect(client.listModes()).resolves.toEqual([
await expect(client.listModes({ cwd: "/tmp/acp-modes", force: false })).resolves.toEqual([
{
id: "review",
label: "Review",
@@ -315,9 +316,44 @@ describe("ACPAgentClient sessionResponseTransformer", () => {
});
describe("ACPAgentClient listModes", () => {
test("passes the requested cwd to list model and mode probes", async () => {
const newSession = vi.fn().mockResolvedValue({ modes: null, models: null, configOptions: [] });
class TestACPAgentClient extends ACPAgentClient {
protected override async spawnProcess(): Promise<SpawnedACPProcess> {
return {
child: { kill: vi.fn(), exitCode: 0, signalCode: null, once: vi.fn() },
connection: { newSession },
initialize: { agentCapabilities: {} },
} as unknown as SpawnedACPProcess;
}
protected override async closeProbe(): Promise<void> {}
}
const client = new TestACPAgentClient({
provider: "pi",
logger: createTestLogger(),
defaultCommand: ["test-acp"],
defaultModes: [],
});
await client.listModels({ cwd: "/tmp/acp-model-cwd", force: false });
await client.listModes({ cwd: "/tmp/acp-mode-cwd", force: false });
expect(newSession).toHaveBeenNthCalledWith(1, {
cwd: "/tmp/acp-model-cwd",
mcpServers: [],
});
expect(newSession).toHaveBeenNthCalledWith(2, {
cwd: "/tmp/acp-mode-cwd",
mcpServers: [],
});
});
test("returns an empty array when no ACP modes are reported and fallback modes are empty", async () => {
class TestACPAgentClient extends ACPAgentClient {
protected override async spawnProcess(): Promise<any> {
protected override async spawnProcess(): Promise<SpawnedACPProcess> {
return {
child: { kill: vi.fn(), exitCode: 0, signalCode: null, once: vi.fn() },
connection: {
@@ -340,7 +376,7 @@ describe("ACPAgentClient listModes", () => {
}),
},
initialize: { agentCapabilities: {} },
};
} as unknown as SpawnedACPProcess;
}
protected override async closeProbe(): Promise<void> {}
@@ -353,7 +389,7 @@ describe("ACPAgentClient listModes", () => {
defaultModes: [],
});
await expect(client.listModes()).resolves.toEqual([]);
await expect(client.listModes({ cwd: "/tmp/acp-modes", force: false })).resolves.toEqual([]);
});
});
@@ -152,7 +152,7 @@ type ACPAgentSessionOptions = {
initialCommandsWaitTimeoutMs?: number;
};
type SpawnedACPProcess = {
export type SpawnedACPProcess = {
child: ChildProcessWithoutNullStreams;
connection: ClientSideConnection;
initialize: InitializeResponse;
@@ -394,8 +394,8 @@ export class ACPAgentClient implements AgentClient {
return session;
}
async listModels(options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
const cwd = options?.cwd ?? process.cwd();
async listModels(options: ListModelsOptions): Promise<AgentModelDefinition[]> {
const { cwd } = options;
const probe = await this.spawnProcess(PROBE_ENV);
try {
const response = await probe.connection.newSession({
@@ -414,8 +414,8 @@ export class ACPAgentClient implements AgentClient {
}
}
async listModes(options?: ListModesOptions): Promise<AgentMode[]> {
const cwd = options?.cwd ?? process.cwd();
async listModes(options: ListModesOptions): Promise<AgentMode[]> {
const { cwd } = options;
const probe = await this.spawnProcess(PROBE_ENV);
try {
const response = await probe.connection.newSession({
@@ -7,6 +7,7 @@ import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import type { AgentSession, AgentStreamEvent, ToolCallTimelineItem } from "../agent-sdk-types.js";
import { isCommandAvailable } from "../../../utils/executable.js";
import { withTimeout } from "../../../utils/promise-timeout.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
@@ -40,21 +41,7 @@ async function nextStreamEvent(
timeoutMs: number,
label: string,
): Promise<IteratorResult<AgentStreamEvent>> {
let timer: ReturnType<typeof setTimeout> | null = null;
try {
return await Promise.race([
stream.next(),
new Promise<never>((_, reject) => {
timer = setTimeout(() => {
reject(new Error(`Timed out waiting for ${label}`));
}, timeoutMs);
}),
]);
} finally {
if (timer) {
clearTimeout(timer);
}
}
return await withTimeout(stream.next(), timeoutMs, `Timed out waiting for ${label}`);
}
async function collectUntilTerminal(
@@ -337,7 +337,7 @@ describe("ClaudeAgentClient.listModels", () => {
test("returns hardcoded claude models", async () => {
const client = new ClaudeAgentClient({ logger });
const models = await client.listModels();
const models = await client.listModels({ cwd: "/tmp/claude-models", force: false });
expect(models.map((m) => m.id)).toEqual([
"claude-opus-4-7[1m]",
@@ -74,6 +74,7 @@ import type {
} from "../agent-sdk-types.js";
import { applyProviderEnv, type ProviderRuntimeSettings } from "../provider-launch-config.js";
import { findExecutable, isCommandAvailable } from "../../../utils/executable.js";
import { withTimeout } from "../../../utils/promise-timeout.js";
import { execCommand, spawnProcess } from "../../../utils/spawn.js";
import { getOrchestratorModeInstructions } from "../orchestrator-instructions.js";
@@ -1107,7 +1108,8 @@ export class ClaudeAgentClient implements AgentClient {
});
}
async listModels(_options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
async listModels(_options: ListModelsOptions): Promise<AgentModelDefinition[]> {
// Claude exposes a static catalog here; cwd/force are intentionally irrelevant.
return getClaudeModels();
}
@@ -1157,7 +1159,7 @@ export class ClaudeAgentClient implements AgentClient {
if (available) {
try {
const models = await this.listModels();
const models = await this.listModels({ cwd: os.homedir(), force: false });
modelsValue = String(models.length);
} catch (error) {
modelsValue = `Error - ${toDiagnosticErrorMessage(error)}`;
@@ -2097,12 +2099,7 @@ class ClaudeAgentSession implements AgentSession {
const startedAt = Date.now();
this.logger.trace({ label }, "Claude query operation wait start");
try {
await Promise.race([
promise,
new Promise<never>((_, reject) => {
setTimeout(() => reject(new Error("timeout")), 3_000);
}),
]);
await withTimeout(promise, 3_000, "timeout");
this.logger.trace(
{ label, durationMs: Date.now() - startedAt },
"Claude query operation settled",
@@ -222,21 +222,26 @@ describe("Codex app-server provider (e2e)", () => {
"lists models and runs a simple prompt",
async () => {
const client = new CodexAppServerAgentClient(createTestLogger());
const models = await client.listModels();
const cwd = mkdtempSync(path.join(os.tmpdir(), "codex-app-server-e2e-"));
const models = await client.listModels({ cwd, force: false });
expect(models.some((m) => m.id.includes("gpt-5.1-codex"))).toBe(true);
const session = await client.createSession({
provider: "codex",
cwd: mkdtempSync(path.join(os.tmpdir(), "codex-app-server-e2e-")),
cwd,
modeId: "auto",
model: CODEX_TEST_MODEL,
thinkingOptionId: CODEX_TEST_THINKING_OPTION_ID,
});
expect(session.features?.some((feature) => feature.id === "plan_mode")).toBe(true);
try {
expect(session.features?.some((feature) => feature.id === "plan_mode")).toBe(true);
const result = await session.run("Say hello in one sentence.");
expect(result.finalText.length).toBeGreaterThan(0);
await session.close();
const result = await session.run("Say hello in one sentence.");
expect(result.finalText.length).toBeGreaterThan(0);
} finally {
await session.close();
rmSync(cwd, { recursive: true, force: true });
}
},
30000,
);
@@ -21,7 +21,7 @@ describe("CodexAppServerAgentClient spawn error handling", () => {
process.on("uncaughtException", onUncaught);
try {
await expect(client.listModels()).rejects.toThrow();
await expect(client.listModels({ cwd: "/tmp/codex-models", force: false })).rejects.toThrow();
// Drain microtask queue to ensure no deferred uncaught errors
await new Promise((resolve) => setTimeout(resolve, 100));
expect(uncaughtErrors).toHaveLength(0);
@@ -27,6 +27,7 @@ import type {
PersistedAgentDescriptor,
} from "../agent-sdk-types.js";
import type { Logger } from "pino";
import { homedir } from "node:os";
import type { ChildProcessWithoutNullStreams } from "node:child_process";
import { randomUUID } from "node:crypto";
@@ -4137,7 +4138,8 @@ export class CodexAppServerAgentClient implements AgentClient {
}
}
async listModels(_options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
async listModels(_options: ListModelsOptions): Promise<AgentModelDefinition[]> {
// 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);
@@ -4247,7 +4249,7 @@ export class CodexAppServerAgentClient implements AgentClient {
entries.push({ label: "Models", value: "Not checked" });
} else {
try {
const models = await this.listModels();
const models = await this.listModels({ cwd: homedir(), force: false });
entries.push({ label: "Models", value: String(models.length) });
} catch (error) {
entries.push({
@@ -1,4 +1,5 @@
import type { Logger } from "pino";
import { homedir } from "node:os";
import type { AgentCapabilityFlags, AgentMode } from "../agent-sdk-types.js";
import type { ProviderRuntimeSettings } from "../provider-launch-config.js";
@@ -69,7 +70,7 @@ export class CopilotACPAgentClient extends ACPAgentClient {
if (available) {
try {
const models = await this.listModels();
const models = await this.listModels({ cwd: homedir(), force: false });
modelsValue = String(models.length);
} catch (error) {
modelsValue = `Error - ${toDiagnosticErrorMessage(error)}`;
@@ -81,7 +82,7 @@ export class CopilotACPAgentClient extends ACPAgentClient {
if (!modelsValue.startsWith("Error -")) {
try {
await this.listModes();
await this.listModes({ cwd: homedir(), force: false });
} catch (error) {
status = formatDiagnosticStatus(available, {
source: "mode fetch",
@@ -19,7 +19,7 @@ describe("MockLoadTestAgentClient", () => {
test("default model is a five minute foreground stream", async () => {
const client = new MockLoadTestAgentClient();
const models = await client.listModels();
const models = await client.listModels({ cwd: "/tmp/mock-models", force: false });
expect(models[0]).toMatchObject({
id: MOCK_LOAD_TEST_DEFAULT_MODEL_ID,
@@ -20,6 +20,7 @@ import type {
AgentSessionConfig,
AgentStreamEvent,
AgentTimelineItem,
ListModesOptions,
ListModelsOptions,
ListPersistedAgentsOptions,
PersistedAgentDescriptor,
@@ -197,11 +198,11 @@ export class MockLoadTestAgentClient implements AgentClient {
});
}
async listModels(_options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
async listModels(_options: ListModelsOptions): Promise<AgentModelDefinition[]> {
return MODELS;
}
async listModes(): Promise<AgentMode[]> {
async listModes(_options: ListModesOptions): Promise<AgentMode[]> {
return getAgentProviderDefinition(MOCK_LOAD_TEST_PROVIDER_ID).modes;
}
@@ -50,11 +50,14 @@ describe("OpenCodeAgentClient.listModels timeout", () => {
} as never);
vi.spyOn(OpenCodeServerManager, "getInstance").mockReturnValue({
ensureRunning: vi.fn().mockResolvedValue({ port: 1234, url: "http://127.0.0.1:1234" }),
acquire: vi.fn().mockResolvedValue({
server: { port: 1234, url: "http://127.0.0.1:1234" },
release: vi.fn(),
}),
} as never);
const client = new OpenCodeAgentClient(createTestLogger());
const modelsPromise = client.listModels();
const modelsPromise = client.listModels({ cwd: "/tmp/opencode-models", force: false });
await vi.advanceTimersByTimeAsync(15_000);
@@ -66,4 +69,28 @@ describe("OpenCodeAgentClient.listModels timeout", () => {
},
]);
});
test("passes explicit refresh force through server acquisition", async () => {
vi.mocked(createOpencodeClient).mockReturnValue({
provider: {
list: vi.fn().mockResolvedValue({
data: {
connected: ["openai"],
all: [{ id: "openai", name: "OpenAI", models: {} }],
},
}),
},
} as never);
const acquire = vi.fn().mockResolvedValue({
server: { port: 1234, url: "http://127.0.0.1:1234" },
release: vi.fn(),
});
vi.spyOn(OpenCodeServerManager, "getInstance").mockReturnValue({ acquire } as never);
const client = new OpenCodeAgentClient(createTestLogger());
await client.listModels({ cwd: "/tmp/opencode-models", force: true });
expect(acquire).toHaveBeenCalledWith({ force: true });
});
});
@@ -66,7 +66,10 @@ describe("OpenCodeAgentSession slash command timeout handling", () => {
} as never);
vi.spyOn(OpenCodeServerManager, "getInstance").mockReturnValue({
ensureRunning: vi.fn().mockResolvedValue({ port: 1234, url: "http://127.0.0.1:1234" }),
acquire: vi.fn().mockResolvedValue({
server: { port: 1234, url: "http://127.0.0.1:1234" },
release: vi.fn(),
}),
} as never);
const client = new OpenCodeAgentClient(createTestLogger());
@@ -103,7 +103,7 @@ const hasOpenCode = isBinaryInstalled("opencode");
logger.info("beforeAll: Starting model selection");
const client = new OpenCodeAgentClient(logger);
const models = await client.listModels();
const models = await client.listModels({ cwd: os.homedir(), force: false });
logger.info(
{ modelCount: models.length, elapsed: Date.now() - startTime },
@@ -182,7 +182,7 @@ const hasOpenCode = isBinaryInstalled("opencode");
test("listModels returns models with required fields", async () => {
const client = new OpenCodeAgentClient(logger);
const models = await client.listModels();
const models = await client.listModels({ cwd: os.homedir(), force: false });
// HARD ASSERT: Returns an array
expect(Array.isArray(models)).toBe(true);
@@ -1,4 +1,5 @@
import type { ChildProcess } from "node:child_process";
import { homedir } from "node:os";
import {
createOpencodeClient,
type AssistantMessage as OpenCodeAssistantMessage,
@@ -45,6 +46,7 @@ import {
type ProviderRuntimeSettings,
} from "../provider-launch-config.js";
import { findExecutable, isCommandAvailable } from "../../../utils/executable.js";
import { withTimeout } from "../../../utils/promise-timeout.js";
import { spawnProcess } from "../../../utils/spawn.js";
import { mapOpencodeToolCall } from "./opencode/tool-call-mapper.js";
import {
@@ -700,9 +702,10 @@ export const __openCodeInternals = {
export class OpenCodeServerManager {
private static instance: OpenCodeServerManager | null = null;
private static exitHandlerRegistered = false;
private server: ChildProcess | null = null;
private port: number | null = null;
private startPromise: Promise<{ port: number; url: string }> | null = null;
private currentServer: OpenCodeServerGeneration | null = null;
private retiredServers = new Set<OpenCodeServerGeneration>();
private startPromise: Promise<OpenCodeServerGeneration> | null = null;
private forcedRefreshPromise: Promise<OpenCodeServerGeneration> | null = null;
private readonly logger: Logger;
private readonly runtimeSettings?: ProviderRuntimeSettings;
private readonly runtimeSettingsKey: string;
@@ -741,9 +744,7 @@ export class OpenCodeServerManager {
const cleanup = () => {
const instance = OpenCodeServerManager.instance;
if (instance?.server && !instance.server.killed) {
instance.server.kill("SIGTERM");
}
void instance?.shutdown();
};
process.on("exit", cleanup);
@@ -752,35 +753,99 @@ export class OpenCodeServerManager {
}
async ensureRunning(): Promise<{ port: number; url: string }> {
const acquisition = await this.acquire({ force: false });
acquisition.release();
return acquisition.server;
}
async acquire(options: { force: boolean }): Promise<{
server: { port: number; url: string };
release: () => void;
}> {
const server = options.force
? await this.getForcedRefreshServer()
: await this.getCurrentServer();
server.refCount += 1;
let released = false;
return {
server: { port: server.port, url: server.url },
release: () => {
if (released) {
return;
}
released = true;
server.refCount -= 1;
this.cleanupRetiredServers();
},
};
}
private async getForcedRefreshServer(): Promise<OpenCodeServerGeneration> {
if (this.forcedRefreshPromise) {
return this.forcedRefreshPromise;
}
this.forcedRefreshPromise = Promise.resolve()
.then(async () => {
await this.rotateCurrentServer();
return this.getCurrentServer();
})
.finally(() => {
this.forcedRefreshPromise = null;
});
return this.forcedRefreshPromise;
}
private async getCurrentServer(): Promise<OpenCodeServerGeneration> {
if (this.startPromise) {
return this.startPromise;
}
if (this.server && this.port && !this.server.killed) {
return { port: this.port, url: `http://127.0.0.1:${this.port}` };
if (this.currentServer && !this.currentServer.process.killed) {
return this.currentServer;
}
this.startPromise = this.startServer();
try {
const result = await this.startPromise;
if (!result.retired) {
this.currentServer = result;
}
return result;
} finally {
this.startPromise = null;
}
}
private async startServer(): Promise<{ port: number; url: string }> {
this.port = await findAvailablePort();
const url = `http://127.0.0.1:${this.port}`;
private async rotateCurrentServer(): Promise<void> {
const existing = this.currentServer;
if (existing) {
existing.retired = true;
this.retiredServers.add(existing);
this.currentServer = null;
this.cleanupRetiredServers();
}
if (this.startPromise) {
const pending = await this.startPromise;
pending.retired = true;
this.retiredServers.add(pending);
this.currentServer = null;
this.cleanupRetiredServers();
}
}
private async startServer(): Promise<OpenCodeServerGeneration> {
const port = await findAvailablePort();
const url = `http://127.0.0.1:${port}`;
const launchPrefix = await resolveProviderCommandPrefix(
this.runtimeSettings?.command,
resolveOpenCodeBinary,
);
return new Promise((resolve, reject) => {
this.server = spawnProcess(
const serverProcess = spawnProcess(
launchPrefix.command,
[...launchPrefix.args, "serve", "--port", String(this.port)],
[...launchPrefix.args, "serve", "--port", String(port)],
{
stdio: ["ignore", "pipe", "pipe"],
env: applyProviderEnv(process.env, this.runtimeSettings),
@@ -794,54 +859,94 @@ export class OpenCodeServerManager {
}
}, 30_000);
this.server.stdout?.on("data", (data: Buffer) => {
serverProcess.stdout?.on("data", (data: Buffer) => {
const output = data.toString();
if (output.includes("listening on") && !started) {
started = true;
clearTimeout(timeout);
resolve({ port: this.port!, url });
resolve({
process: serverProcess,
port,
url,
refCount: 0,
retired: false,
});
}
});
this.server.stderr?.on("data", (data: Buffer) => {
serverProcess.stderr?.on("data", (data: Buffer) => {
this.logger.error({ stderr: data.toString().trim() }, "OpenCode server stderr");
});
this.server.on("error", (error) => {
serverProcess.on("error", (error) => {
clearTimeout(timeout);
reject(error);
});
this.server.on("exit", (code) => {
serverProcess.on("exit", (code) => {
if (!started) {
clearTimeout(timeout);
reject(new Error(`OpenCode server exited with code ${code}`));
}
this.server = null;
this.port = null;
if (this.currentServer?.process === serverProcess) {
this.currentServer = null;
}
for (const retired of Array.from(this.retiredServers)) {
if (retired.process === serverProcess) {
this.retiredServers.delete(retired);
}
}
});
});
}
async shutdown(): Promise<void> {
if (this.server && !this.server.killed) {
this.server.kill("SIGTERM");
await new Promise<void>((resolve) => {
const timeout = setTimeout(() => {
this.server?.kill("SIGKILL");
resolve();
}, 5000);
this.server?.on("exit", () => {
clearTimeout(timeout);
resolve();
});
});
const servers = [
...(this.currentServer ? [this.currentServer] : []),
...Array.from(this.retiredServers),
];
for (const server of servers) {
await this.killServer(server);
}
this.server = null;
this.port = null;
this.currentServer = null;
this.retiredServers.clear();
}
private cleanupRetiredServers(): void {
for (const server of Array.from(this.retiredServers)) {
if (server.refCount === 0) {
this.retiredServers.delete(server);
void this.killServer(server);
}
}
}
private async killServer(server: OpenCodeServerGeneration): Promise<void> {
if (server.process.killed) {
return;
}
await new Promise<void>((resolve) => {
const timeout = setTimeout(() => {
server.process.kill("SIGKILL");
resolve();
}, 5000);
server.process.on("exit", () => {
clearTimeout(timeout);
resolve();
});
server.process.kill("SIGTERM");
});
}
}
type OpenCodeServerGeneration = {
process: ChildProcess;
port: number;
url: string;
refCount: number;
retired: boolean;
};
export class OpenCodeAgentClient implements AgentClient {
readonly provider: "opencode" = "opencode";
readonly capabilities = OPENCODE_CAPABILITIES;
@@ -862,40 +967,43 @@ export class OpenCodeAgentClient implements AgentClient {
_launchContext?: AgentLaunchContext,
): Promise<AgentSession> {
const openCodeConfig = this.assertConfig(config);
const { url } = await this.serverManager.ensureRunning();
const acquisition = await this.serverManager.acquire({ force: false });
const { url } = acquisition.server;
const client = createOpencodeClient({
baseUrl: url,
directory: openCodeConfig.cwd,
});
// Set a timeout for session creation to fail fast
const timeoutPromise = new Promise<never>((_, reject) => {
setTimeout(() => reject(new Error("OpenCode session.create timed out after 10s")), 10_000);
});
try {
const response = await withTimeout(
client.session.create({ directory: openCodeConfig.cwd }),
10_000,
"OpenCode session.create timed out after 10s",
);
const response = await Promise.race([
client.session.create({ directory: openCodeConfig.cwd }),
timeoutPromise,
]);
if (response.error) {
throw new Error(`Failed to create OpenCode session: ${JSON.stringify(response.error)}`);
}
if (response.error) {
throw new Error(`Failed to create OpenCode session: ${JSON.stringify(response.error)}`);
const session = response.data;
if (!session) {
throw new Error("OpenCode session creation returned no data");
}
await this.populateModelContextWindowCache(client, openCodeConfig.cwd);
return new OpenCodeAgentSession(
openCodeConfig,
client,
session.id,
this.logger,
new Map(this.modelContextWindows),
acquisition.release,
);
} catch (error) {
acquisition.release();
throw error;
}
const session = response.data;
if (!session) {
throw new Error("OpenCode session creation returned no data");
}
await this.populateModelContextWindowCache(client, openCodeConfig.cwd);
return new OpenCodeAgentSession(
openCodeConfig,
client,
session.id,
this.logger,
new Map(this.modelContextWindows),
);
}
async resumeSession(
@@ -914,119 +1022,125 @@ export class OpenCodeAgentClient implements AgentClient {
...overrides,
};
const openCodeConfig = this.assertConfig(config);
const { url } = await this.serverManager.ensureRunning();
const acquisition = await this.serverManager.acquire({ force: false });
const { url } = acquisition.server;
const client = createOpencodeClient({
baseUrl: url,
directory: openCodeConfig.cwd,
});
await this.populateModelContextWindowCache(client, openCodeConfig.cwd);
try {
await this.populateModelContextWindowCache(client, openCodeConfig.cwd);
return new OpenCodeAgentSession(
openCodeConfig,
client,
handle.sessionId,
this.logger,
new Map(this.modelContextWindows),
);
return new OpenCodeAgentSession(
openCodeConfig,
client,
handle.sessionId,
this.logger,
new Map(this.modelContextWindows),
acquisition.release,
);
} catch (error) {
acquisition.release();
throw error;
}
}
async listModels(options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
const { url } = await this.serverManager.ensureRunning();
async listModels(options: ListModelsOptions): Promise<AgentModelDefinition[]> {
const acquisition = await this.serverManager.acquire({ force: options.force });
const { url } = acquisition.server;
const client = createOpencodeClient({
baseUrl: url,
directory: options?.cwd ?? process.cwd(),
directory: options.cwd,
});
// Background model discovery can be legitimately slow while OpenCode refreshes
// provider state, so allow longer than turn execution paths.
const timeoutPromise = new Promise<never>((_, reject) => {
setTimeout(
() =>
reject(
new Error(
`OpenCode provider.list timed out after ${OPENCODE_PROVIDER_LIST_TIMEOUT_MS / 1000}s - server may not be authenticated or connected to any providers`,
),
),
try {
// Background model discovery can be legitimately slow while OpenCode refreshes
// provider state, so allow longer than turn execution paths.
const response = await withTimeout(
client.provider.list({ directory: options.cwd }),
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 Promise.race([
client.provider.list({ directory: options?.cwd ?? process.cwd() }),
timeoutPromise,
]);
if (response.error) {
throw new Error(`Failed to fetch OpenCode providers: ${JSON.stringify(response.error)}`);
}
const providers = response.data;
if (!providers) {
return [];
}
// Only include models from connected providers (ones that are actually available)
const connectedProviderIds = new Set(providers.connected);
// Fail fast if no providers are connected
if (connectedProviderIds.size === 0) {
throw new Error(
"OpenCode has no connected providers. Please authenticate with at least one provider (e.g., openai, anthropic) or set appropriate environment variables (e.g., OPENAI_API_KEY).",
);
}
const models: AgentModelDefinition[] = [];
this.modelContextWindows.clear();
for (const provider of providers.all) {
// Skip providers that aren't connected/configured
if (!connectedProviderIds.has(provider.id)) {
continue;
if (response.error) {
throw new Error(`Failed to fetch OpenCode providers: ${JSON.stringify(response.error)}`);
}
for (const [modelId, model] of Object.entries(provider.models)) {
const definition = buildOpenCodeModelDefinition(provider, modelId, model);
const contextWindowMaxTokens = extractOpenCodeModelContextWindow(model);
if (contextWindowMaxTokens !== undefined) {
this.modelContextWindows.set(
buildOpenCodeModelLookupKey(provider.id, modelId),
contextWindowMaxTokens,
);
const providers = response.data;
if (!providers) {
return [];
}
// Only include models from connected providers (ones that are actually available)
const connectedProviderIds = new Set(providers.connected);
// Fail fast if no providers are connected
if (connectedProviderIds.size === 0) {
throw new Error(
"OpenCode has no connected providers. Please authenticate with at least one provider (e.g., openai, anthropic) or set appropriate environment variables (e.g., OPENAI_API_KEY).",
);
}
const models: AgentModelDefinition[] = [];
this.modelContextWindows.clear();
for (const provider of providers.all) {
// Skip providers that aren't connected/configured
if (!connectedProviderIds.has(provider.id)) {
continue;
}
models.push(definition);
}
}
return models;
for (const [modelId, model] of Object.entries(provider.models)) {
const definition = buildOpenCodeModelDefinition(provider, modelId, model);
const contextWindowMaxTokens = extractOpenCodeModelContextWindow(model);
if (contextWindowMaxTokens !== undefined) {
this.modelContextWindows.set(
buildOpenCodeModelLookupKey(provider.id, modelId),
contextWindowMaxTokens,
);
}
models.push(definition);
}
}
return models;
} finally {
acquisition.release();
}
}
async listModes(options?: ListModesOptions): Promise<AgentMode[]> {
const { url } = await this.serverManager.ensureRunning();
const directory = options?.cwd ?? process.cwd();
async listModes(options: ListModesOptions): Promise<AgentMode[]> {
const acquisition = await this.serverManager.acquire({ force: options.force });
const { url } = acquisition.server;
const directory = options.cwd;
const client = createOpencodeClient({ baseUrl: url, directory });
const timeoutPromise = new Promise<never>((_, reject) => {
setTimeout(() => reject(new Error("OpenCode app.agents timed out after 10s")), 10_000);
});
try {
const response = await withTimeout(
client.app.agents({ directory }),
10_000,
"OpenCode app.agents timed out after 10s",
);
const response = await Promise.race([client.app.agents({ directory }), timeoutPromise]);
if (response.error || !response.data) {
return DEFAULT_MODES;
}
if (response.error || !response.data) {
return DEFAULT_MODES;
const discovered = response.data
.filter((agent) => agent.mode === "primary" && agent.hidden !== true)
.map((agent) => ({
id: agent.name,
label: agent.name.charAt(0).toUpperCase() + agent.name.slice(1),
description:
typeof agent.description === "string" && agent.description.trim().length > 0
? agent.description.trim()
: DEFAULT_MODES.find((mode) => mode.id === agent.name)?.description,
}));
return discovered.length > 0 ? sortOpenCodeModes(discovered) : DEFAULT_MODES;
} finally {
acquisition.release();
}
const discovered = response.data
.filter((agent) => agent.mode === "primary" && agent.hidden !== true)
.map((agent) => ({
id: agent.name,
label: agent.name.charAt(0).toUpperCase() + agent.name.slice(1),
description:
typeof agent.description === "string" && agent.description.trim().length > 0
? agent.description.trim()
: DEFAULT_MODES.find((mode) => mode.id === agent.name)?.description,
}));
return discovered.length > 0 ? sortOpenCodeModes(discovered) : DEFAULT_MODES;
}
async listPersistedAgents(
@@ -1061,7 +1175,7 @@ export class OpenCodeAgentClient implements AgentClient {
if (available) {
try {
const models = await this.listModels();
const models = await this.listModels({ cwd: homedir(), force: false });
modelsValue = String(models.length);
} catch (error) {
modelsValue = `Error - ${toDiagnosticErrorMessage(error)}`;
@@ -1073,7 +1187,7 @@ export class OpenCodeAgentClient implements AgentClient {
if (!modelsValue.startsWith("Error -")) {
try {
await this.listModes();
await this.listModes({ cwd: homedir(), force: false });
} catch (error) {
status = formatDiagnosticStatus(available, {
source: "mode fetch",
@@ -1671,12 +1785,14 @@ class OpenCodeAgentSession implements AgentSession {
private activeForegroundTurnId: string | null = null;
private readonly runningToolCalls = new Map<string, ToolCallTimelineItem>();
private selectedModelContextWindowMaxTokens: number | undefined;
private releaseServer: (() => void) | null;
constructor(
config: OpenCodeAgentConfig,
client: OpencodeClient,
sessionId: string,
logger: Logger,
modelContextWindowsByModelKey: ReadonlyMap<string, number> = new Map(),
releaseServer?: () => void,
) {
this.config = config;
this.client = client;
@@ -1684,6 +1800,7 @@ class OpenCodeAgentSession implements AgentSession {
this.logger = logger;
this.modelContextWindowsByModelKey = modelContextWindowsByModelKey;
this.currentMode = normalizeOpenCodeModeId(config.modeId);
this.releaseServer = releaseServer ?? null;
this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens(
config.model,
);
@@ -2278,15 +2395,20 @@ class OpenCodeAgentSession implements AgentSession {
}
async close(): Promise<void> {
this.abortController?.abort();
await reconcileOpenCodeSessionClose({
client: this.client,
sessionId: this.sessionId,
directory: this.config.cwd,
logger: this.logger,
});
this.subscribers.clear();
this.activeForegroundTurnId = null;
try {
this.abortController?.abort();
await reconcileOpenCodeSessionClose({
client: this.client,
sessionId: this.sessionId,
directory: this.config.cwd,
logger: this.logger,
});
this.subscribers.clear();
this.activeForegroundTurnId = null;
} finally {
this.releaseServer?.();
this.releaseServer = null;
}
}
private parseSlashCommandInput(text: string): { commandName: string; args?: string } | null {
@@ -0,0 +1,174 @@
import { EventEmitter } from "node:events";
import { describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { OpenCodeServerManager } from "./opencode-agent.js";
type FakeServerProcess = EventEmitter & {
killed: boolean;
kill: ReturnType<typeof vi.fn>;
};
type FakeGeneration = {
process: FakeServerProcess;
port: number;
url: string;
refCount: number;
retired: boolean;
};
describe("OpenCodeServerManager generations", () => {
test("rotation creates a new current server without killing a referenced old server", async () => {
const manager = createTestManager();
const first = createGeneration(4101);
const second = createGeneration(4102);
stubGenerations(manager, [first, second]);
const oldAcquisition = await manager.acquire({ force: false });
const newAcquisition = await manager.acquire({ force: true });
expect(oldAcquisition.server.url).toBe("http://127.0.0.1:4101");
expect(newAcquisition.server.url).toBe("http://127.0.0.1:4102");
expect(first.process.kill).not.toHaveBeenCalled();
expect(second.process.kill).not.toHaveBeenCalled();
newAcquisition.release();
oldAcquisition.release();
expect(first.process.kill).toHaveBeenCalledWith("SIGTERM");
});
test("new acquisitions after rotation use the new server", async () => {
const manager = createTestManager();
const first = createGeneration(4201);
const second = createGeneration(4202);
stubGenerations(manager, [first, second]);
const oldAcquisition = await manager.acquire({ force: false });
const rotatedAcquisition = await manager.acquire({ force: true });
rotatedAcquisition.release();
const nextAcquisition = await manager.acquire({ force: false });
expect(nextAcquisition.server.url).toBe("http://127.0.0.1:4202");
expect(first.process.kill).not.toHaveBeenCalled();
nextAcquisition.release();
oldAcquisition.release();
});
test("concurrent forced acquisitions share one fresh generation", async () => {
const manager = createTestManager();
const first = createGeneration(4251);
const second = createGeneration(4252);
const third = createGeneration(4253);
const startServer = stubGenerations(manager, [first, second, third]);
const initialAcquisition = await manager.acquire({ force: false });
initialAcquisition.release();
const [modelsAcquisition, modesAcquisition] = await Promise.all([
manager.acquire({ force: true }),
manager.acquire({ force: true }),
]);
expect(modelsAcquisition.server.url).toBe("http://127.0.0.1:4252");
expect(modesAcquisition.server.url).toBe("http://127.0.0.1:4252");
expect(startServer).toHaveBeenCalledTimes(2);
modesAcquisition.release();
modelsAcquisition.release();
});
test("release is idempotent", async () => {
const manager = createTestManager();
const first = createGeneration(4301);
const second = createGeneration(4302);
stubGenerations(manager, [first, second]);
const oldAcquisition = await manager.acquire({ force: false });
const newAcquisition = await manager.acquire({ force: true });
newAcquisition.release();
oldAcquisition.release();
oldAcquisition.release();
expect(first.refCount).toBe(0);
expect(first.process.kill).toHaveBeenCalledTimes(1);
});
test("shutdown kills current and retired servers", async () => {
const manager = createTestManager();
const first = createGeneration(4401);
const second = createGeneration(4402);
stubGenerations(manager, [first, second]);
await manager.acquire({ force: false });
await manager.acquire({ force: true });
await manager.shutdown();
expect(first.process.kill).toHaveBeenCalledWith("SIGTERM");
expect(second.process.kill).toHaveBeenCalledWith("SIGTERM");
});
test("repeated rotations leave zero unreferenced retired servers", async () => {
const manager = createTestManager();
const first = createGeneration(4501);
const second = createGeneration(4502);
const third = createGeneration(4503);
stubGenerations(manager, [first, second, third]);
const firstAcquisition = await manager.acquire({ force: false });
const secondAcquisition = await manager.acquire({ force: true });
secondAcquisition.release();
const thirdAcquisition = await manager.acquire({ force: true });
thirdAcquisition.release();
firstAcquisition.release();
const retiredServers = (manager as unknown as { retiredServers: Set<FakeGeneration> })
.retiredServers;
expect(Array.from(retiredServers).filter((server) => server.refCount === 0)).toHaveLength(0);
expect(first.process.kill).toHaveBeenCalledTimes(1);
expect(second.process.kill).toHaveBeenCalledTimes(1);
});
});
function createTestManager(): OpenCodeServerManager {
const ManagerConstructor = OpenCodeServerManager as unknown as {
new (logger: ReturnType<typeof createTestLogger>): OpenCodeServerManager;
};
return new ManagerConstructor(createTestLogger());
}
function stubGenerations(
manager: OpenCodeServerManager,
generations: FakeGeneration[],
): ReturnType<typeof vi.fn> {
const startServer = vi.fn(async () => {
const generation = generations.shift();
if (!generation) {
throw new Error("No fake OpenCode server generation available");
}
return generation;
});
(manager as unknown as { startServer: typeof startServer }).startServer = startServer;
return startServer;
}
function createGeneration(port: number): FakeGeneration {
const process = new EventEmitter() as FakeServerProcess;
process.killed = false;
process.kill = vi.fn((signal?: NodeJS.Signals) => {
process.killed = true;
process.emit("exit", signal ?? "SIGTERM");
return true;
});
return {
process,
port,
url: `http://127.0.0.1:${port}`,
refCount: 0,
retired: false,
};
}
@@ -45,6 +45,7 @@ import type {
AgentStreamEvent,
AgentTimelineItem,
AgentUsage,
ListModesOptions,
ListModelsOptions,
ToolCallDetail,
} from "../agent-sdk-types.js";
@@ -1448,7 +1449,8 @@ export class PiDirectAgentClient implements AgentClient {
return new PiDirectAgentSession(session, this.getModelRegistry(), mergedConfig);
}
async listModels(_options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
async listModels(_options: ListModelsOptions): Promise<AgentModelDefinition[]> {
// Pi Direct uses an in-process global registry; cwd/force are intentionally irrelevant.
const models = this.getModelRegistry()
.getAll()
.map((model) => ({
@@ -1467,7 +1469,7 @@ export class PiDirectAgentClient implements AgentClient {
return transformPiModels(models);
}
async listModes(): Promise<AgentMode[]> {
async listModes(_options: ListModesOptions): Promise<AgentMode[]> {
return [];
}
@@ -1504,7 +1506,7 @@ export class PiDirectAgentClient implements AgentClient {
if (available) {
try {
const models = await this.listModels();
const models = await this.listModels({ cwd: homedir(), force: false });
modelsValue = String(models.length);
} catch (error) {
modelsValue = `Error - ${toDiagnosticErrorMessage(error)}`;
@@ -13,6 +13,7 @@ import { tmpdir } from "os";
import path from "path";
import { createDaemonTestContext, type DaemonTestContext } from "../test-utils/index.js";
import { createMessageCollector, type MessageCollector } from "../test-utils/message-collector.js";
import { withTimeout } from "../../utils/promise-timeout.js";
import { deriveWorktreeProjectHash } from "../../utils/worktree.js";
import type { AgentTimelineItem } from "../agent/agent-sdk-types.js";
import type { AgentSnapshotPayload, SessionOutboundMessage } from "../messages.js";
@@ -21,26 +22,6 @@ function tmpCwd(): string {
return mkdtempSync(path.join(tmpdir(), "daemon-e2e-"));
}
async function withTimeout<T>(options: {
promise: Promise<T>;
timeoutMs: number;
label: string;
}): Promise<T> {
let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
const timeout = new Promise<never>((_, reject) => {
timeoutHandle = setTimeout(() => {
reject(new Error(`Timed out after ${options.timeoutMs}ms (${options.label})`));
}, options.timeoutMs);
});
try {
return await Promise.race([options.promise, timeout]);
} finally {
if (timeoutHandle) {
clearTimeout(timeoutHandle);
}
}
}
function findTimelineToolCall(
messages: SessionOutboundMessage[],
agentId: string,
@@ -451,15 +451,20 @@ describe(
"PiDirectAgentClient.listModels returns non-empty Pi model definitions",
async () => {
const client = createPiClient();
const models = await client.listModels();
const cwd = tmpCwd("pi-list-models-");
try {
const models = await client.listModels({ cwd, force: false });
expect(models.length).toBeGreaterThan(0);
for (const model of models) {
expect(model.provider).toBe("pi");
expect(typeof model.id).toBe("string");
expect(model.id.length).toBeGreaterThan(0);
expect(typeof model.label).toBe("string");
expect(model.label.length).toBeGreaterThan(0);
expect(models.length).toBeGreaterThan(0);
for (const model of models) {
expect(model.provider).toBe("pi");
expect(typeof model.id).toBe("string");
expect(model.id.length).toBeGreaterThan(0);
expect(typeof model.label).toBe("string");
expect(model.label.length).toBeGreaterThan(0);
}
} finally {
rmSync(cwd, { recursive: true, force: true });
}
},
PI_TEST_TIMEOUT_MS,
+204 -1
View File
@@ -1,12 +1,21 @@
import { execSync } from "child_process";
import { mkdtempSync, rmSync, writeFileSync } from "fs";
import { tmpdir } from "os";
import { homedir, tmpdir } from "os";
import { join } from "path";
import pino from "pino";
import { afterEach, describe, expect, test, vi } from "vitest";
import { CheckoutPrStatusSchema } from "../shared/messages.js";
import { normalizeCheckoutPrStatusPayload, Session } from "./session.js";
import type {
AgentClient,
AgentMode,
AgentModelDefinition,
ListModesOptions,
ListModelsOptions,
} from "./agent/agent-sdk-types.js";
import type { ProviderDefinition } from "./agent/provider-registry.js";
import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
const checkoutGitMocks = vi.hoisted(() => ({
checkoutResolvedBranch: vi.fn(),
@@ -36,6 +45,31 @@ const paseoWorktreeServiceMocks = vi.hoisted(() => ({
createPaseoWorktree: vi.fn(),
}));
type Deferred<T> = {
promise: Promise<T>;
resolve: (value: T) => void;
reject: (reason?: unknown) => void;
};
const TEST_CAPABILITIES = {
supportsStreaming: false,
supportsSessionPersistence: false,
supportsDynamicModes: false,
supportsMcpServers: false,
supportsReasoningStream: false,
supportsToolInvocations: false,
} as const;
function deferred<T>(): Deferred<T> {
let resolve!: (value: T) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
return { promise, resolve, reject };
}
vi.mock("../utils/checkout-git.js", async (importOriginal) => {
const actual = await importOriginal<typeof import("../utils/checkout-git.js")>();
return {
@@ -113,6 +147,7 @@ function createSessionForTest(options?: {
scriptRuntimeStore?: unknown;
getDaemonTcpPort?: () => number | null;
getDaemonTcpHost?: () => string | null;
providerSnapshotManager?: ProviderSnapshotManager;
messages?: unknown[];
}): Session {
const logger = pino({ level: "silent" });
@@ -165,6 +200,7 @@ function createSessionForTest(options?: {
stt: null,
tts: null,
terminalManager: (options?.terminalManager ?? null) as any,
providerSnapshotManager: options?.providerSnapshotManager,
scriptRouteStore: options?.scriptRouteStore as any,
scriptRuntimeStore: options?.scriptRuntimeStore as any,
getDaemonTcpPort: options?.getDaemonTcpPort,
@@ -207,6 +243,20 @@ function createWorkspaceGitSnapshot(
};
}
function createProviderSnapshotManagerStub(): ProviderSnapshotManager {
const stub = {
getSnapshot: vi.fn(() => []),
refreshSnapshotForCwd: vi.fn(async () => {}),
refreshSettingsSnapshot: vi.fn(async () => {}),
warmUpSnapshotForCwd: vi.fn(async () => {}),
on: vi.fn(),
off: vi.fn(),
};
stub.on.mockImplementation(() => stub);
stub.off.mockImplementation(() => stub);
return stub as unknown as ProviderSnapshotManager;
}
afterEach(() => {
vi.clearAllMocks();
});
@@ -243,6 +293,159 @@ describe("session PR status payload normalization", () => {
});
});
describe("session provider refresh cwd routing", () => {
test("routes no-cwd provider snapshot refreshes through settings refresh", async () => {
const providerSnapshotManager = createProviderSnapshotManagerStub();
const session = createSessionForTest({ providerSnapshotManager });
await session.handleMessage({
type: "refresh_providers_snapshot_request",
providers: ["codex"],
requestId: "refresh-settings",
});
expect(providerSnapshotManager.refreshSettingsSnapshot).toHaveBeenCalledWith({
providers: ["codex"],
});
expect(providerSnapshotManager.refreshSnapshotForCwd).not.toHaveBeenCalled();
});
test("routes cwd provider snapshot refreshes through workspace refresh", async () => {
const providerSnapshotManager = createProviderSnapshotManagerStub();
const session = createSessionForTest({ providerSnapshotManager });
await session.handleMessage({
type: "refresh_providers_snapshot_request",
cwd: "/tmp/workspace-refresh",
providers: ["codex"],
requestId: "refresh-workspace",
});
expect(providerSnapshotManager.refreshSnapshotForCwd).toHaveBeenCalledWith({
cwd: "/tmp/workspace-refresh",
providers: ["codex"],
});
expect(providerSnapshotManager.refreshSettingsSnapshot).not.toHaveBeenCalled();
});
test("normalizes legacy model and mode list requests without cwd to home", async () => {
const messages: unknown[] = [];
const session = createSessionForTest({ messages });
const fetchModels = vi.fn(async () => []);
const fetchModes = vi.fn(async () => []);
(session as unknown as { providerRegistry: unknown }).providerRegistry = {
codex: {
fetchModels,
fetchModes,
},
};
await session.handleMessage({
type: "list_provider_models_request",
provider: "codex",
requestId: "models-home",
});
await session.handleMessage({
type: "list_provider_modes_request",
provider: "codex",
requestId: "modes-home",
});
expect(fetchModels).toHaveBeenCalledWith({ cwd: homedir(), force: false });
expect(fetchModes).toHaveBeenCalledWith({ cwd: homedir(), force: false });
});
test("legacy model list request without cwd awaits loading snapshot without forced discovery", async () => {
const messages: unknown[] = [];
const models = deferred<AgentModelDefinition[]>();
const fetchModels = vi.fn(
async (options: ListModelsOptions): Promise<AgentModelDefinition[]> => {
expect(options.cwd).toBe(homedir());
return models.promise;
},
);
const fetchModes = vi.fn(async (_options: ListModesOptions): Promise<AgentMode[]> => []);
const providerDefinition: ProviderDefinition = {
id: "codex",
label: "Codex",
description: "Codex test provider",
defaultModeId: null,
modes: [],
createClient: () =>
({
provider: "codex",
capabilities: TEST_CAPABILITIES,
async createSession() {
throw new Error("not implemented");
},
async resumeSession() {
throw new Error("not implemented");
},
async listModels(options: ListModelsOptions) {
return fetchModels(options);
},
async isAvailable() {
return true;
},
}) satisfies AgentClient,
fetchModels,
fetchModes,
};
const providerSnapshotManager = new ProviderSnapshotManager(
{ codex: providerDefinition },
pino({ level: "silent" }),
);
const session = createSessionForTest({ messages, providerSnapshotManager });
providerSnapshotManager.getSnapshot();
await vi.waitFor(() => {
expect(fetchModels).toHaveBeenCalledTimes(1);
});
const responsePromise = session.handleMessage({
type: "list_provider_models_request",
provider: "codex",
requestId: "models-loading-home",
});
await Promise.resolve();
expect(fetchModels).toHaveBeenCalledTimes(1);
expect(fetchModels).toHaveBeenCalledWith({ cwd: homedir(), force: false });
expect(fetchModels).not.toHaveBeenCalledWith({ cwd: homedir(), force: true });
models.resolve([
{
provider: "codex",
id: "gpt-5.4",
label: "GPT-5.4",
},
]);
await responsePromise;
expect(fetchModels).toHaveBeenCalledTimes(1);
expect(fetchModels).not.toHaveBeenCalledWith({ cwd: homedir(), force: true });
expect(messages).toContainEqual({
type: "list_provider_models_response",
payload: {
provider: "codex",
models: [
{
provider: "codex",
id: "gpt-5.4",
label: "GPT-5.4",
},
],
error: null,
fetchedAt: expect.any(String),
requestId: "models-loading-home",
},
});
providerSnapshotManager.destroy();
});
});
describe("session checkout merge handling", () => {
test("uses workspace git service snapshot for merge-to-base preflight", async () => {
const messages: unknown[] = [];
+92 -19
View File
@@ -86,7 +86,7 @@ import type {
ProviderOverride,
} from "./agent/provider-launch-config.js";
import { AgentManager } from "./agent/agent-manager.js";
import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
import { ProviderSnapshotManager, resolveSnapshotCwd } from "./agent/provider-snapshot-manager.js";
import type {
AgentTimelineCursor,
AgentTimelineFetchDirection,
@@ -805,7 +805,7 @@ export class Session {
);
}
if (this.providerSnapshotManager) {
const handleProviderSnapshotChange = (entries: ProviderSnapshotEntry[], cwd?: string) => {
const handleProviderSnapshotChange = (entries: ProviderSnapshotEntry[], cwd: string) => {
// COMPAT(providersSnapshot): keep provider visibility gating for older clients.
const visibleEntries = entries.filter((entry) =>
this.isProviderVisibleToClient(entry.provider),
@@ -3071,13 +3071,16 @@ export class Session {
private async handleListProviderModelsRequest(
msg: Extract<SessionInboundMessage, { type: "list_provider_models_request" }>,
): Promise<void> {
const cwd = msg.cwd ? expandTilde(msg.cwd) : undefined;
const cwd = resolveSnapshotCwd(msg.cwd ? expandTilde(msg.cwd) : undefined);
const fetchedAt = new Date().toISOString();
const manager = this.providerSnapshotManager;
if (!manager) {
try {
const models = await this.providerRegistry[msg.provider].fetchModels({ cwd });
const models = await this.providerRegistry[msg.provider].fetchModels({
cwd,
force: false,
});
this.emit({
type: "list_provider_models_response",
payload: {
@@ -3106,16 +3109,7 @@ export class Session {
return;
}
const findEntry = () =>
manager.getSnapshot(cwd).find((candidate) => candidate.provider === msg.provider);
let entry = findEntry();
if (!entry || entry.status === "loading") {
// Awaits the in-flight warmup (deduped per-cwd) so old clients still get
// a resolved answer rather than a loading placeholder.
await manager.refresh({ cwd, providers: [msg.provider] });
entry = findEntry();
}
const entry = await this.getProviderSnapshotEntryForRead(cwd, msg.provider);
if (!entry) {
this.emit({
@@ -3164,9 +3158,60 @@ export class Session {
msg: Extract<SessionInboundMessage, { type: "list_provider_modes_request" }>,
): Promise<void> {
const fetchedAt = new Date().toISOString();
const cwd = resolveSnapshotCwd(msg.cwd ? expandTilde(msg.cwd) : undefined);
const manager = this.providerSnapshotManager;
if (manager) {
const entry = await this.getProviderSnapshotEntryForRead(cwd, msg.provider);
if (!entry) {
this.emit({
type: "list_provider_modes_response",
payload: {
provider: msg.provider,
error: `Unknown provider: ${msg.provider}`,
fetchedAt,
requestId: msg.requestId,
},
});
return;
}
if (entry.status === "ready") {
this.emit({
type: "list_provider_modes_response",
payload: {
provider: msg.provider,
modes: entry.modes ?? [],
error: null,
fetchedAt: entry.fetchedAt ?? fetchedAt,
requestId: msg.requestId,
},
});
return;
}
const errorMessage =
entry.status === "error"
? (entry.error ?? `Failed to list modes for ${msg.provider}`)
: `Provider ${msg.provider} is not available`;
this.emit({
type: "list_provider_modes_response",
payload: {
provider: msg.provider,
error: errorMessage,
fetchedAt,
requestId: msg.requestId,
},
});
return;
}
try {
const modes = await this.providerRegistry[msg.provider].fetchModes({
cwd: msg.cwd ? expandTilde(msg.cwd) : undefined,
cwd,
force: false,
});
this.emit({
type: "list_provider_modes_response",
@@ -3195,6 +3240,28 @@ export class Session {
}
}
private async getProviderSnapshotEntryForRead(
cwd: string,
provider: AgentProvider,
): Promise<ProviderSnapshotEntry | undefined> {
const manager = this.providerSnapshotManager;
if (!manager) {
return undefined;
}
const findEntry = () =>
manager.getSnapshot(cwd).find((candidate) => candidate.provider === provider);
let entry = findEntry();
if (!entry || entry.status === "loading") {
// Awaits the in-flight warmup (deduped per-cwd) so old clients still get
// a resolved answer rather than a loading placeholder.
await manager.warmUpSnapshotForCwd({ cwd, providers: [provider] });
entry = findEntry();
}
return entry;
}
private buildDraftAgentSessionConfig(draftConfig: {
provider: AgentProvider;
cwd: string;
@@ -3301,10 +3368,16 @@ export class Session {
private async handleRefreshProvidersSnapshotRequest(
msg: Extract<SessionInboundMessage, { type: "refresh_providers_snapshot_request" }>,
): Promise<void> {
await this.providerSnapshotManager?.refresh({
cwd: msg.cwd ? expandTilde(msg.cwd) : undefined,
providers: msg.providers,
});
if (msg.cwd) {
await this.providerSnapshotManager?.refreshSnapshotForCwd({
cwd: expandTilde(msg.cwd),
providers: msg.providers,
});
} else {
await this.providerSnapshotManager?.refreshSettingsSnapshot({
providers: msg.providers,
});
}
this.emit({
type: "refresh_providers_snapshot_response",
payload: {
@@ -906,7 +906,7 @@ class FakeAgentClient implements AgentClient {
);
}
async listModels(_options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
async listModels(_options: ListModelsOptions): Promise<AgentModelDefinition[]> {
if (this.provider === "claude") {
return [
{ provider: this.provider, id: "haiku", label: "Haiku", isDefault: true },
@@ -9,6 +9,7 @@ import { createDaemonTestContext, type DaemonTestContext } from "./test-utils/in
import { OpenAITTS } from "./speech/providers/openai/tts.js";
import { OpenAISTT } from "./speech/providers/openai/stt.js";
import { STTManager } from "./agent/stt-manager.js";
import { withTimeout } from "../utils/promise-timeout.js";
const openaiApiKey = process.env.OPENAI_API_KEY ?? null;
const shouldRun = process.env.PASEO_VOICE_ROUNDTRIP_E2E === "1" && Boolean(openaiApiKey);
@@ -71,24 +72,6 @@ function waitForSignal<T>(
});
}
async function withTimeout<T>(label: string, timeoutMs: number, task: Promise<T>): Promise<T> {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => {
reject(new Error(`Timed out during ${label} after ${timeoutMs}ms`));
}, timeoutMs);
task.then(
(value) => {
clearTimeout(timer);
resolve(value);
},
(error) => {
clearTimeout(timer);
reject(error);
},
);
});
}
async function streamToBuffer(stream: Readable): Promise<Buffer> {
const chunks: Buffer[] = [];
for await (const chunk of stream) {
@@ -148,20 +131,20 @@ describe("voice roundtrip e2e", () => {
path.join(tmpdir(), `voice-roundtrip-agent-${targetProvider}-`),
);
const voiceAgent = await withTimeout(
"createVoiceTargetAgent",
30000,
ctx.client.createAgent({
config: {
...getVoiceRoundtripConfig(targetProvider),
cwd: voiceCwd,
},
}),
30000,
"Timed out during createVoiceTargetAgent after 30000ms",
);
const voiceAgentId = voiceAgent.id;
const voiceMode = await withTimeout(
"setVoiceMode",
15000,
ctx.client.setVoiceMode(true, voiceAgentId),
15000,
"Timed out during setVoiceMode after 15000ms",
);
expect(voiceMode.accepted).toBe(true);
expect(voiceMode.enabled).toBe(true);
@@ -194,14 +177,14 @@ describe("voice roundtrip e2e", () => {
});
const inputSpeech = await withTimeout(
"synthesizeInputAudio",
30000,
ttsProvider.synthesizeSpeech("Use the speak tool and say exactly round trip successful."),
30000,
"Timed out during synthesizeInputAudio after 30000ms",
);
const inputPcm = await withTimeout(
"collectInputAudio",
15000,
streamToBuffer(inputSpeech.stream),
15000,
"Timed out during collectInputAudio after 15000ms",
);
const outputAudio = await (async () => {
try {
@@ -281,16 +264,24 @@ describe("voice roundtrip e2e", () => {
);
const isLast = offset + chunkBytes >= inputPcm.length;
await withTimeout(
"sendVoiceAudioChunk",
5000,
ctx.client.sendVoiceAudioChunk(chunk.toString("base64"), format, isLast),
5000,
"Timed out during sendVoiceAudioChunk after 5000ms",
);
}
const transcript = await withTimeout("waitForTranscription", 35000, transcriptPromise);
const transcript = await withTimeout(
transcriptPromise,
35000,
"Timed out during waitForTranscription after 35000ms",
);
if (transcript.text.trim().length === 0) {
throw new Error(`empty transcription (lowConfidence=${transcript.isLowConfidence})`);
}
return await withTimeout("waitForAudioOutput", 95000, outputAudioPromise);
return await withTimeout(
outputAudioPromise,
95000,
"Timed out during waitForAudioOutput after 95000ms",
);
} catch (error) {
throw new Error(
`${error instanceof Error ? error.message : String(error)} | requestedVoiceAgentId=${voiceAgentId} | timelineTools=${JSON.stringify(timelineTools)} | timelineToolAgentIds=${JSON.stringify(Array.from(timelineToolAgentIds))} | activityErrors=${JSON.stringify(activityErrors)}`,
@@ -311,11 +302,11 @@ describe("voice roundtrip e2e", () => {
? "audio/wav"
: `audio/${outputAudio.format}`;
const transcription = await withTimeout(
"transcribeOutputAudio",
60000,
sttOutput.transcribe(outputRaw, outputFormat, {
label: "voice-roundtrip-output",
}),
60000,
"Timed out during transcribeOutputAudio after 60000ms",
);
const normalized = transcription.text.trim().toLowerCase();
@@ -0,0 +1,45 @@
type TimeoutOptions<T> = {
promise: Promise<T>;
timeoutMs: number;
label: string;
};
export function withTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T>;
export function withTimeout<T>(options: TimeoutOptions<T>): Promise<T>;
export function withTimeout<T>(
promiseOrOptions: Promise<T> | TimeoutOptions<T>,
timeoutMs?: number,
message?: string,
): Promise<T> {
const options =
typeof timeoutMs === "number"
? { promise: promiseOrOptions as Promise<T>, timeoutMs, message }
: resolveTimeoutOptions(promiseOrOptions as TimeoutOptions<T>);
if (typeof options.timeoutMs !== "number" || !options.message) {
return Promise.reject(new Error("Timeout duration and message are required"));
}
let timeout: ReturnType<typeof setTimeout> | undefined;
const timeoutPromise = new Promise<never>((_, reject) => {
timeout = setTimeout(() => reject(new Error(options.message)), options.timeoutMs);
});
return Promise.race([options.promise, timeoutPromise]).finally(() => {
if (timeout) {
clearTimeout(timeout);
}
});
}
function resolveTimeoutOptions<T>(options: TimeoutOptions<T>): {
promise: Promise<T>;
timeoutMs: number;
message: string;
} {
return {
promise: options.promise,
timeoutMs: options.timeoutMs,
message: `Timed out after ${options.timeoutMs}ms (${options.label})`,
};
}