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

fix: stabilize ACP gateway connect failures

Browse Source
This commit is contained in:
Haitao Pan 2026-05-09 15:12:51 +08:00
parent 92ea83cbda
commit 093db7dfc5
7 changed files with 408 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
lib/app/app_controller_desktop_runtime_helpers.dart
View File

@ -237,6 +237,19 @@ extension AppControllerDesktopRuntimeHelpers on AppController {
final recoverableTransportCode = recoverableAcpHttpTransportCodeInternal(
error,
);
final unconfirmedConnectCode = unconfirmedAcpHttpConnectCodeInternal(error);
if (unconfirmedConnectCode == gatewayAcpHttpConnectTimeoutCode) {
return appText(
'Bridge 连接超时，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_TIMEOUT',
'Bridge connection timed out; this request was not confirmed and can be retried. Error code: ACP_HTTP_CONNECT_TIMEOUT',
);
}
if (unconfirmedConnectCode == gatewayAcpHttpConnectFailedCode) {
return appText(
'Bridge 连接失败，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_FAILED',
'Bridge connection failed; this request was not confirmed and can be retried. Error code: ACP_HTTP_CONNECT_FAILED',
);
}
if (recoverableTransportCode == 'ACP_HTTP_CONNECTION_CLOSED') {
return appText(
'Bridge 响应读取中断；当前对话已保留，下一次发送会继续同一会话。错误码：ACP_HTTP_CONNECTION_CLOSED',
@ -362,6 +375,23 @@ extension AppControllerDesktopRuntimeHelpers on AppController {
return null;
}
String? unconfirmedAcpHttpConnectCodeInternal(Object error) {
final raw = error.toString().trim();
final primaryCode = gatewayExecutionPrimaryCodeInternal(error);
final detailCode = gatewayExecutionDetailCodeInternal(error);
if (primaryCode == gatewayAcpHttpConnectTimeoutCode ||
detailCode == gatewayAcpHttpConnectTimeoutCode ||
raw.contains(gatewayAcpHttpConnectTimeoutCode)) {
return gatewayAcpHttpConnectTimeoutCode;
}
if (primaryCode == gatewayAcpHttpConnectFailedCode ||
detailCode == gatewayAcpHttpConnectFailedCode ||
raw.contains(gatewayAcpHttpConnectFailedCode)) {
return gatewayAcpHttpConnectFailedCode;
}
return null;
}
String formatAiGatewayHttpErrorInternal(int statusCode, String detail) {
final base = switch (statusCode) {
400 => appText(

38
lib/app/app_controller_desktop_thread_actions.dart
View File

@ -330,7 +330,7 @@ extension AppControllerDesktopThreadActions on AppController {
final sessionKey = normalizedAssistantSessionKeyInternal(
currentSessionKey,
);
final resumeSession = hasCommittedUserTurnForGatewaySessionInternal(
final resumeSession = shouldResumeGatewaySessionForNextSendInternal(
sessionKey,
);
final lifecycleStatus = taskThreadForSessionInternal(
@ -339,7 +339,13 @@ extension AppControllerDesktopThreadActions on AppController {
final lastResultCode = taskThreadForSessionInternal(
sessionKey,
)?.lifecycleState.lastResultCode?.trim().toLowerCase();
final runStatus = resumeSession && lifecycleStatus == 'interrupted'
final continuableTransportResult =
lastResultCode == 'acp_http_connection_closed' ||
lastResultCode ==
gatewayAcpHttpHandshakeInterruptedCode.toLowerCase();
final runStatus =
resumeSession &&
(lifecycleStatus == 'interrupted' || continuableTransportResult)
? 'continuing'
: resumeSession && lastResultCode == 'error'
? 'retrying'
@ -375,6 +381,13 @@ extension AppControllerDesktopThreadActions on AppController {
try {
final dispatch = await codeAgentNodeOrchestratorInternal
.buildGatewayDispatch(buildCodeAgentNodeStateInternal());
upsertTaskThreadInternal(
sessionKey,
lifecycleStatus: runStatus,
lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
lastResultCode: runStatus,
updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
);
final result = await goTaskServiceClientInternal.executeTask(
GoTaskServiceRequest(
sessionId: sessionKey,
@ -490,15 +503,20 @@ extension AppControllerDesktopThreadActions on AppController {
clearAiGatewayStreamingTextInternal(sessionKey);
final recoverableTransportCode =
recoverableAcpHttpTransportCodeInternal(error);
final unconfirmedConnectCode = unconfirmedAcpHttpConnectCodeInternal(
error,
);
final recoverableTransportInterrupted =
recoverableTransportCode != null;
final visibleResultCode =
unconfirmedConnectCode ?? recoverableTransportCode;
upsertTaskThreadInternal(
sessionKey,
lifecycleStatus: recoverableTransportInterrupted
? 'interrupted'
: 'ready',
lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
lastResultCode: recoverableTransportCode ?? 'error',
lastResultCode: visibleResultCode ?? 'error',
lastArtifactSyncStatus: recoverableTransportInterrupted
? 'interrupted'
: null,
@ -537,6 +555,20 @@ extension AppControllerDesktopThreadActions on AppController {
});
}
bool shouldResumeGatewaySessionForNextSendInternal(String sessionKey) {
final normalizedSessionKey = normalizedAssistantSessionKeyInternal(
sessionKey,
);
if (!hasCommittedUserTurnForGatewaySessionInternal(normalizedSessionKey)) {
return false;
}
final lastResultCode = taskThreadForSessionInternal(
normalizedSessionKey,
)?.lifecycleState.lastResultCode?.trim().toUpperCase();
return lastResultCode != gatewayAcpHttpConnectTimeoutCode &&
lastResultCode != gatewayAcpHttpConnectFailedCode;
}
Future<void> abortRun() async {
if (multiAgentRunPendingInternal) {
final sessionKey = normalizedAssistantSessionKeyInternal(

19
lib/runtime/external_code_agent_acp_desktop_transport.dart
View File

@ -1,4 +1,5 @@
import 'dart:async';
import 'dart:io';
import 'package:flutter/foundation.dart';
@ -134,6 +135,17 @@ class ExternalCodeAgentAcpDesktopTransport
);
} on GatewayAcpException {
rethrow;
} on SocketException catch (error) {
final timeout = _socketExceptionLooksLikeConnectTimeout(error);
throw GatewayAcpException(
timeout
? 'ACP HTTP connection timed out before the request was confirmed'
: 'ACP HTTP connection failed before the request was confirmed',
code: timeout
? gatewayAcpHttpConnectTimeoutCode
: gatewayAcpHttpConnectFailedCode,
details: <String, dynamic>{'originalError': error.toString()},
);
} catch (error) {
throw GatewayAcpException(
error.toString(),
@ -304,4 +316,11 @@ class ExternalCodeAgentAcpDesktopTransport
}
return <AssistantExecutionTarget>[defaultTarget];
}
bool _socketExceptionLooksLikeConnectTimeout(SocketException error) {
final lowered = error.toString().toLowerCase();
return lowered.contains('connection timed out') ||
lowered.contains('timed out') ||
lowered.contains('timeout');
}
}

128
lib/runtime/gateway_acp_client.dart
View File

@ -6,8 +6,12 @@ import 'acp_endpoint_paths.dart';
import 'runtime_models.dart';
const int gatewayAcpHttpHandshakeInterruptedRetryCount = 5;
const int gatewayAcpHttpConnectFailureRetryCount = 2;
const Duration gatewayAcpHttpConnectTimeout = Duration(seconds: 12);
const String gatewayAcpHttpHandshakeInterruptedCode =
'ACP_HTTP_HANDSHAKE_INTERRUPTED';
const String gatewayAcpHttpConnectTimeoutCode = 'ACP_HTTP_CONNECT_TIMEOUT';
const String gatewayAcpHttpConnectFailedCode = 'ACP_HTTP_CONNECT_FAILED';
class GatewayAcpException implements Exception {
const GatewayAcpException(
@ -77,6 +81,13 @@ class _GatewayAcpSessionUpdate {
final Map<String, dynamic> payload;
}
enum _GatewayAcpHttpRequestPhase {
connect,
write,
waitingForResponse,
bodyRead,
}
class GatewayAcpMultiAgentRequest {
const GatewayAcpMultiAgentRequest({
required this.sessionId,
@ -564,7 +575,7 @@ class GatewayAcpClient {
);
}
GatewayAcpException? lastHandshakeError;
GatewayAcpException? lastRetryableError;
for (
var attempt = 0;
attempt <= gatewayAcpHttpHandshakeInterruptedRetryCount;
@ -579,21 +590,39 @@ class GatewayAcpClient {
retryAttempt: attempt,
);
} on GatewayAcpException catch (error) {
if (error.code != gatewayAcpHttpHandshakeInterruptedCode ||
attempt == gatewayAcpHttpHandshakeInterruptedRetryCount) {
final retryLimit = _httpRetryCountForError(error);
if (retryLimit == null || attempt >= retryLimit) {
rethrow;
}
lastHandshakeError = error;
await Future<void>.delayed(Duration(milliseconds: 50 * (attempt + 1)));
lastRetryableError = error;
await Future<void>.delayed(_httpRetryDelayFor(error, attempt));
}
}
throw lastHandshakeError ??
throw lastRetryableError ??
const GatewayAcpException(
'ACP HTTP handshake was interrupted before the response started',
code: gatewayAcpHttpHandshakeInterruptedCode,
);
}
int? _httpRetryCountForError(GatewayAcpException error) {
return switch (error.code) {
gatewayAcpHttpHandshakeInterruptedCode =>
gatewayAcpHttpHandshakeInterruptedRetryCount,
gatewayAcpHttpConnectTimeoutCode ||
gatewayAcpHttpConnectFailedCode => gatewayAcpHttpConnectFailureRetryCount,
_ => null,
};
}
Duration _httpRetryDelayFor(GatewayAcpException error, int attempt) {
if (error.code == gatewayAcpHttpConnectTimeoutCode ||
error.code == gatewayAcpHttpConnectFailedCode) {
return Duration(milliseconds: 200 * (1 << attempt));
}
return Duration(milliseconds: 50 * (attempt + 1));
}
Future<Map<String, dynamic>> _requestViaHttpAttempt(
_GatewayAcpRpcRequest request, {
required Uri endpoint,
@ -601,12 +630,22 @@ class GatewayAcpClient {
required String authorizationOverride,
required int retryAttempt,
}) async {
final client = HttpClient()..connectionTimeout = const Duration(seconds: 8);
final client = HttpClient()
..connectionTimeout = gatewayAcpHttpConnectTimeout;
var statusCode = 0;
var contentType = '';
var bodyRead = false;
var phase = _GatewayAcpHttpRequestPhase.connect;
try {
final httpRequest = await client.postUrl(endpoint);
final authorization = await _resolveAuthorizationHeader(
endpoint,
authorizationOverride: authorizationOverride,
);
phase = _GatewayAcpHttpRequestPhase.connect;
final httpRequest = await client
.postUrl(endpoint)
.timeout(gatewayAcpHttpConnectTimeout);
phase = _GatewayAcpHttpRequestPhase.write;
httpRequest.headers.set(
HttpHeaders.contentTypeHeader,
'application/json; charset=utf-8',
@ -615,10 +654,6 @@ class GatewayAcpClient {
HttpHeaders.acceptHeader,
'text/event-stream, application/json',
);
final authorization = await _resolveAuthorizationHeader(
endpoint,
authorizationOverride: authorizationOverride,
);
if (authorization.isNotEmpty) {
httpRequest.headers.set(HttpHeaders.authorizationHeader, authorization);
}
@ -632,6 +667,7 @@ class GatewayAcpClient {
}),
),
);
phase = _GatewayAcpHttpRequestPhase.waitingForResponse;
final response = await httpRequest.close().timeout(
gatewayAcpHttpResponseTimeoutFor(
endpoint,
@ -646,6 +682,7 @@ class GatewayAcpClient {
.value(HttpHeaders.contentTypeHeader)
?.toLowerCase() ??
'';
phase = _GatewayAcpHttpRequestPhase.bodyRead;
if (response.statusCode < 200 || response.statusCode >= 300) {
final body = await response.transform(utf8.decoder).join();
bodyRead = body.isNotEmpty;
@ -700,6 +737,20 @@ class GatewayAcpClient {
};
} on GatewayAcpException {
rethrow;
} on TimeoutException catch (error) {
if (phase == _GatewayAcpHttpRequestPhase.connect) {
throw _connectException(
endpoint: endpoint,
statusCode: statusCode,
contentType: contentType,
bodyRead: bodyRead,
retryAttempt: retryAttempt,
phase: phase,
originalError: error,
timeout: true,
);
}
rethrow;
} on HandshakeException catch (error) {
throw _handshakeInterruptedException(
endpoint: endpoint,
@ -725,6 +776,20 @@ class GatewayAcpClient {
originalError: error,
);
}
if (phase == _GatewayAcpHttpRequestPhase.connect &&
statusCode == 0 &&
!bodyRead) {
throw _connectException(
endpoint: endpoint,
statusCode: statusCode,
contentType: contentType,
bodyRead: bodyRead,
retryAttempt: retryAttempt,
phase: phase,
originalError: error,
timeout: _looksLikeConnectTimeout(error.toString()),
);
}
rethrow;
} on HttpException catch (error) {
if (_looksLikeConnectionClosedBeforeResponse(error.toString())) {
@ -746,6 +811,38 @@ class GatewayAcpClient {
}
}
GatewayAcpException _connectException({
required Uri endpoint,
required int statusCode,
required String contentType,
required bool bodyRead,
required int retryAttempt,
required _GatewayAcpHttpRequestPhase phase,
required Object originalError,
required bool timeout,
}) {
final code = timeout
? gatewayAcpHttpConnectTimeoutCode
: gatewayAcpHttpConnectFailedCode;
final message = timeout
? 'ACP HTTP connection timed out before the request was confirmed'
: 'ACP HTTP connection failed before the request was confirmed';
return GatewayAcpException(
message,
code: code,
details: <String, dynamic>{
'requestUrl': endpoint.toString(),
'statusCode': statusCode,
'contentType': contentType,
'bodyRead': bodyRead,
'phase': phase.name,
'retryAttempt': retryAttempt,
'maxRetryAttempts': gatewayAcpHttpConnectFailureRetryCount,
'originalError': originalError.toString(),
},
);
}
GatewayAcpException _handshakeInterruptedException({
required Uri endpoint,
required int statusCode,
@ -792,6 +889,13 @@ class GatewayAcpClient {
lowered.contains('stream closed');
}
bool _looksLikeConnectTimeout(String raw) {
final lowered = raw.toLowerCase();
return lowered.contains('connection timed out') ||
lowered.contains('timed out') ||
lowered.contains('timeout');
}
String _describeHttpError({
required int statusCode,
required String contentType,

39
test/runtime/assistant_connection_state_test.dart
View File

@ -208,6 +208,45 @@ void main() {
},
);
test(
'labels ACP HTTP connect timeouts as unconfirmed retryable requests',
() async {
final controller = await _isolatedController();
addTearDown(controller.dispose);
final label = controller.gatewayExecutionErrorLabelInternal(
const GatewayAcpException(
'SocketException: HTTP connection timed out after 0:00:08.000000, host: xworkmate-bridge.svc.plus, port: 443',
code: gatewayAcpHttpConnectTimeoutCode,
),
target: AssistantExecutionTarget.gateway,
);
expect(label, 'Bridge 连接超时，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_TIMEOUT');
expect(label, isNot(contains('SocketException')));
expect(label, isNot(contains('0:00:08')));
},
);
test(
'labels ACP HTTP connect failures as unconfirmed retryable requests',
() async {
final controller = await _isolatedController();
addTearDown(controller.dispose);
final label = controller.gatewayExecutionErrorLabelInternal(
const GatewayAcpException(
'Connection refused',
code: gatewayAcpHttpConnectFailedCode,
),
target: AssistantExecutionTarget.gateway,
);
expect(label, 'Bridge 连接失败，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_FAILED');
expect(label, isNot(contains('Connection refused')));
},
);
test(
'labels unavailable session continuation without starting a new flow',
() async {

66
test/runtime/assistant_execution_target_test.dart
View File

@ -726,6 +726,68 @@ void main() {
},
);
test(
'sendChatMessage starts a new session after ACP HTTP connect timeout',
() async {
final fakeGoTaskService = _RecordingGoTaskServiceClient()
..outcomes.add(
const GatewayAcpException(
'ACP HTTP connection timed out before the request was confirmed',
code: gatewayAcpHttpConnectTimeoutCode,
),
)
..outcomes.add(
const GoTaskServiceResult(
success: true,
message: 'retried from a confirmed new start',
turnId: 'turn-2',
raw: <String, dynamic>{},
errorMessage: '',
resolvedModel: '',
route: GoTaskServiceRoute.externalAcpSingle,
),
);
final controller = _connectedController(fakeGoTaskService);
addTearDown(controller.dispose);
await controller.sessionsController.switchSession('session-1');
await controller.sendChatMessage('first turn');
expect(fakeGoTaskService.requests, hasLength(1));
expect(fakeGoTaskService.requests.single.resumeSession, isFalse);
final failedThread = controller.taskThreadForSessionInternal(
'session-1',
);
expect(failedThread?.lifecycleState.status, 'ready');
expect(
failedThread?.lifecycleState.lastResultCode,
gatewayAcpHttpConnectTimeoutCode,
);
expect(failedThread?.lastArtifactSyncStatus, isNull);
expect(
controller.chatMessages.last.text,
'Bridge 连接超时，本轮请求未确认，可重试。错误码：ACP_HTTP_CONNECT_TIMEOUT',
);
await controller.sendChatMessage('retry after unconfirmed connect');
expect(fakeGoTaskService.requests, hasLength(2));
expect(fakeGoTaskService.requests.last.resumeSession, isFalse);
await _waitForLastChatMessageText(
controller,
'retried from a confirmed new start',
);
expect(
controller.chatMessages.last.text,
'retried from a confirmed new start',
);
final thread = controller.taskThreadForSessionInternal('session-1');
expect(thread?.lifecycleState.status, 'ready');
expect(thread?.lifecycleState.lastResultCode, 'success');
},
);
test(
'sendChatMessage hides OpenClaw artifact guard text after an interrupted continuation',
() async {
@ -1325,7 +1387,7 @@ Future<void> _waitForLastChatMessageText(
AppController controller,
String expectedText,
) async {
final deadline = DateTime.now().add(const Duration(seconds: 2));
final deadline = DateTime.now().add(const Duration(seconds: 15));
while (DateTime.now().isBefore(deadline)) {
if (controller.chatMessages.isNotEmpty &&
controller.chatMessages.last.text == expectedText) {
@ -1541,7 +1603,7 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient {
}
Future<void> waitForRequestCount(int count) async {
final deadline = DateTime.now().add(const Duration(seconds: 5));
final deadline = DateTime.now().add(const Duration(seconds: 15));
while (requests.length < count && DateTime.now().isBefore(deadline)) {
await Future<void>.delayed(const Duration(milliseconds: 10));
}

105
test/runtime/gateway_acp_client_auth_test.dart
View File

@ -530,6 +530,91 @@ void main() {
},
);
test(
'retries failed connect attempts before surfacing unconfirmed diagnostics',
() async {
final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
final port = server.port;
await server.close();
final endpoint = Uri.parse('http://127.0.0.1:$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',
gatewayAcpHttpConnectFailedCode,
)
.having(
(error) => error.message,
'message',
contains('before the request was confirmed'),
)
.having(
(error) => error.details,
'details',
allOf(
containsPair('requestUrl', '$endpoint/acp/rpc'),
containsPair(
'maxRetryAttempts',
gatewayAcpHttpConnectFailureRetryCount,
),
containsPair(
'retryAttempt',
gatewayAcpHttpConnectFailureRetryCount,
),
containsPair('phase', 'connect'),
),
),
),
);
},
);
test(
'desktop transport preserves socket timeout as unconfirmed ACP diagnostics',
() async {
final transport = ExternalCodeAgentAcpDesktopTransport(
client: _SocketThrowingGatewayAcpClient(
const SocketException(
'HTTP connection timed out after 0:00:08.000000, host: xworkmate-bridge.svc.plus, port: 443',
),
),
endpointResolver: (_) =>
Uri.parse('https://xworkmate-bridge.svc.plus'),
);
await expectLater(
transport.executeTask(
_taskRequest(
target: AssistantExecutionTarget.gateway,
provider: SingleAgentProvider.openclaw,
),
onUpdate: (_) {},
),
throwsA(
isA<GatewayAcpException>()
.having(
(error) => error.code,
'code',
gatewayAcpHttpConnectTimeoutCode,
)
.having(
(error) => error.toString(),
'diagnostic',
isNot(contains('EXTERNAL_ACP_GATEWAY_ERROR')),
),
),
);
},
);
test('desktop bridge auth resolver skips unrelated endpoints', () async {
final storeRoot = await Directory.systemTemp.createTemp(
'xworkmate-acp-auth-unrelated-',
@ -1437,6 +1522,26 @@ void main() {
});
}
class _SocketThrowingGatewayAcpClient extends GatewayAcpClient {
_SocketThrowingGatewayAcpClient(this.error)
: super(
endpointResolver: () => Uri.parse('https://xworkmate-bridge.svc.plus'),
);
final SocketException error;
@override
Future<Map<String, dynamic>> request({
required String method,
required Map<String, dynamic> params,
void Function(Map<String, dynamic>)? onNotification,
Uri? endpointOverride,
String authorizationOverride = '',
}) async {
throw error;
}
}
GoTaskServiceRequest _taskRequest({
required AssistantExecutionTarget target,
required SingleAgentProvider provider,