ai-workspace-lab/xworkmate-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix ACP endpoint scheme handling

Browse Source
This commit is contained in:
Haitao Pan 2026-04-05 17:58:55 +08:00
parent c1e96a7f9b
commit b542229311
4 changed files with 442 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
lib/runtime/gateway_acp_client.dart
View File

@ -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,
);
}

226
lib/web/web_acp_client.dart
View File

@ -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<Map<String, dynamic>> _requestViaWebSocket({
required String requestId,
required Uri endpoint,
required String method,
required Map<String, dynamic> params,
void Function(Map<String, dynamic> notification)? onNotification,
required Duration timeout,
}) async {
final wsEndpoint = resolveWebSocketEndpointInternal(endpoint);
if (wsEndpoint == null) {
throw const WebAcpException(
@ -166,10 +226,176 @@ class WebAcpClient {
}
}
Future<Map<String, dynamic>> _requestViaHttp({
required String requestId,
required Uri endpoint,
required String method,
required Map<String, dynamic> params,
void Function(Map<String, dynamic> 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 <String, String>{
'content-type': 'application/json; charset=utf-8',
'accept': 'text/event-stream, application/json',
},
body: jsonEncode(<String, dynamic>{
'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: <String, dynamic>{
'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<Map<String, dynamic>> _consumeSseRpcResponse({
required String body,
required String requestId,
void Function(Map<String, dynamic> notification)? onNotification,
}) async {
final eventLines = <String>[];
Map<String, dynamic>? 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<String, dynamic> response) {
final error = asMapInternal(response['error']);
if (error.isEmpty) {

64
test/runtime/gateway_acp_client_suite.dart
View File

@ -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(

147
test/web/web_acp_client_suite.dart Normal file
View File

@ -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<void> close() async {
await _server.close(force: true);
}
Future<void> _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(<String, dynamic>{
'jsonrpc': '2.0',
'id': _decodeId(raw),
'result': <String, dynamic>{
'singleAgent': true,
'multiAgent': true,
'providers': const <String>['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 = <String, dynamic>{
'jsonrpc': '2.0',
'id': _decodeId(rawBody),
'result': <String, dynamic>{
'singleAgent': true,
'multiAgent': true,
'providers': const <String>['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';
}
}