From b4518cbf3352210cefec1715c00ad000c4abd20f Mon Sep 17 00:00:00 2001 From: "Jason@HND" Date: Sat, 18 Jul 2026 22:20:03 +0900 Subject: [PATCH] fix(server): remove wall-clock timeout for Pi compact RPC (#2181) Pi compact is a blocking LLM summarization job that often exceeds the default 30s control-plane timeout on long sessions, causing a false UI error while the real compact continues. Wait for the RPC response, process exit, or session close instead. Also cover the no-timeout lifecycle: null timeout still rejects on close, and compact waits past 30s for a late success response. Closes #1946 --- .../agent/providers/jsonl-rpc-process.test.ts | 23 +++ .../agent/providers/jsonl-rpc-process.ts | 41 ++++-- .../agent/providers/pi/cli-runtime.test.ts | 134 ++++++++++++++---- .../server/agent/providers/pi/cli-runtime.ts | 32 ++++- .../src/server/agent/providers/pi/runtime.ts | 7 +- .../agent/providers/pi/test-utils/fake-pi.ts | 2 +- 6 files changed, 192 insertions(+), 47 deletions(-) diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts index f8c059307..26e33043f 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts @@ -140,6 +140,29 @@ describe("JsonlRpcProcess", () => { } }); + test("null timeout waits past short wall-clock limits until the response arrives", async () => { + const transport = startProcess(); + + try { + await expect( + transport.request({ type: "echo", value: "slow", delayMs: 80 }, null), + ).resolves.toMatchObject({ value: "slow" }); + } finally { + await transport.close(); + } + }); + + test("null timeout still rejects when the process is closed", async () => { + const transport = startProcess(); + await transport.request({ type: "echo", value: "ready" }); + const request = transport.request({ type: "hang" }, null); + + const rejection = expect(request).rejects.toThrow("JSONL RPC process is closed"); + await transport.close(); + + await rejection; + }); + test("rejects pending requests and publishes stderr when the child exits", async () => { const transport = startProcess(); const exit = nextExit(transport); diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts index 2fddf5995..2e56a0659 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts @@ -4,7 +4,14 @@ import type { Logger } from "pino"; import { spawnProcess } from "../../../utils/spawn.js"; import { terminateWithTreeKill } from "../../../utils/tree-kill.js"; -const DEFAULT_TIMEOUT_MS = 30_000; +/** Default wall-clock timeout for control-plane / short RPC calls. */ +export const JSONL_RPC_DEFAULT_TIMEOUT_MS = 30_000; +/** + * Pass as `timeoutMs` to wait only for a response, process death, or `close()`. + * Use for long-running blocking RPCs (e.g. LLM-backed compact). + */ +export const JSONL_RPC_NO_TIMEOUT = null; + const STDERR_BUFFER_LIMIT = 8192; const GRACEFUL_SHUTDOWN_TIMEOUT_MS = 2_000; const FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; @@ -28,7 +35,7 @@ interface JsonlRpcResponse { interface PendingRequest { resolve: (value: unknown) => void; reject: (error: Error) => void; - timer: NodeJS.Timeout; + timer: NodeJS.Timeout | null; } export interface JsonlRpcExit { @@ -116,7 +123,7 @@ export class JsonlRpcProcess { startRequest( command: { type: string; [key: string]: unknown }, - timeoutMs = DEFAULT_TIMEOUT_MS, + timeoutMs: number | null = JSONL_RPC_DEFAULT_TIMEOUT_MS, ): { id: string; promise: Promise } { if (this.disposed) { return { @@ -127,14 +134,14 @@ export class JsonlRpcProcess { const id = `req_${this.nextRequestId}`; this.nextRequestId += 1; const promise = new Promise((resolve, reject) => { - const timer = setTimeout(() => { + const timer = createRequestTimeout(timeoutMs, () => { this.pending.delete(id); reject( new Error( `${this.diagnosticName} request timed out for ${command.type}\n${this.stderrBuffer}`.trim(), ), ); - }, timeoutMs); + }); this.pending.set(id, { resolve, reject, timer }); this.send({ ...command, id }); }); @@ -143,7 +150,7 @@ export class JsonlRpcProcess { request( command: { type: string; [key: string]: unknown }, - timeoutMs = DEFAULT_TIMEOUT_MS, + timeoutMs: number | null = JSONL_RPC_DEFAULT_TIMEOUT_MS, ): Promise { return this.startRequest(command, timeoutMs).promise; } @@ -228,7 +235,9 @@ export class JsonlRpcProcess { if (!pending) { return; } - clearTimeout(pending.timer); + if (pending.timer) { + clearTimeout(pending.timer); + } this.pending.delete(response.id); if (!response.success) { pending.reject( @@ -247,9 +256,25 @@ export class JsonlRpcProcess { } this.disposed = true; for (const pending of this.pending.values()) { - clearTimeout(pending.timer); + if (pending.timer) { + clearTimeout(pending.timer); + } pending.reject(error); } this.pending.clear(); } } + +/** + * Schedule a request timeout, or return null when the call should wait + * indefinitely for a response, process exit, or close(). + */ +function createRequestTimeout( + timeoutMs: number | null, + onTimeout: () => void, +): NodeJS.Timeout | null { + if (timeoutMs == null || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { + return null; + } + return setTimeout(onTimeout, timeoutMs); +} diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts index 07cf1341d..f3b2af84a 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts @@ -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 { PiCliRuntime } from "./cli-runtime.js"; import type { PiRuntimeLaunch } from "./runtime.js"; @@ -47,10 +47,7 @@ function createRuntime( }); } -function replyToCommands( - child: PiChild, - handler: (command: Record) => unknown, -): void { +function onPiCommand(child: PiChild, handler: (command: Record) => void): void { let buffer = ""; child.stdin.on("data", (chunk) => { buffer += chunk.toString(); @@ -59,34 +56,68 @@ function replyToCommands( if (newlineIndex === -1) break; const line = buffer.slice(0, newlineIndex); buffer = buffer.slice(newlineIndex + 1); - const command = JSON.parse(line) as Record; - try { - const result = handler(command); - child.stdout.write( - `${JSON.stringify({ - id: command.id, - type: "response", - command: command.type, - success: true, - data: result, - })}\n`, - ); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - child.stdout.write( - `${JSON.stringify({ - id: command.id, - type: "response", - command: command.type, - success: false, - error: message, - })}\n`, - ); - } + handler(JSON.parse(line) as Record); } }); } +function replyToCommands( + child: PiChild, + handler: (command: Record) => unknown, +): void { + onPiCommand(child, (command) => { + try { + const result = handler(command); + child.stdout.write( + `${JSON.stringify({ + id: command.id, + type: "response", + command: command.type, + success: true, + data: result, + })}\n`, + ); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + child.stdout.write( + `${JSON.stringify({ + id: command.id, + type: "response", + command: command.type, + success: false, + error: message, + })}\n`, + ); + } + }); +} + +function capturePendingCommand(child: PiChild, type: string): Promise> { + return new Promise((resolve) => { + onPiCommand(child, (command) => { + if (command.type === type) { + resolve(command); + } + }); + }); +} + +function writePiResponse( + child: PiChild, + command: Record, + data: unknown = {}, +): void { + child.stdout.write( + `${JSON.stringify({ + id: command.id, + type: "response", + command: command.type, + success: true, + data, + })}\n`, + ); +} + describe("PiCliRuntime", () => { test("starts pi in rpc mode and resolves command responses", async () => { const child = createPiChild(); @@ -282,6 +313,51 @@ describe("PiCliRuntime", () => { await rejection; }); + test("compact waits beyond the default control-plane timeout for a late response", async () => { + vi.useFakeTimers(); + const child = createPiChild(); + const pendingCompact = capturePendingCommand(child, "compact"); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + + try { + const compactPromise = session.compact("focus on tests"); + const compactCommand = await pendingCompact; + await vi.advanceTimersByTimeAsync(35_000); + + expect(compactCommand).toMatchObject({ + type: "compact", + customInstructions: "focus on tests", + id: expect.any(String), + }); + + writePiResponse(child, compactCommand, { + summary: "done", + firstKeptEntryId: "entry-1", + tokensBefore: 120_000, + estimatedTokensAfter: 20_000, + }); + + await expect(compactPromise).resolves.toBeUndefined(); + } finally { + vi.useRealTimers(); + await session.close(); + } + }); + + test("compact without a wall-clock timeout rejects when the session closes", async () => { + const child = createPiChild(); + const pendingCompact = capturePendingCommand(child, "compact"); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + + const compactPromise = session.compact(); + await pendingCompact; + + const rejection = expect(compactPromise).rejects.toThrow("Pi RPC session is closed"); + await session.close(); + + await rejection; + }); + test("disposes the Pi process", async () => { const child = createPiChild(); const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.ts index aee7b2a78..66cff0daf 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.ts @@ -2,7 +2,11 @@ import type { ChildProcessWithoutNullStreams } from "node:child_process"; import type { Logger } from "pino"; import type { ProviderRuntimeSettings } from "../../provider-launch-config.js"; -import { JsonlRpcProcess, type JsonlRpcLaunch } from "../jsonl-rpc-process.js"; +import { + JSONL_RPC_NO_TIMEOUT, + JsonlRpcProcess, + type JsonlRpcLaunch, +} from "../jsonl-rpc-process.js"; import { buildPiLaunch, type PiRuntime, @@ -26,6 +30,14 @@ const DEFAULT_PI_COMMAND: [string, ...string[]] = [ ]; const DEFAULT_COMMANDS_RPC_NAME = "get_commands"; +/** + * Pi RPC timeout policy: + * - Control-plane / accept-and-stream (`prompt`, `get_state`, `abort`, …): default 30s + * - Long-running blocking LLM jobs (`compact`): no wall-clock timeout — complete on + * response, process death, or session close (`JsonlRpcProcess.failAll` / `close`). + */ +const PI_COMPACT_REQUEST_TIMEOUT_MS = JSONL_RPC_NO_TIMEOUT; + export interface PiCliRuntimeOptions { logger: Logger; runtimeSettings?: ProviderRuntimeSettings; @@ -112,10 +124,16 @@ class PiCliRuntimeSession implements PiRuntimeSession { } async compact(customInstructions?: string): Promise { - await this.request({ - type: "compact", - ...(customInstructions ? { customInstructions } : {}), - }); + // Compact is a blocking LLM summarization job; Pi only returns the RPC + // response after the summary is written. A control-plane 30s timeout falsely + // fails long sessions while the real compact continues (issue #1946). + await this.request( + { + type: "compact", + ...(customInstructions ? { customInstructions } : {}), + }, + PI_COMPACT_REQUEST_TIMEOUT_MS, + ); } async setAutoCompaction(enabled: boolean): Promise { @@ -135,7 +153,7 @@ class PiCliRuntimeSession implements PiRuntimeSession { return data.messages ?? []; } - async getAvailableModels(timeoutMs?: number): Promise { + async getAvailableModels(timeoutMs?: number | null): Promise { const data = (await this.request({ type: "get_available_models" }, timeoutMs)) as { models?: PiModel[]; }; @@ -208,7 +226,7 @@ class PiCliRuntimeSession implements PiRuntimeSession { await this.process.close(new Error("Pi RPC session is closed")); } - request(command: PiRpcCommand, timeoutMs?: number): Promise { + request(command: PiRpcCommand, timeoutMs?: number | null): Promise { return this.process.request(command, timeoutMs); } diff --git a/packages/server/src/server/agent/providers/pi/runtime.ts b/packages/server/src/server/agent/providers/pi/runtime.ts index 0b4d38afa..6d50693ed 100644 --- a/packages/server/src/server/agent/providers/pi/runtime.ts +++ b/packages/server/src/server/agent/providers/pi/runtime.ts @@ -51,12 +51,15 @@ export interface PiRuntimeSession { abort(): Promise; getState(): Promise; getMessages(): Promise; - getAvailableModels(timeoutMs?: number): Promise; + getAvailableModels(timeoutMs?: number | null): Promise; setModel(provider: string, modelId: string): Promise; setThinkingLevel(level: string): Promise; getSessionStats(): Promise; getCommands(): Promise; - request(command: { type: string; [key: string]: unknown }, timeoutMs?: number): Promise; + request( + command: { type: string; [key: string]: unknown }, + timeoutMs?: number | null, + ): Promise; sendRawFrame(frame: object & { type: string }): void; respondToExtensionUiRequest( id: string, diff --git a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts index 8332c3f50..8093baeb2 100644 --- a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts +++ b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts @@ -227,7 +227,7 @@ export class FakePiSession implements PiRuntimeSession { return this.messages; } - async getAvailableModels(_timeoutMs?: number): Promise { + async getAvailableModels(_timeoutMs?: number | null): Promise { return this.models; }