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

fix: route openclaw tasks through bridge rpc

Browse Source
This commit is contained in:
Haitao Pan 2026-05-07 08:25:52 +08:00
parent a66fe27d30
commit 3ee9404aa2
4 changed files with 167 additions and 184 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
lib/app/app_controller_desktop_runtime_helpers.dart
View File

@ -867,9 +867,6 @@ extension AppControllerDesktopRuntimeHelpers on AppController {
if (bridgeEndpoint == null) {
return null;
}
if (_usesOpenClawTaskSubmitEndpointInternal(request)) {
return bridgeEndpoint.replace(path: '/gateway/openclaw');
}
return bridgeEndpoint;
}
@ -970,14 +967,6 @@ extension AppControllerDesktopRuntimeHelpers on AppController {
) => kGatewayRemoteProfileIndex;
}
bool _usesOpenClawTaskSubmitEndpointInternal(GoTaskServiceRequest request) {
if (request.isMultiAgentRequest || !request.target.isGateway) {
return false;
}
return normalizeSingleAgentProviderId(request.provider.providerId) ==
kCanonicalGatewayProviderId;
}
String _normalizeAuthorizationHeaderInternal(String raw) {
final trimmed = raw.trim();
if (trimmed.isEmpty) {

47
lib/runtime/gateway_acp_client.dart
View File

@ -566,7 +566,7 @@ class GatewayAcpClient {
);
httpRequest.headers.set(
HttpHeaders.acceptHeader,
_httpAcceptHeaderFor(endpoint, request.method),
'text/event-stream, application/json',
);
final authorization = await _resolveAuthorizationHeader(
endpoint,
@ -647,10 +647,10 @@ class GatewayAcpClient {
} on GatewayAcpException {
rethrow;
} on HttpException catch (error) {
if (_looksLikeConnectionClosedWhileReceivingData(error.toString())) {
if (_looksLikeConnectionClosedBeforeResponse(error.toString())) {
throw GatewayAcpException(
'ACP HTTP response stream closed before the body finished arriving',
code: 'ACP_HTTP_STREAM_CLOSED',
'ACP HTTP connection closed before the response finished arriving',
code: 'ACP_HTTP_CONNECTION_CLOSED',
details: <String, dynamic>{
'requestUrl': endpoint.toString(),
'statusCode': statusCode,
@ -666,9 +666,10 @@ class GatewayAcpClient {
}
}
bool _looksLikeConnectionClosedWhileReceivingData(String raw) {
bool _looksLikeConnectionClosedBeforeResponse(String raw) {
final lowered = raw.toLowerCase();
return lowered.contains('connection closed while receiving data') ||
return lowered.contains('connection closed before full header') ||
lowered.contains('connection closed while receiving data') ||
lowered.contains('connection terminated during body read') ||
lowered.contains('stream closed');
}
@ -1048,14 +1049,6 @@ class GatewayAcpClient {
Uri? _resolveHttpRpcEndpoint([Uri? endpointOverride, String method = '']) {
final endpoint = endpointOverride ?? endpointResolver();
if (_isOpenClawTaskSubmitEndpoint(endpoint) &&
_isOpenClawTaskSubmitMethod(method)) {
return endpoint?.replace(
path: '/gateway/openclaw',
query: null,
fragment: null,
);
}
return resolveAcpHttpRpcEndpoint(endpoint);
}
@ -1114,33 +1107,7 @@ class GatewayAcpClient {
}
}
bool _isOpenClawTaskSubmitEndpoint(Uri? endpoint) {
var path = endpoint?.path.trim() ?? '';
if (!path.startsWith('/')) {
path = '/$path';
}
path = path.replaceFirst(RegExp(r'/+$'), '');
return path == '/gateway/openclaw';
}
bool _isOpenClawTaskSubmitMethod(String method) {
final normalized = method.trim();
return normalized == 'session.start' || normalized == 'session.message';
}
String _httpAcceptHeaderFor(Uri endpoint, String method) {
if (_isOpenClawTaskSubmitEndpoint(endpoint) &&
_isOpenClawTaskSubmitMethod(method)) {
return 'application/json';
}
return 'text/event-stream, application/json';
}
Duration gatewayAcpHttpResponseTimeoutFor(Uri endpoint, String method) {
if (_isOpenClawTaskSubmitEndpoint(endpoint) &&
_isOpenClawTaskSubmitMethod(method)) {
return const Duration(minutes: 10);
}
return const Duration(seconds: 120);
}

2
test/runtime/assistant_execution_target_test.dart
View File

@ -558,7 +558,7 @@ void main() {
GatewayChatMessage(
id: 'error-1',
role: 'assistant',
text: 'ACP_HTTP_STREAM_CLOSED',
text: 'ACP_HTTP_CONNECTION_CLOSED',
timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
toolCallId: null,
toolName: null,

291
test/runtime/gateway_acp_client_auth_test.dart
View File

@ -375,6 +375,46 @@ void main() {
);
});
test(
'surfaces closed-before-header HTTP failures as ACP diagnostics',
() async {
final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
addTearDown(() => server.close());
server.listen((socket) {
socket.listen((_) {
socket.destroy();
});
});
final endpoint = Uri.parse('http://127.0.0.1:${server.port}');
final client = GatewayAcpClient(endpointResolver: () => endpoint);
await expectLater(
client.request(
method: 'session.start',
params: const <String, dynamic>{},
),
throwsA(
isA<GatewayAcpException>()
.having(
(error) => error.code,
'code',
'ACP_HTTP_CONNECTION_CLOSED',
)
.having(
(error) => error.message,
'message',
contains('closed before the response finished arriving'),
)
.having(
(error) => error.details,
'details',
containsPair('requestUrl', '$endpoint/acp/rpc'),
),
),
);
},
);
test('desktop bridge auth resolver skips unrelated endpoints', () async {
final storeRoot = await Directory.systemTemp.createTemp(
'xworkmate-acp-auth-unrelated-',
@ -647,84 +687,78 @@ void main() {
},
);
test(
'desktop task execution routes OpenClaw through dedicated bridge gateway path',
() async {
final capture = await _startAcpHttpServer();
addTearDown(capture.close);
final client = GatewayAcpClient(
endpointResolver: () => capture.baseEndpoint,
authorizationResolver: (_) async => 'bridge-token',
);
test('desktop task execution routes OpenClaw through bridge RPC', () async {
final capture = await _startAcpHttpServer();
addTearDown(capture.close);
final client = GatewayAcpClient(
endpointResolver: () => capture.baseEndpoint,
authorizationResolver: (_) async => 'bridge-token',
);
final transport = ExternalCodeAgentAcpDesktopTransport(
client: client,
endpointResolver: (_) => capture.baseEndpoint,
taskEndpointResolver: (_) =>
capture.baseEndpoint.replace(path: '/gateway/openclaw'),
);
final transport = ExternalCodeAgentAcpDesktopTransport(
client: client,
endpointResolver: (_) => capture.baseEndpoint,
taskEndpointResolver: (_) => capture.baseEndpoint,
);
await transport.executeTask(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
),
onUpdate: (_) {},
);
await transport.executeTask(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
),
onUpdate: (_) {},
);
expect(capture.authorizationHeader, 'Bearer bridge-token');
expect(capture.acceptHeader, 'application/json');
expect(capture.requestPath, '/gateway/openclaw');
expect(capture.requestPath, isNot(contains('/acp-server')));
expect(capture.requestPath, isNot(contains('/acp-server/gateway')));
final params = _lastRequestParams(capture);
final routing = params['routing'] as Map<String, dynamic>;
expect(params.containsKey('gatewayProvider'), isFalse);
expect(params.containsKey('gatewayProviderId'), isFalse);
expect(params['executionTarget'], 'gateway');
expect(params['requestedExecutionTarget'], 'gateway');
expect(routing['preferredGatewayProviderId'], 'openclaw');
expect(routing['explicitExecutionTarget'], 'gateway');
expect(routing.containsKey('explicitProviderId'), isFalse);
expect(capture.requestBody, contains('"method":"session.start"'));
expect(capture.requestBody, isNot(contains('"method":"thread/start"')));
},
);
expect(capture.authorizationHeader, 'Bearer bridge-token');
expect(capture.acceptHeader, 'text/event-stream, application/json');
expect(capture.requestPath, '/acp/rpc');
expect(capture.requestPath, isNot(contains('/acp-server')));
expect(capture.requestPath, isNot(contains('/acp-server/gateway')));
expect(capture.requestPath, isNot(contains('/gateway/openclaw')));
final params = _lastRequestParams(capture);
final routing = params['routing'] as Map<String, dynamic>;
expect(params.containsKey('gatewayProvider'), isFalse);
expect(params.containsKey('gatewayProviderId'), isFalse);
expect(params['executionTarget'], 'gateway');
expect(params['requestedExecutionTarget'], 'gateway');
expect(routing['preferredGatewayProviderId'], 'openclaw');
expect(routing['explicitExecutionTarget'], 'gateway');
expect(routing.containsKey('explicitProviderId'), isFalse);
expect(capture.requestBody, contains('"method":"session.start"'));
expect(capture.requestBody, isNot(contains('"method":"thread/start"')));
});
test('desktop OpenClaw follow-up routes through bridge RPC', () async {
final capture = await _startAcpHttpServer();
addTearDown(capture.close);
final client = GatewayAcpClient(
endpointResolver: () => capture.baseEndpoint,
authorizationResolver: (_) async => 'bridge-token',
);
final transport = ExternalCodeAgentAcpDesktopTransport(
client: client,
endpointResolver: (_) => capture.baseEndpoint,
taskEndpointResolver: (_) => capture.baseEndpoint,
);
await transport.executeTask(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
resumeSession: true,
),
onUpdate: (_) {},
);
expect(capture.acceptHeader, 'text/event-stream, application/json');
expect(capture.requestPath, '/acp/rpc');
expect(capture.requestPath, isNot(contains('/gateway/openclaw')));
expect(capture.requestBody, contains('"method":"session.message"'));
});
test(
'desktop OpenClaw follow-up routes through dedicated bridge gateway path',
() async {
final capture = await _startAcpHttpServer();
addTearDown(capture.close);
final client = GatewayAcpClient(
endpointResolver: () => capture.baseEndpoint,
authorizationResolver: (_) async => 'bridge-token',
);
final transport = ExternalCodeAgentAcpDesktopTransport(
client: client,
endpointResolver: (_) => capture.baseEndpoint,
taskEndpointResolver: (_) =>
capture.baseEndpoint.replace(path: '/gateway/openclaw'),
);
await transport.executeTask(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
resumeSession: true,
),
onUpdate: (_) {},
);
expect(capture.acceptHeader, 'application/json');
expect(capture.requestPath, '/gateway/openclaw');
expect(capture.requestBody, contains('"method":"session.message"'));
},
);
test(
'OpenClaw task submit uses extended HTTP response timeout',
'bridge RPC session methods use the standard HTTP response timeout',
() {
final openClawEndpoint = Uri.parse(
'https://xworkmate-bridge.svc.plus/gateway/openclaw',
@ -735,11 +769,11 @@ void main() {
expect(
gatewayAcpHttpResponseTimeoutFor(openClawEndpoint, 'session.start'),
const Duration(minutes: 10),
const Duration(seconds: 120),
);
expect(
gatewayAcpHttpResponseTimeoutFor(openClawEndpoint, 'session.message'),
const Duration(minutes: 10),
const Duration(seconds: 120),
);
expect(
gatewayAcpHttpResponseTimeoutFor(acpEndpoint, 'session.start'),
@ -755,62 +789,58 @@ void main() {
},
);
test(
'desktop controller only uses gateway path for OpenClaw task submit',
() {
final controller = AppController(
environmentOverride: const <String, String>{},
);
addTearDown(controller.dispose);
test('desktop controller resolves task requests to the bridge origin', () {
final controller = AppController(
environmentOverride: const <String, String>{},
);
addTearDown(controller.dispose);
final openClawStart = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
),
);
final openClawFollowUp = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
resumeSession: true,
),
);
final unspecifiedGateway = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.unspecified,
),
);
final multiAgentGateway = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
multiAgent: true,
),
);
final agentTask = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.agent,
provider: SingleAgentProvider.codex,
),
);
final openClawStart = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
),
);
final openClawFollowUp = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
resumeSession: true,
),
);
final unspecifiedGateway = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.unspecified,
),
);
final multiAgentGateway = controller
.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
multiAgent: true,
),
);
final agentTask = controller.resolveExternalAcpEndpointForRequestInternal(
_taskRequest(
target: AssistantExecutionTarget.agent,
provider: SingleAgentProvider.codex,
),
);
expect(openClawStart?.path, '/gateway/openclaw');
expect(openClawFollowUp?.path, '/gateway/openclaw');
expect(unspecifiedGateway?.path, '');
expect(multiAgentGateway?.path, '');
expect(agentTask?.path, '');
},
);
expect(openClawStart?.path, '');
expect(openClawFollowUp?.path, '');
expect(unspecifiedGateway?.path, '');
expect(multiAgentGateway?.path, '');
expect(agentTask?.path, '');
});
test(
'desktop controller resolves OpenClaw gateway submit on managed bridge origin',
'desktop controller does not expose OpenClaw gateway path as task endpoint',
() {
final controller = AppController(
environmentOverride: const <String, String>{},
@ -825,12 +855,9 @@ void main() {
),
);
expect(
endpoint.toString(),
'https://xworkmate-bridge.svc.plus/gateway/openclaw',
);
expect(endpoint.toString(), 'https://xworkmate-bridge.svc.plus');
expect(endpoint, isNotNull);
expect(endpoint!.path, isNot('/acp/rpc'));
expect(endpoint!.path, isNot('/gateway/openclaw'));
},
);