fix(providers): tolerate slow Pi and OMP RPC startup (#4008)

MCP and extension initialization can keep provider control-plane requests pending beyond 30 seconds even when the response will arrive. Use a configurable 60-second provider default and include the active RPC phase in timeout errors.
This commit is contained in:
Mohamed Boudra
2026-08-29 00:00:41 +02:00
committed by GitHub
parent 693280d996
commit 463415ae84
14 changed files with 120 additions and 20 deletions
+5 -3
View File
@@ -393,6 +393,7 @@ Custom OMP profiles should extend `omp`. They inherit the OMP adapter's `rpc-ui`
},
"params": {
"sessionDir": "~/.local/state/omp-work/omp/agent/sessions",
"rpcTimeoutMs": 60000,
"smolModel": "openai/gpt-5-mini",
"slowModel": "anthropic/claude-opus-4-1",
"planModel": "openai/o3"
@@ -403,7 +404,7 @@ Custom OMP profiles should extend `omp`. They inherit the OMP adapter's `rpc-ui`
}
```
`params.sessionDir` is used only for importing sessions that were started outside Paseo. If `command` or XDG env vars move OMP's state directory, set `params.sessionDir` to the resulting OMP JSONL session directory; launching and resuming still go through the configured command.
`params.sessionDir` is used only for importing sessions that were started outside Paseo. If `command` or XDG env vars move OMP's state directory, set `params.sessionDir` to the resulting OMP JSONL session directory; launching and resuming still go through the configured command. `params.rpcTimeoutMs` overrides the 60-second OMP control-plane RPC deadline.
For other providers that keep Pi's `--mode rpc` API but write sessions somewhere else, extend `pi`, replace the command, and provide the JSONL session directory:
@@ -416,7 +417,8 @@ For other providers that keep Pi's `--mode rpc` API but write sessions somewhere
"label": "My Pi Fork",
"command": ["my-pi-fork"],
"params": {
"sessionDir": "~/.my-pi-fork/sessions"
"sessionDir": "~/.my-pi-fork/sessions",
"rpcTimeoutMs": 60000
}
}
}
@@ -424,7 +426,7 @@ For other providers that keep Pi's `--mode rpc` API but write sessions somewhere
}
```
This session directory is also import-only. Launching and resuming still go through the configured command, so this example resumes with `my-pi-fork --mode rpc --session <session-file>`.
This session directory is also import-only. Launching and resuming still go through the configured command, so this example resumes with `my-pi-fork --mode rpc --session <session-file>`. `params.rpcTimeoutMs` overrides the 60-second Pi control-plane RPC deadline.
---
+2
View File
@@ -67,6 +67,8 @@ Pi model records expose input capabilities through `model.input`. Only send raw
Pi MCP support depends on the open-source `pi-mcp-adapter` extension being loaded for the agent cwd. Probe with Pi RPC `get_commands`; the adapter registers an extension command named `mcp` (often with `sourceInfo.source` containing `pi-mcp-adapter`). When Paseo injects MCP servers into Pi, write a per-agent MCP config and pass it with `--mcp-config` instead of modifying user or project MCP files. Because that flag replaces the Pi global config layer, preserve the existing `<Pi agent dir>/mcp.json` in the generated file before overlaying injected servers. For local HTTP servers such as Paseo's own `/mcp/agents` endpoint, explicitly disable adapter OAuth (`auth: false`, `oauth: false`) in the generated config.
Pi and OMP control-plane RPCs wait 60 seconds by default. Override the provider's `params.rpcTimeoutMs` when extension or MCP startup on a slow host needs a different deadline. Timeout errors name the pending RPC phase and report both elapsed time and the configured deadline. This setting does not govern long-running Pi compaction or Pi extension UI results.
Pi import discovery reads Pi's persisted JSONL session files because Pi RPC does not expose a recent-session listing command. Resume and full history hydration still go through `pi --mode rpc` using the session file as `nativeHandle`.
OMP is a first-class built-in provider, disabled by default. Its launch contract, typed runtime, agent/session behavior, history, permissions, imports, and test fake live under `providers/omp/`; only the provider-neutral JSONL child-process transport is shared with Pi. It launches `omp --mode rpc-ui`, uses OMP's `get_available_commands` RPC for slash-command discovery, bridges OMP `rpc-ui` approval dialogs into Paseo permissions, and imports terminal-started sessions from `~/.omp/agent/sessions` when enabled.
@@ -62,6 +62,7 @@ interface InMemoryChildProcess extends ChildProcessWithoutNullStreams {
interface StartProcessOptions {
child?: ChildProcessWithoutNullStreams;
defaultRequestTimeoutMs?: number;
}
function createInMemoryChildProcess(): InMemoryChildProcess {
@@ -89,6 +90,7 @@ function startProcess(options: StartProcessOptions = {}): JsonlRpcProcess {
env: { JSONL_RPC_TEST_VALUE: "resolved-env" },
},
logger: pino({ level: "silent" }),
defaultRequestTimeoutMs: options.defaultRequestTimeoutMs,
...(child ? { spawn: () => child } : {}),
});
}
@@ -157,13 +159,13 @@ describe("JsonlRpcProcess", () => {
test("includes buffered stderr when a request times out", async () => {
const child = createInMemoryChildProcess();
const transport = startProcess({ child });
const transport = startProcess({ child, defaultRequestTimeoutMs: 50 });
try {
child.stderr.write("still waiting");
await expect(transport.request({ type: "hang" }, 50)).rejects.toThrow(
"JSONL RPC request timed out for hang\nstill waiting",
await expect(transport.request({ type: "hang" })).rejects.toThrow(
/JSONL RPC request timed out phase=hang elapsedMs=\d+ timeoutMs=50\nstill waiting/,
);
} finally {
await transport.close();
@@ -50,6 +50,7 @@ export interface JsonlRpcProcessOptions {
launch: JsonlRpcLaunch;
logger: Logger;
diagnosticName?: string;
defaultRequestTimeoutMs?: number;
spawn?: (launch: JsonlRpcLaunch) => ChildProcessWithoutNullStreams;
}
@@ -134,7 +135,7 @@ export class JsonlRpcProcess {
startRequest(
command: { type: string; [key: string]: unknown },
timeoutMs: number | null = JSONL_RPC_DEFAULT_TIMEOUT_MS,
timeoutMs?: number | null,
): { id: string; promise: Promise<unknown> } {
if (this.disposed) {
return {
@@ -144,12 +145,17 @@ export class JsonlRpcProcess {
}
const id = `req_${this.nextRequestId}`;
this.nextRequestId += 1;
const requestTimeoutMs =
timeoutMs === undefined
? (this.options.defaultRequestTimeoutMs ?? JSONL_RPC_DEFAULT_TIMEOUT_MS)
: timeoutMs;
const startedAt = Date.now();
const promise = new Promise<unknown>((resolve, reject) => {
const timer = createRequestTimeout(timeoutMs, () => {
const timer = createRequestTimeout(requestTimeoutMs, () => {
this.pending.delete(id);
reject(
new Error(
`${this.diagnosticName} request timed out for ${command.type}\n${this.stderrBuffer}`.trim(),
`${this.diagnosticName} request timed out phase=${command.type} elapsedMs=${Date.now() - startedAt} timeoutMs=${requestTimeoutMs}\n${this.stderrBuffer}`.trim(),
),
);
});
@@ -161,7 +167,7 @@ export class JsonlRpcProcess {
request(
command: { type: string; [key: string]: unknown },
timeoutMs: number | null = JSONL_RPC_DEFAULT_TIMEOUT_MS,
timeoutMs?: number | null,
): Promise<unknown> {
return this.startRequest(command, timeoutMs).promise;
}
@@ -4,8 +4,16 @@ import { setImmediate as waitForImmediate } from "node:timers/promises";
import type { PaseoToolCatalog } from "../../tools/types.js";
import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js";
import type { OmpUsagePollScheduler } from "./usage-poller.js";
import { resolveOmpProviderParams } from "./provider-config.js";
import { OmpHarness } from "./test-utils/omp-harness.js";
test("OMP RPC timeout defaults to 60 seconds and accepts an override", () => {
expect(resolveOmpProviderParams({}).runtimeProviderParams.rpcTimeoutMs).toBe(60_000);
expect(
resolveOmpProviderParams({ rpcTimeoutMs: 90_000 }).runtimeProviderParams.rpcTimeoutMs,
).toBe(90_000);
});
class ManualIdleScheduler implements OmpProviderIdleScheduler {
private readonly retries: Array<() => void> = [];
private readonly waiters: Array<{ count: number; resolve: () => void }> = [];
@@ -828,12 +828,14 @@ function buildExtensionUiResponse(
function createRuntime(
logger: Logger,
runtimeSettings: ProviderRuntimeSettings | undefined,
requestTimeoutMs: number,
): OmpRuntime {
return new OmpCliRuntime({
logger,
runtimeSettings,
command: ["omp"],
commandsRpcName: "get_available_commands",
requestTimeoutMs,
});
}
@@ -2213,7 +2215,9 @@ export class OmpAgentClient implements AgentClient {
this.providerIdleScheduler = options.providerIdleScheduler;
this.noTurnScheduler = options.noTurnScheduler;
this.usagePollScheduler = options.usagePollScheduler;
this.runtime = options.runtime ?? createRuntime(options.logger, runtimeSettings);
this.runtime =
options.runtime ??
createRuntime(options.logger, runtimeSettings, this.providerParams.rpcTimeoutMs);
}
private async configureNativePaseoTools(
@@ -2,7 +2,7 @@ import type { ChildProcessWithoutNullStreams } from "node:child_process";
import { EventEmitter } from "node:events";
import { PassThrough } from "node:stream";
import pino from "pino";
import { describe, expect, test } from "vitest";
import { describe, expect, test, vi } from "vitest";
import { OmpCliRuntime } from "./cli-runtime.js";
import type { OmpRuntimeLaunch } from "./runtime.js";
@@ -49,11 +49,16 @@ function createOmpChild(options?: {
return child;
}
function createRuntime(child: OmpChild, launches: OmpRuntimeLaunch[] = []): OmpCliRuntime {
function createRuntime(
child: OmpChild,
launches: OmpRuntimeLaunch[] = [],
options?: { requestTimeoutMs?: number },
): OmpCliRuntime {
return new OmpCliRuntime({
logger: pino({ level: "silent" }),
command: ["omp"],
commandsRpcName: "get_available_commands",
requestTimeoutMs: options?.requestTimeoutMs,
spawnProcess: (launch) => {
launches.push(launch);
return child;
@@ -94,6 +99,26 @@ function withoutRequestId(command: Record<string, unknown>): Record<string, unkn
}
describe("OMP CLI runtime", () => {
test("uses the configured RPC timeout and attributes the pending phase", async () => {
vi.useFakeTimers();
const child = createOmpChild();
const session = await createRuntime(child, [], { requestTimeoutMs: 100 }).startSession({
cwd: "/workspace/project",
});
try {
const state = session.getState();
const rejection = expect(state).rejects.toThrow(
"OMP RPC request timed out phase=get_state elapsedMs=100 timeoutMs=100",
);
await vi.advanceTimersByTimeAsync(100);
await rejection;
} finally {
vi.useRealTimers();
await session.close();
}
});
test("validates session state with the documented queued message count", async () => {
const child = createOmpChild();
replyToCommands(child, () => ({
@@ -47,6 +47,7 @@ export interface OmpCliRuntimeOptions {
runtimeSettings?: ProviderRuntimeSettings;
command?: [string, ...string[]];
commandsRpcName?: "get_available_commands";
requestTimeoutMs?: number;
spawnProcess?: (launch: OmpRuntimeLaunch) => ChildProcessWithoutNullStreams;
}
@@ -79,13 +80,14 @@ export class OmpCliRuntime implements OmpRuntime {
launch: processLaunch,
logger: this.options.logger,
diagnosticName: "OMP RPC",
defaultRequestTimeoutMs: this.options.requestTimeoutMs,
...(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 establishOmpProtocol(process, this.options.logger);
await establishOmpProtocol(process, this.options.logger, this.options.requestTimeoutMs);
input.signal?.throwIfAborted();
return new OmpCliRuntimeSession(process, this.commandsRpcName);
} catch (error) {
@@ -16,12 +16,13 @@ export interface OmpProtocolTransport {
export async function establishOmpProtocol(
transport: OmpProtocolTransport,
logger: Logger,
requestTimeoutMs = JSONL_RPC_DEFAULT_TIMEOUT_MS,
): Promise<void> {
const ready = await waitForReady(transport);
if (!supportsJsonlRpcProtocolV2(ready)) return;
const response = (await transport.request(
{ type: "negotiate_protocol", protocolVersion: 2 },
JSONL_RPC_DEFAULT_TIMEOUT_MS,
requestTimeoutMs,
)) as { protocolVersion?: unknown } | undefined;
if (response?.protocolVersion !== 2) throw new Error("OMP did not accept RPC protocol v2");
logger.debug({}, "Negotiated OMP RPC protocol v2 (chunked frame transport)");
@@ -8,6 +8,7 @@ import type { ProviderRuntimeSettings } from "../../provider-launch-config.js";
const OMP_SESSION_DIR = "~/.omp/agent/sessions";
const DEFAULT_OMP_MODE_ID = "full";
const DEFAULT_OMP_RPC_TIMEOUT_MS = 60_000;
export const MIN_SUPPORTED_OMP_VERSION = "16.3.9";
export { OMP_MODES };
@@ -15,6 +16,7 @@ export { OMP_MODES };
export const OmpProviderParamsSchema = z
.object({
sessionDir: z.string().min(1).optional(),
rpcTimeoutMs: z.number().int().positive().default(DEFAULT_OMP_RPC_TIMEOUT_MS),
smolModel: z.string().min(1).optional(),
slowModel: z.string().min(1).optional(),
planModel: z.string().min(1).optional(),
@@ -23,6 +25,7 @@ export const OmpProviderParamsSchema = z
export interface OmpRuntimeProviderParams {
sessionDir: string;
rpcTimeoutMs: number;
}
export interface OmpModelRoleParams {
@@ -130,7 +133,10 @@ export function resolveOmpProviderParams(providerParams: unknown): {
} {
const params = OmpProviderParamsSchema.parse(providerParams ?? {});
return {
runtimeProviderParams: { sessionDir: params.sessionDir ?? OMP_SESSION_DIR },
runtimeProviderParams: {
sessionDir: params.sessionDir ?? OMP_SESSION_DIR,
rpcTimeoutMs: params.rpcTimeoutMs,
},
modelRoleParams: {
...(params.smolModel ? { smolModel: params.smolModel } : {}),
...(params.slowModel ? { slowModel: params.slowModel } : {}),
@@ -17,13 +17,23 @@ import { pathToFileURL } from "node:url";
import { describe, expect, onTestFinished, test } from "vitest";
import type { AgentSession, AgentSessionConfig, AgentStreamEvent } from "../../agent-sdk-types.js";
import { PiRpcAgentClient, PiRpcAgentSession, transformPiModels } from "./agent.js";
import {
PiProviderParamsSchema,
PiRpcAgentClient,
PiRpcAgentSession,
transformPiModels,
} from "./agent.js";
import { FakePi } from "./test-utils/fake-pi.js";
import type { PiUsagePollScheduler } from "./usage-poller.js";
const ONE_BY_ONE_PNG_BASE64 =
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/p9sAAAAASUVORK5CYII=";
test("Pi RPC timeout defaults to 60 seconds and accepts an override", () => {
expect(PiProviderParamsSchema.parse({}).rpcTimeoutMs).toBe(60_000);
expect(PiProviderParamsSchema.parse({ rpcTimeoutMs: 90_000 }).rpcTimeoutMs).toBe(90_000);
});
function createClient(
pi = new FakePi(),
usagePollScheduler?: PiUsagePollScheduler,
@@ -95,6 +95,7 @@ const PASEO_PI_ENTRY_CAPTURE_MARKER = "PASEO_ENTRY_CAPTURE";
const PASEO_PI_SUBMITTED_USER_ENTRY_MARKER = "PASEO_SUBMITTED_USER_ENTRY";
const PASEO_PI_COMMAND_RESULT_MARKER = "PASEO_COMMAND_RESULT";
const DEFAULT_PI_EXTENSION_RESULT_TIMEOUT_MS = 30_000;
const DEFAULT_PI_RPC_TIMEOUT_MS = 60_000;
const QUESTION_RESPONSE_HEADER = "Response";
const QUESTION_COMMENT_HEADER = "Comment";
const PI_ASK_USER_FREEFORM_SENTINEL = "✏️ Type custom response...";
@@ -103,6 +104,7 @@ const COMBINED_ASK_USER_METADATA = "ask_user_select_optional_comment";
export const PiProviderParamsSchema = z
.object({
sessionDir: z.string().min(1).optional(),
rpcTimeoutMs: z.number().int().positive().default(DEFAULT_PI_RPC_TIMEOUT_MS),
extensionTimeoutMs: z.number().int().positive().default(DEFAULT_PI_EXTENSION_RESULT_TIMEOUT_MS),
})
.strict();
@@ -1184,12 +1186,17 @@ function mapPiModel(model: PiModel, provider: AgentProvider): AgentModelDefiniti
};
}
function createRuntime(logger: Logger, runtimeSettings?: ProviderRuntimeSettings): PiRuntime {
function createRuntime(
logger: Logger,
runtimeSettings: ProviderRuntimeSettings | undefined,
requestTimeoutMs: number,
): PiRuntime {
return new PiCliRuntime({
logger,
runtimeSettings,
command: [PI_BINARY_COMMAND],
commandsRpcName: "get_commands",
requestTimeoutMs,
});
}
@@ -2407,7 +2414,9 @@ export class PiRpcAgentClient implements AgentClient {
this.logger = options.logger;
this.runtimeSettings = options.runtimeSettings;
this.providerParams = PiProviderParamsSchema.parse(options.providerParams ?? {});
this.runtime = options.runtime ?? createRuntime(options.logger, options.runtimeSettings);
this.runtime =
options.runtime ??
createRuntime(options.logger, options.runtimeSettings, this.providerParams.rpcTimeoutMs);
this.usagePollScheduler = options.usagePollScheduler;
}
@@ -34,12 +34,13 @@ function createPiChild(): PiChild {
function createRuntime(
child: PiChild,
launches: PiRuntimeLaunch[] = [],
options?: { commandsRpcName?: string },
options?: { commandsRpcName?: string; requestTimeoutMs?: number },
): PiCliRuntime {
return new PiCliRuntime({
logger: pino({ level: "silent" }),
command: ["pi"],
commandsRpcName: options?.commandsRpcName,
requestTimeoutMs: options?.requestTimeoutMs,
spawnProcess: (launch) => {
launches.push(launch);
return child;
@@ -293,6 +294,26 @@ describe("PiCliRuntime", () => {
await rejection;
});
test("uses the configured RPC timeout and attributes the pending phase", async () => {
vi.useFakeTimers();
const child = createPiChild();
const session = await createRuntime(child, [], { requestTimeoutMs: 100 }).startSession({
cwd: "/workspace/project",
});
try {
const state = session.getState();
const rejection = expect(state).rejects.toThrow(
"Pi RPC request timed out phase=get_state elapsedMs=100 timeoutMs=100",
);
await vi.advanceTimersByTimeAsync(100);
await rejection;
} finally {
vi.useRealTimers();
await session.close();
}
});
test("compact waits beyond the default control-plane timeout for a late response", async () => {
vi.useFakeTimers();
const child = createPiChild();
@@ -35,6 +35,7 @@ export interface PiCliRuntimeOptions {
runtimeSettings?: ProviderRuntimeSettings;
command?: [string, ...string[]];
commandsRpcName?: string;
requestTimeoutMs?: number;
spawnProcess?: (launch: PiRuntimeLaunch) => ChildProcessWithoutNullStreams;
}
@@ -67,6 +68,7 @@ export class PiCliRuntime implements PiRuntime {
launch: processLaunch,
logger: this.options.logger,
diagnosticName: "Pi RPC",
defaultRequestTimeoutMs: this.options.requestTimeoutMs,
...(spawn ? { spawn: () => spawn(launch) } : {}),
};
const process = new JsonlRpcProcess(processOptions);