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

fix: harden ACP websocket fallback for openclaw gateway

Browse Source
This commit is contained in:
Haitao Pan 2026-04-08 20:07:47 +08:00
parent 4e19ba3ded
commit b09d56587a
2 changed files with 168 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
lib/runtime/gateway_acp_client.dart
View File

@ -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<Map<String, dynamic>> _requestViaWebSocketEndpoint(
_GatewayAcpRpcRequest request, {
required Uri endpoint,
required void Function(Map<String, dynamic>) 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: <String, dynamic>{
...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 <String, dynamic>{};
}
Uri? _resolveWebSocketRpcEndpoint([Uri? endpointOverride]) {
return resolveAcpWebSocketEndpoint(endpointOverride ?? endpointResolver());
List<Uri> _resolveWebSocketRpcEndpoints([Uri? endpointOverride]) {
final endpoint = endpointOverride ?? endpointResolver();
if (endpoint == null || endpoint.host.trim().isEmpty) {
return const <Uri>[];
}
final candidates = <Uri>[];
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]) {

92
test/runtime/gateway_acp_client_suite.dart
View File

@ -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<GatewayAcpException>()
.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<String> rpcMethods = <String>[];
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<void> _listen() async {
await for (final request in _server) {
final wsPaths = <String>[
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(