From 093db7dfc510849c001c3cc07b0d71ef3bc82df2 Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Sat, 9 May 2026 15:12:51 +0800 Subject: [PATCH] fix: stabilize ACP gateway connect failures --- ...pp_controller_desktop_runtime_helpers.dart | 30 ++++ ...app_controller_desktop_thread_actions.dart | 38 +++++- ...rnal_code_agent_acp_desktop_transport.dart | 19 +++ lib/runtime/gateway_acp_client.dart | 128 ++++++++++++++++-- .../assistant_connection_state_test.dart | 39 ++++++ .../assistant_execution_target_test.dart | 66 ++++++++- .../runtime/gateway_acp_client_auth_test.dart | 105 ++++++++++++++ 7 files changed, 408 insertions(+), 17 deletions(-) diff --git a/lib/app/app_controller_desktop_runtime_helpers.dart b/lib/app/app_controller_desktop_runtime_helpers.dart index 6169d401..cff64550 100644 --- a/lib/app/app_controller_desktop_runtime_helpers.dart +++ b/lib/app/app_controller_desktop_runtime_helpers.dart @@ -237,6 +237,19 @@ extension AppControllerDesktopRuntimeHelpers on AppController { final recoverableTransportCode = recoverableAcpHttpTransportCodeInternal( error, ); + final unconfirmedConnectCode = unconfirmedAcpHttpConnectCodeInternal(error); + if (unconfirmedConnectCode == gatewayAcpHttpConnectTimeoutCode) { + return appText( + 'Bridge 连接超时，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_TIMEOUT', + 'Bridge connection timed out; this request was not confirmed and can be retried. Error code: ACP_HTTP_CONNECT_TIMEOUT', + ); + } + if (unconfirmedConnectCode == gatewayAcpHttpConnectFailedCode) { + return appText( + 'Bridge 连接失败，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_FAILED', + 'Bridge connection failed; this request was not confirmed and can be retried. Error code: ACP_HTTP_CONNECT_FAILED', + ); + } if (recoverableTransportCode == 'ACP_HTTP_CONNECTION_CLOSED') { return appText( 'Bridge 响应读取中断；当前对话已保留，下一次发送会继续同一会话。错误码：ACP_HTTP_CONNECTION_CLOSED', @@ -362,6 +375,23 @@ extension AppControllerDesktopRuntimeHelpers on AppController { return null; } + String? unconfirmedAcpHttpConnectCodeInternal(Object error) { + final raw = error.toString().trim(); + final primaryCode = gatewayExecutionPrimaryCodeInternal(error); + final detailCode = gatewayExecutionDetailCodeInternal(error); + if (primaryCode == gatewayAcpHttpConnectTimeoutCode || + detailCode == gatewayAcpHttpConnectTimeoutCode || + raw.contains(gatewayAcpHttpConnectTimeoutCode)) { + return gatewayAcpHttpConnectTimeoutCode; + } + if (primaryCode == gatewayAcpHttpConnectFailedCode || + detailCode == gatewayAcpHttpConnectFailedCode || + raw.contains(gatewayAcpHttpConnectFailedCode)) { + return gatewayAcpHttpConnectFailedCode; + } + return null; + } + String formatAiGatewayHttpErrorInternal(int statusCode, String detail) { final base = switch (statusCode) { 400 => appText( diff --git a/lib/app/app_controller_desktop_thread_actions.dart b/lib/app/app_controller_desktop_thread_actions.dart index 1513441d..e948b7d3 100644 --- a/lib/app/app_controller_desktop_thread_actions.dart +++ b/lib/app/app_controller_desktop_thread_actions.dart @@ -330,7 +330,7 @@ extension AppControllerDesktopThreadActions on AppController { final sessionKey = normalizedAssistantSessionKeyInternal( currentSessionKey, ); - final resumeSession = hasCommittedUserTurnForGatewaySessionInternal( + final resumeSession = shouldResumeGatewaySessionForNextSendInternal( sessionKey, ); final lifecycleStatus = taskThreadForSessionInternal( @@ -339,7 +339,13 @@ extension AppControllerDesktopThreadActions on AppController { final lastResultCode = taskThreadForSessionInternal( sessionKey, )?.lifecycleState.lastResultCode?.trim().toLowerCase(); - final runStatus = resumeSession && lifecycleStatus == 'interrupted' + final continuableTransportResult = + lastResultCode == 'acp_http_connection_closed' || + lastResultCode == + gatewayAcpHttpHandshakeInterruptedCode.toLowerCase(); + final runStatus = + resumeSession && + (lifecycleStatus == 'interrupted' || continuableTransportResult) ? 'continuing' : resumeSession && lastResultCode == 'error' ? 'retrying' @@ -375,6 +381,13 @@ extension AppControllerDesktopThreadActions on AppController { try { final dispatch = await codeAgentNodeOrchestratorInternal .buildGatewayDispatch(buildCodeAgentNodeStateInternal()); + upsertTaskThreadInternal( + sessionKey, + lifecycleStatus: runStatus, + lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + lastResultCode: runStatus, + updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + ); final result = await goTaskServiceClientInternal.executeTask( GoTaskServiceRequest( sessionId: sessionKey, @@ -490,15 +503,20 @@ extension AppControllerDesktopThreadActions on AppController { clearAiGatewayStreamingTextInternal(sessionKey); final recoverableTransportCode = recoverableAcpHttpTransportCodeInternal(error); + final unconfirmedConnectCode = unconfirmedAcpHttpConnectCodeInternal( + error, + ); final recoverableTransportInterrupted = recoverableTransportCode != null; + final visibleResultCode = + unconfirmedConnectCode ?? recoverableTransportCode; upsertTaskThreadInternal( sessionKey, lifecycleStatus: recoverableTransportInterrupted ? 'interrupted' : 'ready', lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - lastResultCode: recoverableTransportCode ?? 'error', + lastResultCode: visibleResultCode ?? 'error', lastArtifactSyncStatus: recoverableTransportInterrupted ? 'interrupted' : null, @@ -537,6 +555,20 @@ extension AppControllerDesktopThreadActions on AppController { }); } + bool shouldResumeGatewaySessionForNextSendInternal(String sessionKey) { + final normalizedSessionKey = normalizedAssistantSessionKeyInternal( + sessionKey, + ); + if (!hasCommittedUserTurnForGatewaySessionInternal(normalizedSessionKey)) { + return false; + } + final lastResultCode = taskThreadForSessionInternal( + normalizedSessionKey, + )?.lifecycleState.lastResultCode?.trim().toUpperCase(); + return lastResultCode != gatewayAcpHttpConnectTimeoutCode && + lastResultCode != gatewayAcpHttpConnectFailedCode; + } + Future abortRun() async { if (multiAgentRunPendingInternal) { final sessionKey = normalizedAssistantSessionKeyInternal( diff --git a/lib/runtime/external_code_agent_acp_desktop_transport.dart b/lib/runtime/external_code_agent_acp_desktop_transport.dart index 15a177ab..bf44fc8a 100644 --- a/lib/runtime/external_code_agent_acp_desktop_transport.dart +++ b/lib/runtime/external_code_agent_acp_desktop_transport.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:io'; import 'package:flutter/foundation.dart'; @@ -134,6 +135,17 @@ class ExternalCodeAgentAcpDesktopTransport ); } on GatewayAcpException { rethrow; + } on SocketException catch (error) { + final timeout = _socketExceptionLooksLikeConnectTimeout(error); + throw GatewayAcpException( + timeout + ? 'ACP HTTP connection timed out before the request was confirmed' + : 'ACP HTTP connection failed before the request was confirmed', + code: timeout + ? gatewayAcpHttpConnectTimeoutCode + : gatewayAcpHttpConnectFailedCode, + details: {'originalError': error.toString()}, + ); } catch (error) { throw GatewayAcpException( error.toString(), @@ -304,4 +316,11 @@ class ExternalCodeAgentAcpDesktopTransport } return [defaultTarget]; } + + bool _socketExceptionLooksLikeConnectTimeout(SocketException error) { + final lowered = error.toString().toLowerCase(); + return lowered.contains('connection timed out') || + lowered.contains('timed out') || + lowered.contains('timeout'); + } } diff --git a/lib/runtime/gateway_acp_client.dart b/lib/runtime/gateway_acp_client.dart index 8b799676..7f874e36 100644 --- a/lib/runtime/gateway_acp_client.dart +++ b/lib/runtime/gateway_acp_client.dart @@ -6,8 +6,12 @@ import 'acp_endpoint_paths.dart'; import 'runtime_models.dart'; const int gatewayAcpHttpHandshakeInterruptedRetryCount = 5; +const int gatewayAcpHttpConnectFailureRetryCount = 2; +const Duration gatewayAcpHttpConnectTimeout = Duration(seconds: 12); const String gatewayAcpHttpHandshakeInterruptedCode = 'ACP_HTTP_HANDSHAKE_INTERRUPTED'; +const String gatewayAcpHttpConnectTimeoutCode = 'ACP_HTTP_CONNECT_TIMEOUT'; +const String gatewayAcpHttpConnectFailedCode = 'ACP_HTTP_CONNECT_FAILED'; class GatewayAcpException implements Exception { const GatewayAcpException( @@ -77,6 +81,13 @@ class _GatewayAcpSessionUpdate { final Map payload; } +enum _GatewayAcpHttpRequestPhase { + connect, + write, + waitingForResponse, + bodyRead, +} + class GatewayAcpMultiAgentRequest { const GatewayAcpMultiAgentRequest({ required this.sessionId, @@ -564,7 +575,7 @@ class GatewayAcpClient { ); } - GatewayAcpException? lastHandshakeError; + GatewayAcpException? lastRetryableError; for ( var attempt = 0; attempt <= gatewayAcpHttpHandshakeInterruptedRetryCount; @@ -579,21 +590,39 @@ class GatewayAcpClient { retryAttempt: attempt, ); } on GatewayAcpException catch (error) { - if (error.code != gatewayAcpHttpHandshakeInterruptedCode || - attempt == gatewayAcpHttpHandshakeInterruptedRetryCount) { + final retryLimit = _httpRetryCountForError(error); + if (retryLimit == null || attempt >= retryLimit) { rethrow; } - lastHandshakeError = error; - await Future.delayed(Duration(milliseconds: 50 * (attempt + 1))); + lastRetryableError = error; + await Future.delayed(_httpRetryDelayFor(error, attempt)); } } - throw lastHandshakeError ?? + throw lastRetryableError ?? const GatewayAcpException( 'ACP HTTP handshake was interrupted before the response started', code: gatewayAcpHttpHandshakeInterruptedCode, ); } + int? _httpRetryCountForError(GatewayAcpException error) { + return switch (error.code) { + gatewayAcpHttpHandshakeInterruptedCode => + gatewayAcpHttpHandshakeInterruptedRetryCount, + gatewayAcpHttpConnectTimeoutCode || + gatewayAcpHttpConnectFailedCode => gatewayAcpHttpConnectFailureRetryCount, + _ => null, + }; + } + + Duration _httpRetryDelayFor(GatewayAcpException error, int attempt) { + if (error.code == gatewayAcpHttpConnectTimeoutCode || + error.code == gatewayAcpHttpConnectFailedCode) { + return Duration(milliseconds: 200 * (1 << attempt)); + } + return Duration(milliseconds: 50 * (attempt + 1)); + } + Future> _requestViaHttpAttempt( _GatewayAcpRpcRequest request, { required Uri endpoint, @@ -601,12 +630,22 @@ class GatewayAcpClient { required String authorizationOverride, required int retryAttempt, }) async { - final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); + final client = HttpClient() + ..connectionTimeout = gatewayAcpHttpConnectTimeout; var statusCode = 0; var contentType = ''; var bodyRead = false; + var phase = _GatewayAcpHttpRequestPhase.connect; try { - final httpRequest = await client.postUrl(endpoint); + final authorization = await _resolveAuthorizationHeader( + endpoint, + authorizationOverride: authorizationOverride, + ); + phase = _GatewayAcpHttpRequestPhase.connect; + final httpRequest = await client + .postUrl(endpoint) + .timeout(gatewayAcpHttpConnectTimeout); + phase = _GatewayAcpHttpRequestPhase.write; httpRequest.headers.set( HttpHeaders.contentTypeHeader, 'application/json; charset=utf-8', @@ -615,10 +654,6 @@ class GatewayAcpClient { HttpHeaders.acceptHeader, 'text/event-stream, application/json', ); - final authorization = await _resolveAuthorizationHeader( - endpoint, - authorizationOverride: authorizationOverride, - ); if (authorization.isNotEmpty) { httpRequest.headers.set(HttpHeaders.authorizationHeader, authorization); } @@ -632,6 +667,7 @@ class GatewayAcpClient { }), ), ); + phase = _GatewayAcpHttpRequestPhase.waitingForResponse; final response = await httpRequest.close().timeout( gatewayAcpHttpResponseTimeoutFor( endpoint, @@ -646,6 +682,7 @@ class GatewayAcpClient { .value(HttpHeaders.contentTypeHeader) ?.toLowerCase() ?? ''; + phase = _GatewayAcpHttpRequestPhase.bodyRead; if (response.statusCode < 200 || response.statusCode >= 300) { final body = await response.transform(utf8.decoder).join(); bodyRead = body.isNotEmpty; @@ -700,6 +737,20 @@ class GatewayAcpClient { }; } on GatewayAcpException { rethrow; + } on TimeoutException catch (error) { + if (phase == _GatewayAcpHttpRequestPhase.connect) { + throw _connectException( + endpoint: endpoint, + statusCode: statusCode, + contentType: contentType, + bodyRead: bodyRead, + retryAttempt: retryAttempt, + phase: phase, + originalError: error, + timeout: true, + ); + } + rethrow; } on HandshakeException catch (error) { throw _handshakeInterruptedException( endpoint: endpoint, @@ -725,6 +776,20 @@ class GatewayAcpClient { originalError: error, ); } + if (phase == _GatewayAcpHttpRequestPhase.connect && + statusCode == 0 && + !bodyRead) { + throw _connectException( + endpoint: endpoint, + statusCode: statusCode, + contentType: contentType, + bodyRead: bodyRead, + retryAttempt: retryAttempt, + phase: phase, + originalError: error, + timeout: _looksLikeConnectTimeout(error.toString()), + ); + } rethrow; } on HttpException catch (error) { if (_looksLikeConnectionClosedBeforeResponse(error.toString())) { @@ -746,6 +811,38 @@ class GatewayAcpClient { } } + GatewayAcpException _connectException({ + required Uri endpoint, + required int statusCode, + required String contentType, + required bool bodyRead, + required int retryAttempt, + required _GatewayAcpHttpRequestPhase phase, + required Object originalError, + required bool timeout, + }) { + final code = timeout + ? gatewayAcpHttpConnectTimeoutCode + : gatewayAcpHttpConnectFailedCode; + final message = timeout + ? 'ACP HTTP connection timed out before the request was confirmed' + : 'ACP HTTP connection failed before the request was confirmed'; + return GatewayAcpException( + message, + code: code, + details: { + 'requestUrl': endpoint.toString(), + 'statusCode': statusCode, + 'contentType': contentType, + 'bodyRead': bodyRead, + 'phase': phase.name, + 'retryAttempt': retryAttempt, + 'maxRetryAttempts': gatewayAcpHttpConnectFailureRetryCount, + 'originalError': originalError.toString(), + }, + ); + } + GatewayAcpException _handshakeInterruptedException({ required Uri endpoint, required int statusCode, @@ -792,6 +889,13 @@ class GatewayAcpClient { lowered.contains('stream closed'); } + bool _looksLikeConnectTimeout(String raw) { + final lowered = raw.toLowerCase(); + return lowered.contains('connection timed out') || + lowered.contains('timed out') || + lowered.contains('timeout'); + } + String _describeHttpError({ required int statusCode, required String contentType, diff --git a/test/runtime/assistant_connection_state_test.dart b/test/runtime/assistant_connection_state_test.dart index cc03e6e5..ece1e8ba 100644 --- a/test/runtime/assistant_connection_state_test.dart +++ b/test/runtime/assistant_connection_state_test.dart @@ -208,6 +208,45 @@ void main() { }, ); + test( + 'labels ACP HTTP connect timeouts as unconfirmed retryable requests', + () async { + final controller = await _isolatedController(); + addTearDown(controller.dispose); + + final label = controller.gatewayExecutionErrorLabelInternal( + const GatewayAcpException( + 'SocketException: HTTP connection timed out after 0:00:08.000000, host: xworkmate-bridge.svc.plus, port: 443', + code: gatewayAcpHttpConnectTimeoutCode, + ), + target: AssistantExecutionTarget.gateway, + ); + + expect(label, 'Bridge 连接超时，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_TIMEOUT'); + expect(label, isNot(contains('SocketException'))); + expect(label, isNot(contains('0:00:08'))); + }, + ); + + test( + 'labels ACP HTTP connect failures as unconfirmed retryable requests', + () async { + final controller = await _isolatedController(); + addTearDown(controller.dispose); + + final label = controller.gatewayExecutionErrorLabelInternal( + const GatewayAcpException( + 'Connection refused', + code: gatewayAcpHttpConnectFailedCode, + ), + target: AssistantExecutionTarget.gateway, + ); + + expect(label, 'Bridge 连接失败，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_FAILED'); + expect(label, isNot(contains('Connection refused'))); + }, + ); + test( 'labels unavailable session continuation without starting a new flow', () async { diff --git a/test/runtime/assistant_execution_target_test.dart b/test/runtime/assistant_execution_target_test.dart index 06e6fbc5..b76ce164 100644 --- a/test/runtime/assistant_execution_target_test.dart +++ b/test/runtime/assistant_execution_target_test.dart @@ -726,6 +726,68 @@ void main() { }, ); + test( + 'sendChatMessage starts a new session after ACP HTTP connect timeout', + () async { + final fakeGoTaskService = _RecordingGoTaskServiceClient() + ..outcomes.add( + const GatewayAcpException( + 'ACP HTTP connection timed out before the request was confirmed', + code: gatewayAcpHttpConnectTimeoutCode, + ), + ) + ..outcomes.add( + const GoTaskServiceResult( + success: true, + message: 'retried from a confirmed new start', + turnId: 'turn-2', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + final controller = _connectedController(fakeGoTaskService); + addTearDown(controller.dispose); + + await controller.sessionsController.switchSession('session-1'); + + await controller.sendChatMessage('first turn'); + + expect(fakeGoTaskService.requests, hasLength(1)); + expect(fakeGoTaskService.requests.single.resumeSession, isFalse); + final failedThread = controller.taskThreadForSessionInternal( + 'session-1', + ); + expect(failedThread?.lifecycleState.status, 'ready'); + expect( + failedThread?.lifecycleState.lastResultCode, + gatewayAcpHttpConnectTimeoutCode, + ); + expect(failedThread?.lastArtifactSyncStatus, isNull); + expect( + controller.chatMessages.last.text, + 'Bridge 连接超时，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_TIMEOUT', + ); + + await controller.sendChatMessage('retry after unconfirmed connect'); + + expect(fakeGoTaskService.requests, hasLength(2)); + expect(fakeGoTaskService.requests.last.resumeSession, isFalse); + await _waitForLastChatMessageText( + controller, + 'retried from a confirmed new start', + ); + expect( + controller.chatMessages.last.text, + 'retried from a confirmed new start', + ); + final thread = controller.taskThreadForSessionInternal('session-1'); + expect(thread?.lifecycleState.status, 'ready'); + expect(thread?.lifecycleState.lastResultCode, 'success'); + }, + ); + test( 'sendChatMessage hides OpenClaw artifact guard text after an interrupted continuation', () async { @@ -1325,7 +1387,7 @@ Future _waitForLastChatMessageText( AppController controller, String expectedText, ) async { - final deadline = DateTime.now().add(const Duration(seconds: 2)); + final deadline = DateTime.now().add(const Duration(seconds: 15)); while (DateTime.now().isBefore(deadline)) { if (controller.chatMessages.isNotEmpty && controller.chatMessages.last.text == expectedText) { @@ -1541,7 +1603,7 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient { } Future waitForRequestCount(int count) async { - final deadline = DateTime.now().add(const Duration(seconds: 5)); + final deadline = DateTime.now().add(const Duration(seconds: 15)); while (requests.length < count && DateTime.now().isBefore(deadline)) { await Future.delayed(const Duration(milliseconds: 10)); } diff --git a/test/runtime/gateway_acp_client_auth_test.dart b/test/runtime/gateway_acp_client_auth_test.dart index 5cf78851..e181a95b 100644 --- a/test/runtime/gateway_acp_client_auth_test.dart +++ b/test/runtime/gateway_acp_client_auth_test.dart @@ -530,6 +530,91 @@ void main() { }, ); + test( + 'retries failed connect attempts before surfacing unconfirmed diagnostics', + () async { + final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + final port = server.port; + await server.close(); + final endpoint = Uri.parse('http://127.0.0.1:$port'); + final client = GatewayAcpClient(endpointResolver: () => endpoint); + + await expectLater( + client.request( + method: 'session.start', + params: const {}, + ), + throwsA( + isA() + .having( + (error) => error.code, + 'code', + gatewayAcpHttpConnectFailedCode, + ) + .having( + (error) => error.message, + 'message', + contains('before the request was confirmed'), + ) + .having( + (error) => error.details, + 'details', + allOf( + containsPair('requestUrl', '$endpoint/acp/rpc'), + containsPair( + 'maxRetryAttempts', + gatewayAcpHttpConnectFailureRetryCount, + ), + containsPair( + 'retryAttempt', + gatewayAcpHttpConnectFailureRetryCount, + ), + containsPair('phase', 'connect'), + ), + ), + ), + ); + }, + ); + + test( + 'desktop transport preserves socket timeout as unconfirmed ACP diagnostics', + () async { + final transport = ExternalCodeAgentAcpDesktopTransport( + client: _SocketThrowingGatewayAcpClient( + const SocketException( + 'HTTP connection timed out after 0:00:08.000000, host: xworkmate-bridge.svc.plus, port: 443', + ), + ), + endpointResolver: (_) => + Uri.parse('https://xworkmate-bridge.svc.plus'), + ); + + await expectLater( + transport.executeTask( + _taskRequest( + target: AssistantExecutionTarget.gateway, + provider: SingleAgentProvider.openclaw, + ), + onUpdate: (_) {}, + ), + throwsA( + isA() + .having( + (error) => error.code, + 'code', + gatewayAcpHttpConnectTimeoutCode, + ) + .having( + (error) => error.toString(), + 'diagnostic', + isNot(contains('EXTERNAL_ACP_GATEWAY_ERROR')), + ), + ), + ); + }, + ); + test('desktop bridge auth resolver skips unrelated endpoints', () async { final storeRoot = await Directory.systemTemp.createTemp( 'xworkmate-acp-auth-unrelated-', @@ -1437,6 +1522,26 @@ void main() { }); } +class _SocketThrowingGatewayAcpClient extends GatewayAcpClient { + _SocketThrowingGatewayAcpClient(this.error) + : super( + endpointResolver: () => Uri.parse('https://xworkmate-bridge.svc.plus'), + ); + + final SocketException error; + + @override + Future> request({ + required String method, + required Map params, + void Function(Map)? onNotification, + Uri? endpointOverride, + String authorizationOverride = '', + }) async { + throw error; + } +} + GoTaskServiceRequest _taskRequest({ required AssistantExecutionTarget target, required SingleAgentProvider provider,