wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-08-05 06:24:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: improve ACP stability and UX with initialization retry and error handling

Browse Source 
- 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
This commit is contained in:
weishu
2026-01-22 11:53:17 +08:00
parent ca63984f7e
commit e36759115c
2 changed files with 203 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/agent/backends/acp/AcpSdkBackend.ts
+87 -16
View File
@@ -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<string, PendingPermission>();
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<string, string> }) {}
@@ -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<void> {
if (!this.isProcessingMessage) {
return;
}
return new Promise<void>((resolve) => {
this.responseCompleteResolvers.push(resolve);
});
}
async disconnect(): Promise<void> {
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();
}
}
}
cli/src/agent/backends/acp/AcpStdioTransport.ts
+116 -4
View File
@@ -28,6 +28,14 @@ interface JsonRpcResponse {
type RequestHandler = (params: unknown, requestId: string | number | null) => Promise<unknown>;
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<string | number, {
@@ -36,6 +44,7 @@ export class AcpStdioTransport {
}>();
private readonly requestHandlers = new Map<string, RequestHandler>();
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<unknown> {
/** 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<unknown> {
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<unknown>((resolve, reject) => {
this.pending.set(id, { resolve, reject });
this.writePayload(payload);
});
}
return new Promise<unknown>((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
});
}
}
}