From b09d56587af25cf371cbae7ac460f498d4b18cd5 Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Wed, 8 Apr 2026 20:07:47 +0800 Subject: [PATCH] fix: harden ACP websocket fallback for openclaw gateway --- lib/runtime/gateway_acp_client.dart | 97 ++++++++++++++++++++-- test/runtime/gateway_acp_client_suite.dart | 92 ++++++++++++++++---- 2 files changed, 168 insertions(+), 21 deletions(-) diff --git a/lib/runtime/gateway_acp_client.dart b/lib/runtime/gateway_acp_client.dart index d5c22b1e..10cf6f95 100644 --- a/lib/runtime/gateway_acp_client.dart +++ b/lib/runtime/gateway_acp_client.dart @@ -334,8 +334,11 @@ class GatewayAcpClient { endpointOverride: resolvedEndpoint, authorizationOverride: authorizationOverride, ); - } catch (_) { - rethrow; + } catch (wsError) { + throw _toPreferredHttpFailureAfterWebSocketFallback( + httpError: error, + webSocketError: wsError, + ); } } return _requestViaWebSocket( @@ -373,14 +376,44 @@ class GatewayAcpClient { Uri? endpointOverride, String authorizationOverride = '', }) async { - final endpoint = _resolveWebSocketRpcEndpoint(endpointOverride); - if (endpoint == null) { + final endpoints = _resolveWebSocketRpcEndpoints(endpointOverride); + if (endpoints.isEmpty) { throw const GatewayAcpException( 'Missing ACP endpoint', code: 'ACP_ENDPOINT_MISSING', ); } + Object? lastError; + for (var index = 0; index < endpoints.length; index += 1) { + final endpoint = endpoints[index]; + try { + return await _requestViaWebSocketEndpoint( + request, + endpoint: endpoint, + onNotification: onNotification, + authorizationOverride: authorizationOverride, + ); + } catch (error) { + lastError = error; + if (index == endpoints.length - 1 || + !_shouldTryNextWebSocketCandidate(error)) { + rethrow; + } + } + } + throw GatewayAcpException( + lastError?.toString() ?? 'ACP websocket request failed', + code: 'ACP_WS_RUNTIME_ERROR', + ); + } + + Future> _requestViaWebSocketEndpoint( + _GatewayAcpRpcRequest request, { + required Uri endpoint, + required void Function(Map) onNotification, + String authorizationOverride = '', + }) async { final authorization = await _resolveAuthorizationHeader( endpoint, authorizationOverride: authorizationOverride, @@ -575,6 +608,38 @@ class GatewayAcpClient { (contentType.isEmpty || contentType.contains('text/plain')); } + bool _shouldTryNextWebSocketCandidate(Object error) { + if (error is WebSocketException || + error is SocketException || + error is HandshakeException || + error is HttpException) { + return true; + } + if (error is! GatewayAcpException) { + return false; + } + return error.code == 'ACP_WS_EARLY_CLOSE' || + error.code == 'ACP_WS_RUNTIME_ERROR' || + error.code == 'ACP_WS_CONNECT_TIMEOUT'; + } + + GatewayAcpException _toPreferredHttpFailureAfterWebSocketFallback({ + required GatewayAcpException httpError, + required Object webSocketError, + }) { + final wsError = webSocketError is GatewayAcpException + ? webSocketError + : null; + return GatewayAcpException( + httpError.message, + code: httpError.code, + details: { + ...asMap(httpError.details), + if (wsError?.code != null) 'websocketFallbackCode': wsError!.code, + }, + ); + } + bool _contentTypeLooksJsonOrSse(String contentType) { return contentType.contains('application/json') || contentType.contains('application/problem+json') || @@ -814,8 +879,28 @@ class GatewayAcpClient { return const {}; } - Uri? _resolveWebSocketRpcEndpoint([Uri? endpointOverride]) { - return resolveAcpWebSocketEndpoint(endpointOverride ?? endpointResolver()); + List _resolveWebSocketRpcEndpoints([Uri? endpointOverride]) { + final endpoint = endpointOverride ?? endpointResolver(); + if (endpoint == null || endpoint.host.trim().isEmpty) { + return const []; + } + final candidates = []; + final derived = resolveAcpWebSocketEndpoint(endpoint); + if (derived != null) { + candidates.add(derived); + } + final scheme = switch (endpoint.scheme.trim().toLowerCase()) { + 'https' || 'wss' => 'wss', + _ => 'ws', + }; + final raw = endpoint.replace(scheme: scheme, query: null, fragment: null); + final duplicate = candidates.any( + (candidate) => candidate.toString() == raw.toString(), + ); + if (!duplicate) { + candidates.add(raw); + } + return candidates; } Uri? _resolveHttpRpcEndpoint([Uri? endpointOverride]) { diff --git a/test/runtime/gateway_acp_client_suite.dart b/test/runtime/gateway_acp_client_suite.dart index cad39010..03928d5c 100644 --- a/test/runtime/gateway_acp_client_suite.dart +++ b/test/runtime/gateway_acp_client_suite.dart @@ -11,23 +11,26 @@ import 'package:xworkmate/runtime/runtime_models.dart'; void main() { group('GatewayAcpClient', () { - test('loads ACP capabilities over websocket when ws endpoint is provided', () async { - final server = await _AcpFakeServer.start(); - addTearDown(server.close); + test( + 'loads ACP capabilities over websocket when ws endpoint is provided', + () async { + final server = await _AcpFakeServer.start(); + addTearDown(server.close); - final client = GatewayAcpClient( - endpointResolver: () => server.baseHttpUri.replace(scheme: 'ws'), - ); + final client = GatewayAcpClient( + endpointResolver: () => server.baseHttpUri.replace(scheme: 'ws'), + ); - final capabilities = await client.loadCapabilities(forceRefresh: true); + final capabilities = await client.loadCapabilities(forceRefresh: true); - expect(capabilities.singleAgent, isTrue); - expect(capabilities.multiAgent, isTrue); - expect(capabilities.providers, contains(SingleAgentProvider.codex)); - expect(server.rpcMethods, contains('acp.capabilities')); - expect(server.lastWebSocketRequestPath, '/acp'); - expect(server.lastHttpRequestPath, isNull); - }); + expect(capabilities.singleAgent, isTrue); + expect(capabilities.multiAgent, isTrue); + expect(capabilities.providers, contains(SingleAgentProvider.codex)); + expect(server.rpcMethods, contains('acp.capabilities')); + expect(server.lastWebSocketRequestPath, '/acp'); + expect(server.lastHttpRequestPath, isNull); + }, + ); test('preserves prefixed websocket ACP endpoints', () async { final server = await _AcpFakeServer.start(pathPrefix: '/codex'); @@ -138,6 +141,57 @@ void main() { }, ); + test( + 'keeps HTTP 404 as primary error when websocket fallback also fails', + () async { + final server = await _AcpFakeServer.start( + disableWebSocket: true, + respondWithPlainTextNotFound: true, + ); + addTearDown(server.close); + + final client = GatewayAcpClient( + endpointResolver: () => server.baseHttpUri, + ); + + await expectLater( + () => client.loadCapabilities(forceRefresh: true), + throwsA( + isA() + .having((error) => error.code, 'code', 'ACP_HTTP_404') + .having( + (error) => error.toString(), + 'message', + contains('ACP HTTP request failed (404)'), + ), + ), + ); + }, + ); + + test( + 'falls back to raw websocket path when derived ACP path is unavailable', + () async { + final server = await _AcpFakeServer.start( + respondWithPlainTextNotFound: true, + pathPrefix: '/opencode', + useRawWebSocketPathOnly: true, + ); + addTearDown(server.close); + + final client = GatewayAcpClient( + endpointResolver: () => server.baseHttpUri, + ); + + final capabilities = await client.loadCapabilities(forceRefresh: true); + + expect(capabilities.singleAgent, isTrue); + expect(server.lastHttpRequestPath, '/opencode/acp/rpc'); + expect(server.lastWebSocketRequestPath, '/opencode'); + expect(server.rpcMethods, contains('acp.capabilities')); + }, + ); + test( 'forwards ACP authorization resolver headers over websocket', () async { @@ -256,6 +310,7 @@ class _AcpFakeServer { required this.disableWebSocket, required this.respondWithHtmlError, required this.respondWithPlainTextNotFound, + required this.useRawWebSocketPathOnly, required this.pathPrefix, }); @@ -263,6 +318,7 @@ class _AcpFakeServer { final bool disableWebSocket; final bool respondWithHtmlError; final bool respondWithPlainTextNotFound; + final bool useRawWebSocketPathOnly; final String pathPrefix; final List rpcMethods = []; String? lastWebSocketAuthorization; @@ -277,6 +333,7 @@ class _AcpFakeServer { bool disableWebSocket = false, bool respondWithHtmlError = false, bool respondWithPlainTextNotFound = false, + bool useRawWebSocketPathOnly = false, String pathPrefix = '', }) async { final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); @@ -285,6 +342,7 @@ class _AcpFakeServer { disableWebSocket: disableWebSocket, respondWithHtmlError: respondWithHtmlError, respondWithPlainTextNotFound: respondWithPlainTextNotFound, + useRawWebSocketPathOnly: useRawWebSocketPathOnly, pathPrefix: _normalizePathPrefix(pathPrefix), ); unawaited(fake._listen()); @@ -297,8 +355,12 @@ class _AcpFakeServer { Future _listen() async { await for (final request in _server) { + final wsPaths = [ + if (!useRawWebSocketPathOnly) '$pathPrefix/acp', + if (useRawWebSocketPathOnly) (pathPrefix.isEmpty ? '/' : pathPrefix), + ]; if (!disableWebSocket && - request.uri.path == '$pathPrefix/acp' && + wsPaths.contains(request.uri.path) && WebSocketTransformer.isUpgradeRequest(request)) { lastWebSocketRequestPath = request.uri.path; lastWebSocketAuthorization = request.headers.value(