From e36759115c60e7696a985fdfeb55962d20245c22 Mon Sep 17 00:00:00 2001 From: weishu Date: Thu, 22 Jan 2026 11:53:17 +0800 Subject: [PATCH] feat: improve ACP stability and UX with initialization retry and error handling - AcpStdioTransport: Fix stdout JSON validation to filter non-object JSON (numbers, strings, booleans) that could break protocol - AcpStdioTransport: Add request timeout with 2-minute default, properly cleared on settle, using unref() for graceful shutdown - AcpStdioTransport: Add stderr error parsing for rate limit (429), model not found (404), auth errors, quota exceeded - AcpSdkBackend: Add initialization retry mechanism using withRetry() for initialize and session/new (3 attempts, 1-5s backoff) - AcpSdkBackend: Add waitForResponseComplete() for safe session swaps and preventing race conditions - AcpSdkBackend: Add onStderrError() handler and processingMessage getter for error reporting - AcpSdkBackend: Use Infinity timeout for session/prompt to support long-running tasks --- cli/src/agent/backends/acp/AcpSdkBackend.ts | 103 ++++++++++++--- .../agent/backends/acp/AcpStdioTransport.ts | 120 +++++++++++++++++- 2 files changed, 203 insertions(+), 20 deletions(-) diff --git a/cli/src/agent/backends/acp/AcpSdkBackend.ts b/cli/src/agent/backends/acp/AcpSdkBackend.ts index dc27dfd0..b64a15af 100644 --- a/cli/src/agent/backends/acp/AcpSdkBackend.ts +++ b/cli/src/agent/backends/acp/AcpSdkBackend.ts @@ -1,8 +1,9 @@ import type { AgentBackend, AgentMessage, AgentSessionConfig, PermissionRequest, PermissionResponse, PromptContent } from '@/agent/types'; import { asString, isObject } from '@/agent/utils'; -import { AcpStdioTransport } from './AcpStdioTransport'; +import { AcpStdioTransport, type AcpStderrError } from './AcpStdioTransport'; import { AcpMessageHandler } from './AcpMessageHandler'; import { logger } from '@/ui/logger'; +import { withRetry } from '@/utils/time'; import packageJson from '../../../../package.json'; type PendingPermission = { @@ -12,9 +13,19 @@ type PendingPermission = { export class AcpSdkBackend implements AgentBackend { private transport: AcpStdioTransport | null = null; private permissionHandler: ((request: PermissionRequest) => void) | null = null; + private stderrErrorHandler: ((error: AcpStderrError) => void) | null = null; private readonly pendingPermissions = new Map(); private messageHandler: AcpMessageHandler | null = null; private activeSessionId: string | null = null; + private isProcessingMessage = false; + private responseCompleteResolvers: Array<() => void> = []; + + /** Retry configuration for ACP initialization */ + private static readonly INIT_RETRY_OPTIONS = { + maxAttempts: 3, + minDelay: 1000, + maxDelay: 5000 + }; constructor(private readonly options: { command: string; args?: string[]; env?: Record }) {} @@ -33,21 +44,33 @@ export class AcpSdkBackend implements AgentBackend { } }); + this.transport.onStderrError((error) => { + this.stderrErrorHandler?.(error); + }); + this.transport.registerRequestHandler('session/request_permission', async (params, requestId) => { return await this.handlePermissionRequest(params, requestId); }); - const response = await this.transport.sendRequest('initialize', { - protocolVersion: 1, - clientCapabilities: { - fs: { readTextFile: false, writeTextFile: false }, - terminal: false - }, - clientInfo: { - name: 'hapi', - version: packageJson.version + const response = await withRetry( + () => this.transport!.sendRequest('initialize', { + protocolVersion: 1, + clientCapabilities: { + fs: { readTextFile: false, writeTextFile: false }, + terminal: false + }, + clientInfo: { + name: 'hapi', + version: packageJson.version + } + }), + { + ...AcpSdkBackend.INIT_RETRY_OPTIONS, + onRetry: (error, attempt, nextDelayMs) => { + logger.debug(`[ACP] Initialize attempt ${attempt} failed, retrying in ${nextDelayMs}ms`, error); + } } - }); + ); if (!isObject(response) || typeof response.protocolVersion !== 'number') { throw new Error('Invalid initialize response from ACP agent'); @@ -61,10 +84,18 @@ export class AcpSdkBackend implements AgentBackend { throw new Error('ACP transport not initialized'); } - const response = await this.transport.sendRequest('session/new', { - cwd: config.cwd, - mcpServers: config.mcpServers - }); + const response = await withRetry( + () => this.transport!.sendRequest('session/new', { + cwd: config.cwd, + mcpServers: config.mcpServers + }), + { + ...AcpSdkBackend.INIT_RETRY_OPTIONS, + onRetry: (error, attempt, nextDelayMs) => { + logger.debug(`[ACP] session/new attempt ${attempt} failed, retrying in ${nextDelayMs}ms`, error); + } + } + ); const sessionId = isObject(response) ? asString(response.sessionId) : null; if (!sessionId) { @@ -86,12 +117,15 @@ export class AcpSdkBackend implements AgentBackend { this.activeSessionId = sessionId; this.messageHandler = new AcpMessageHandler(onUpdate); + this.isProcessingMessage = true; try { + // No timeout for prompt requests - they can run for extended periods + // during complex tasks, tool-heavy operations, or slow model responses const response = await this.transport.sendRequest('session/prompt', { sessionId, prompt: content - }); + }, { timeoutMs: Infinity }); const stopReason = isObject(response) ? asString(response.stopReason) : null; if (stopReason) { @@ -99,6 +133,8 @@ export class AcpSdkBackend implements AgentBackend { } } finally { this.messageHandler = null; + this.isProcessingMessage = false; + this.notifyResponseComplete(); } } @@ -140,6 +176,33 @@ export class AcpSdkBackend implements AgentBackend { this.permissionHandler = handler; } + onStderrError(handler: (error: AcpStderrError) => void): void { + this.stderrErrorHandler = handler; + } + + /** + * Returns true if currently processing a message (prompt in progress). + * Useful for checking if it's safe to perform session operations. + */ + get processingMessage(): boolean { + return this.isProcessingMessage; + } + + /** + * Wait for any in-progress response to complete. + * Resolves immediately if no response is being processed. + * Use this before performing operations that require the response to be complete, + * like session swap or sending task_complete. + */ + async waitForResponseComplete(): Promise { + if (!this.isProcessingMessage) { + return; + } + return new Promise((resolve) => { + this.responseCompleteResolvers.push(resolve); + }); + } + async disconnect(): Promise { if (!this.transport) return; await this.transport.close(); @@ -201,4 +264,12 @@ export class AcpSdkBackend implements AgentBackend { this.pendingPermissions.set(toolCallId, { resolve }); }); } + + private notifyResponseComplete(): void { + const resolvers = this.responseCompleteResolvers; + this.responseCompleteResolvers = []; + for (const resolve of resolvers) { + resolve(); + } + } } diff --git a/cli/src/agent/backends/acp/AcpStdioTransport.ts b/cli/src/agent/backends/acp/AcpStdioTransport.ts index e8f83873..28673f00 100644 --- a/cli/src/agent/backends/acp/AcpStdioTransport.ts +++ b/cli/src/agent/backends/acp/AcpStdioTransport.ts @@ -28,6 +28,14 @@ interface JsonRpcResponse { type RequestHandler = (params: unknown, requestId: string | number | null) => Promise; +export type AcpStderrErrorType = 'rate_limit' | 'model_not_found' | 'authentication' | 'quota_exceeded' | 'unknown'; + +export type AcpStderrError = { + type: AcpStderrErrorType; + message: string; + raw: string; +}; + export class AcpStdioTransport { private readonly process: ChildProcessWithoutNullStreams; private readonly pending = new Map(); private readonly requestHandlers = new Map(); private notificationHandler: ((method: string, params: unknown) => void) | null = null; + private stderrErrorHandler: ((error: AcpStderrError) => void) | null = null; private buffer = ''; private nextId = 1; private protocolError: Error | null = null; @@ -56,7 +65,9 @@ export class AcpStdioTransport { this.process.stderr.setEncoding('utf8'); this.process.stderr.on('data', (chunk) => { - logger.debug(`[ACP][stderr] ${chunk.toString().trim()}`); + const text = chunk.toString().trim(); + logger.debug(`[ACP][stderr] ${text}`); + this.parseStderrError(text); }); this.process.on('exit', (code, signal) => { @@ -79,11 +90,18 @@ export class AcpStdioTransport { this.notificationHandler = handler; } + onStderrError(handler: ((error: AcpStderrError) => void) | null): void { + this.stderrErrorHandler = handler; + } + registerRequestHandler(method: string, handler: RequestHandler): void { this.requestHandlers.set(method, handler); } - async sendRequest(method: string, params?: unknown): Promise { + /** Default timeout for requests in milliseconds (2 minutes) */ + static readonly DEFAULT_TIMEOUT_MS = 120_000; + + async sendRequest(method: string, params?: unknown, options?: { timeoutMs?: number }): Promise { const id = this.nextId++; const payload: JsonRpcRequest = { jsonrpc: '2.0', @@ -92,8 +110,36 @@ export class AcpStdioTransport { params }; + const timeoutMs = options?.timeoutMs ?? AcpStdioTransport.DEFAULT_TIMEOUT_MS; + + // Skip timeout for infinite/no-timeout requests (e.g., long-running prompts) + if (!Number.isFinite(timeoutMs)) { + return new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject }); + this.writePayload(payload); + }); + } + return new Promise((resolve, reject) => { - this.pending.set(id, { resolve, reject }); + const timer = setTimeout(() => { + if (this.pending.has(id)) { + this.pending.delete(id); + reject(new Error(`ACP request '${method}' timed out after ${timeoutMs}ms`)); + } + }, timeoutMs); + // Don't let timer keep Node alive if process wants to exit + timer.unref(); + + this.pending.set(id, { + resolve: (value) => { + clearTimeout(timer); + resolve(value); + }, + reject: (error) => { + clearTimeout(timer); + reject(error); + } + }); this.writePayload(payload); }); } @@ -135,7 +181,14 @@ export class AcpStdioTransport { } let message: JsonRpcRequest | JsonRpcResponse | JsonRpcNotification | null = null; try { - message = JSON.parse(line) as JsonRpcRequest | JsonRpcResponse | JsonRpcNotification; + const parsed = JSON.parse(line); + // Validate JSON is an object (not primitive types like numbers/strings/booleans) + // Gemini CLI may output non-JSON-RPC data (e.g., numeric IDs) that would break protocol + if (typeof parsed !== 'object' || parsed === null) { + logger.debug('[ACP] Ignoring non-object JSON from stdout', { line }); + return; + } + message = parsed as JsonRpcRequest | JsonRpcResponse | JsonRpcNotification; } catch (error) { const protocolError = new Error('Failed to parse JSON-RPC from ACP agent'); this.protocolError = protocolError; @@ -228,4 +281,63 @@ export class AcpStdioTransport { } this.pending.clear(); } + + private parseStderrError(text: string): void { + if (!this.stderrErrorHandler) { + return; + } + + const lowerText = text.toLowerCase(); + + // Rate limit errors (429) + if (lowerText.includes('status 429') || lowerText.includes('ratelimitexceeded') || lowerText.includes('rate limit')) { + this.stderrErrorHandler({ + type: 'rate_limit', + message: 'Rate limit exceeded. Please wait before sending more requests.', + raw: text + }); + return; + } + + // Model not found errors (404) + if (lowerText.includes('status 404') || lowerText.includes('model not found') || lowerText.includes('not_found')) { + this.stderrErrorHandler({ + type: 'model_not_found', + message: 'Model not found. Available models: gemini-2.5-pro, gemini-2.5-flash, gemini-2.0-flash', + raw: text + }); + return; + } + + // Authentication errors (401/403) + if (lowerText.includes('status 401') || lowerText.includes('status 403') || + lowerText.includes('unauthenticated') || lowerText.includes('permission denied') || + lowerText.includes('authentication')) { + this.stderrErrorHandler({ + type: 'authentication', + message: 'Authentication failed. Please check your credentials or run "gemini auth login".', + raw: text + }); + return; + } + + // Quota exceeded + if (lowerText.includes('quota') || lowerText.includes('resource exhausted') || lowerText.includes('resourceexhausted')) { + this.stderrErrorHandler({ + type: 'quota_exceeded', + message: 'API quota exceeded. Please check your billing or wait for quota reset.', + raw: text + }); + return; + } + + // Only report as unknown if it looks like an actual error + if (lowerText.includes('error') || lowerText.includes('failed') || lowerText.includes('exception')) { + this.stderrErrorHandler({ + type: 'unknown', + message: text, + raw: text + }); + } + } }