From b542229311b728968545b35deabc7c92696d2c1a Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Sun, 5 Apr 2026 17:58:55 +0800 Subject: [PATCH] Fix ACP endpoint scheme handling --- lib/runtime/gateway_acp_client.dart | 32 ++- lib/web/web_acp_client.dart | 226 +++++++++++++++++++++ test/runtime/gateway_acp_client_suite.dart | 64 +++--- test/web/web_acp_client_suite.dart | 147 ++++++++++++++ 4 files changed, 442 insertions(+), 27 deletions(-) create mode 100644 test/web/web_acp_client_suite.dart diff --git a/lib/runtime/gateway_acp_client.dart b/lib/runtime/gateway_acp_client.dart index 87a74e44..fcef6a4b 100644 --- a/lib/runtime/gateway_acp_client.dart +++ b/lib/runtime/gateway_acp_client.dart @@ -307,18 +307,46 @@ class GatewayAcpClient { Uri? endpointOverride, String authorizationOverride = '', }) async { + final resolvedEndpoint = endpointOverride ?? endpointResolver(); + final scheme = resolvedEndpoint?.scheme.trim().toLowerCase() ?? ''; + final canUseHttp = resolveAcpHttpRpcEndpoint(resolvedEndpoint) != null; + + if (scheme == 'http' || scheme == 'https') { + try { + return await _requestViaHttp( + request, + onNotification: onNotification, + endpointOverride: resolvedEndpoint, + authorizationOverride: authorizationOverride, + ); + } catch (error) { + if (error is GatewayAcpException) { + rethrow; + } + return _requestViaWebSocket( + request, + onNotification: onNotification, + endpointOverride: resolvedEndpoint, + authorizationOverride: authorizationOverride, + ); + } + } + try { return await _requestViaWebSocket( request, onNotification: onNotification, - endpointOverride: endpointOverride, + endpointOverride: resolvedEndpoint, authorizationOverride: authorizationOverride, ); } catch (_) { + if (!canUseHttp) { + rethrow; + } return _requestViaHttp( request, onNotification: onNotification, - endpointOverride: endpointOverride, + endpointOverride: resolvedEndpoint, authorizationOverride: authorizationOverride, ); } diff --git a/lib/web/web_acp_client.dart b/lib/web/web_acp_client.dart index fdb57c64..f29d8f67 100644 --- a/lib/web/web_acp_client.dart +++ b/lib/web/web_acp_client.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'dart:convert'; +import 'package:http/http.dart' as http; import 'package:web_socket_channel/web_socket_channel.dart'; import '../runtime/acp_endpoint_paths.dart'; @@ -101,6 +102,65 @@ class WebAcpClient { Duration timeout = defaultTimeoutInternal, }) async { final requestId = '${DateTime.now().microsecondsSinceEpoch}-$method'; + final scheme = endpoint.scheme.trim().toLowerCase(); + final canUseHttp = resolveHttpRpcEndpointInternal(endpoint) != null; + if (scheme == 'http' || scheme == 'https') { + try { + return await _requestViaHttp( + requestId: requestId, + endpoint: endpoint, + method: method, + params: params, + onNotification: onNotification, + timeout: timeout, + ); + } catch (error) { + if (error is WebAcpException) { + rethrow; + } + return _requestViaWebSocket( + requestId: requestId, + endpoint: endpoint, + method: method, + params: params, + onNotification: onNotification, + timeout: timeout, + ); + } + } + + try { + return await _requestViaWebSocket( + requestId: requestId, + endpoint: endpoint, + method: method, + params: params, + onNotification: onNotification, + timeout: timeout, + ); + } catch (_) { + if (!canUseHttp) { + rethrow; + } + return _requestViaHttp( + requestId: requestId, + endpoint: endpoint, + method: method, + params: params, + onNotification: onNotification, + timeout: timeout, + ); + } + } + + Future> _requestViaWebSocket({ + required String requestId, + required Uri endpoint, + required String method, + required Map params, + void Function(Map notification)? onNotification, + required Duration timeout, + }) async { final wsEndpoint = resolveWebSocketEndpointInternal(endpoint); if (wsEndpoint == null) { throw const WebAcpException( @@ -166,10 +226,176 @@ class WebAcpClient { } } + Future> _requestViaHttp({ + required String requestId, + required Uri endpoint, + required String method, + required Map params, + void Function(Map notification)? onNotification, + required Duration timeout, + }) async { + final httpEndpoint = resolveHttpRpcEndpointInternal(endpoint); + if (httpEndpoint == null) { + throw const WebAcpException( + 'Missing ACP HTTP endpoint', + code: 'ACP_HTTP_ENDPOINT_MISSING', + ); + } + + final response = await http + .post( + httpEndpoint, + headers: const { + 'content-type': 'application/json; charset=utf-8', + 'accept': 'text/event-stream, application/json', + }, + body: jsonEncode({ + 'jsonrpc': '2.0', + 'id': requestId, + 'method': method, + 'params': params, + }), + ) + .timeout(timeout); + final contentType = + response.headers['content-type']?.toLowerCase().trim() ?? ''; + if (response.statusCode < 200 || response.statusCode >= 300) { + throw WebAcpException( + _describeHttpError( + statusCode: response.statusCode, + contentType: contentType, + body: response.body, + ), + code: 'ACP_HTTP_${response.statusCode}', + details: { + 'statusCode': response.statusCode, + 'contentType': contentType, + }, + ); + } + if (contentType.contains('text/event-stream')) { + return _consumeSseRpcResponse( + body: response.body, + requestId: requestId, + onNotification: onNotification, + ); + } + final decoded = decodeMapInternal(response.body); + throwIfJsonRpcErrorInternal(decoded); + return decoded; + } + static Uri? resolveWebSocketEndpointInternal(Uri? endpoint) { return resolveAcpWebSocketEndpoint(endpoint); } + static Uri? resolveHttpRpcEndpointInternal(Uri? endpoint) { + return resolveAcpHttpRpcEndpoint(endpoint); + } + + String _describeHttpError({ + required int statusCode, + required String contentType, + required String body, + }) { + final base = 'ACP HTTP request failed ($statusCode)'; + final normalizedType = contentType.trim(); + if (normalizedType.isNotEmpty && + !_contentTypeLooksJsonOrSse(normalizedType)) { + return '$base · unexpected content type: $normalizedType'; + } + + final detail = _extractErrorDetail(body); + if (detail.isNotEmpty) { + return '$base · $detail'; + } + return base; + } + + bool _contentTypeLooksJsonOrSse(String contentType) { + return contentType.contains('application/json') || + contentType.contains('application/problem+json') || + contentType.contains('text/json') || + contentType.contains('text/event-stream'); + } + + String _extractErrorDetail(String body) { + final trimmed = body.trim(); + if (trimmed.isEmpty) { + return ''; + } + try { + final decoded = decodeMapInternal(trimmed); + final error = asMapInternal(decoded['error']); + return (stringValueInternal(error['message']) ?? + stringValueInternal(decoded['message']) ?? + stringValueInternal(decoded['detail']) ?? + '') + .trim(); + } on FormatException { + // Fall through to textual snippet extraction below. + } + + final singleLine = trimmed.replaceAll(RegExp(r'\s+'), ' '); + if (singleLine.isEmpty) { + return ''; + } + return singleLine.length <= 160 + ? singleLine + : '${singleLine.substring(0, 157)}...'; + } + + Future> _consumeSseRpcResponse({ + required String body, + required String requestId, + void Function(Map notification)? onNotification, + }) async { + final eventLines = []; + Map? responseEnvelope; + + void consumeEventPayload(String payload) { + final trimmed = payload.trim(); + if (trimmed.isEmpty || trimmed == '[DONE]') { + return; + } + final json = decodeMapInternal(trimmed); + if (stringValueInternal(json['id']) == requestId && + (json.containsKey('result') || json.containsKey('error'))) { + responseEnvelope = json; + return; + } + if ((stringValueInternal(json['method']) ?? '').isNotEmpty && + onNotification != null) { + onNotification(json); + } + } + + for (final line in const LineSplitter().convert(body)) { + if (line.isEmpty) { + if (eventLines.isNotEmpty) { + consumeEventPayload(eventLines.join('\n')); + eventLines.clear(); + } + continue; + } + if (line.startsWith('data:')) { + eventLines.add(line.substring(5).trimLeft()); + } + } + + if (eventLines.isNotEmpty) { + consumeEventPayload(eventLines.join('\n')); + } + if (responseEnvelope == null) { + throw const WebAcpException( + 'ACP SSE ended without JSON-RPC response', + code: 'ACP_SSE_NO_RESULT', + ); + } + throwIfJsonRpcErrorInternal(responseEnvelope!); + return responseEnvelope!; + } + void throwIfJsonRpcErrorInternal(Map response) { final error = asMapInternal(response['error']); if (error.isEmpty) { diff --git a/test/runtime/gateway_acp_client_suite.dart b/test/runtime/gateway_acp_client_suite.dart index 70dd38d2..0813623c 100644 --- a/test/runtime/gateway_acp_client_suite.dart +++ b/test/runtime/gateway_acp_client_suite.dart @@ -11,7 +11,39 @@ import 'package:xworkmate/runtime/runtime_models.dart'; void main() { group('GatewayAcpClient', () { - test('loads ACP capabilities over websocket when available', () async { + 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 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); + }); + + test('preserves prefixed websocket ACP endpoints', () async { + final server = await _AcpFakeServer.start(pathPrefix: '/codex'); + addTearDown(server.close); + + final client = GatewayAcpClient( + endpointResolver: () => server.baseHttpUri.replace(scheme: 'ws'), + ); + + final capabilities = await client.loadCapabilities(forceRefresh: true); + + expect(capabilities.singleAgent, isTrue); + expect(server.rpcMethods, contains('acp.capabilities')); + }); + + test('prefers HTTP RPC when http endpoint is provided', () async { final server = await _AcpFakeServer.start(); addTearDown(server.close); @@ -22,23 +54,8 @@ void main() { 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')); - }); - - test('preserves prefixed websocket ACP endpoints', () async { - final server = await _AcpFakeServer.start(pathPrefix: '/codex'); - addTearDown(server.close); - - final client = GatewayAcpClient( - endpointResolver: () => server.baseHttpUri, - ); - - final capabilities = await client.loadCapabilities(forceRefresh: true); - - expect(capabilities.singleAgent, isTrue); - expect(server.rpcMethods, contains('acp.capabilities')); + expect(server.lastHttpRequestPath, '/acp/rpc'); + expect(server.lastWebSocketRequestPath, isNull); }); test('falls back to HTTP+SSE when websocket is unavailable', () async { @@ -107,7 +124,7 @@ void main() { addTearDown(server.close); final client = GatewayAcpClient( - endpointResolver: () => server.baseHttpUri, + endpointResolver: () => server.baseHttpUri.replace(scheme: 'ws'), authorizationResolver: (_) async => 'Bearer ws-secret', ); @@ -142,7 +159,7 @@ void main() { addTearDown(server.close); final client = GatewayAcpClient( - endpointResolver: () => server.baseHttpUri, + endpointResolver: () => server.baseHttpUri.replace(scheme: 'ws'), ); final capabilities = await client.loadCapabilities(forceRefresh: true); @@ -151,11 +168,8 @@ void main() { expect(server.lastWebSocketRequestPath, '/opencode/acp'); }); - test('preserves hosted ACP base path for HTTP fallback requests', () async { - final server = await _AcpFakeServer.start( - disableWebSocket: true, - pathPrefix: '/opencode', - ); + test('preserves hosted ACP base path for HTTP requests', () async { + final server = await _AcpFakeServer.start(pathPrefix: '/opencode'); addTearDown(server.close); final client = GatewayAcpClient( diff --git a/test/web/web_acp_client_suite.dart b/test/web/web_acp_client_suite.dart new file mode 100644 index 00000000..c8c8fdcb --- /dev/null +++ b/test/web/web_acp_client_suite.dart @@ -0,0 +1,147 @@ +@TestOn('vm') +library; + +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:xworkmate/runtime/runtime_models.dart'; +import 'package:xworkmate/web/web_acp_client.dart'; + +void main() { + group('WebAcpClient', () { + test('uses websocket when ws endpoint is provided', () async { + final server = await _WebAcpFakeServer.start(); + addTearDown(server.close); + + const client = WebAcpClient(); + final capabilities = await client.loadCapabilities( + endpoint: server.baseHttpUri.replace(scheme: 'ws'), + ); + + expect(capabilities.providers, contains(SingleAgentProvider.codex)); + expect(server.lastWebSocketRequestPath, '/acp'); + expect(server.lastHttpRequestPath, isNull); + }); + + test('uses HTTP RPC when http endpoint is provided', () async { + final server = await _WebAcpFakeServer.start(); + addTearDown(server.close); + + const client = WebAcpClient(); + final capabilities = await client.loadCapabilities( + endpoint: server.baseHttpUri, + ); + + expect(capabilities.providers, contains(SingleAgentProvider.codex)); + expect(server.lastHttpRequestPath, '/acp/rpc'); + expect(server.lastWebSocketRequestPath, isNull); + }); + + test('preserves prefixed HTTP RPC paths for hosted bases', () async { + final server = await _WebAcpFakeServer.start(pathPrefix: '/codex'); + addTearDown(server.close); + + const client = WebAcpClient(); + final capabilities = await client.loadCapabilities( + endpoint: server.baseHttpUri, + ); + + expect(capabilities.providers, contains(SingleAgentProvider.codex)); + expect(server.lastHttpRequestPath, '/codex/acp/rpc'); + }); + }); +} + +class _WebAcpFakeServer { + _WebAcpFakeServer._(this._server, {required this.pathPrefix}); + + final HttpServer _server; + final String pathPrefix; + String? lastWebSocketRequestPath; + String? lastHttpRequestPath; + + Uri get baseHttpUri => + Uri.parse('http://127.0.0.1:${_server.port}$pathPrefix'); + + static Future<_WebAcpFakeServer> start({String pathPrefix = ''}) async { + final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + final fake = _WebAcpFakeServer._( + server, + pathPrefix: _normalizePathPrefix(pathPrefix), + ); + unawaited(fake._listen()); + return fake; + } + + Future close() async { + await _server.close(force: true); + } + + Future _listen() async { + await for (final request in _server) { + if (WebSocketTransformer.isUpgradeRequest(request) && + request.uri.path == '$pathPrefix/acp') { + lastWebSocketRequestPath = request.uri.path; + final socket = await WebSocketTransformer.upgrade(request); + socket.listen((raw) { + socket.add( + jsonEncode({ + 'jsonrpc': '2.0', + 'id': _decodeId(raw), + 'result': { + 'singleAgent': true, + 'multiAgent': true, + 'providers': const ['codex'], + }, + }), + ); + }); + continue; + } + + if (request.uri.path == '$pathPrefix/acp/rpc' && + request.method == 'POST') { + lastHttpRequestPath = request.uri.path; + request.response.statusCode = HttpStatus.ok; + request.response.headers.set( + HttpHeaders.contentTypeHeader, + 'text/event-stream', + ); + final rawBody = await utf8.decoder.bind(request).join(); + final envelope = { + 'jsonrpc': '2.0', + 'id': _decodeId(rawBody), + 'result': { + 'singleAgent': true, + 'multiAgent': true, + 'providers': const ['codex'], + }, + }; + request.response.write('data: ${jsonEncode(envelope)}\n\n'); + await request.response.close(); + continue; + } + + request.response.statusCode = HttpStatus.notFound; + await request.response.close(); + } + } + + static String _decodeId(Object raw) { + final decoded = jsonDecode(raw.toString()); + if (decoded is Map && decoded['id'] != null) { + return decoded['id'].toString(); + } + return 'unknown'; + } + + static String _normalizePathPrefix(String raw) { + final trimmed = raw.trim(); + if (trimmed.isEmpty || trimmed == '/') { + return ''; + } + return trimmed.startsWith('/') ? trimmed : '/$trimmed'; + } +}