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

fix: retry interrupted acp handshakes

Browse Source
This commit is contained in:
Haitao Pan 2026-05-08 16:01:18 +08:00
parent ad361c1ac9
commit 7f166d07bb
8 changed files with 351 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
lib/app/app_controller_desktop_runtime_helpers.dart
View File

@ -230,12 +230,21 @@ extension AppControllerDesktopRuntimeHelpers on AppController {
final lowered = raw.toLowerCase();
final detailCode = gatewayExecutionDetailCodeInternal(error);
final primaryCode = gatewayExecutionPrimaryCodeInternal(error);
if (isAcpHttpConnectionClosedErrorInternal(error)) {
final recoverableTransportCode = recoverableAcpHttpTransportCodeInternal(
error,
);
if (recoverableTransportCode == 'ACP_HTTP_CONNECTION_CLOSED') {
return appText(
'Bridge 响应读取中断；当前对话已保留，下一次发送会继续同一会话。错误码：ACP_HTTP_CONNECTION_CLOSED',
'Bridge response was interrupted; this conversation was kept, and the next send will continue the same session. Error code: ACP_HTTP_CONNECTION_CLOSED',
);
}
if (recoverableTransportCode == 'ACP_HTTP_HANDSHAKE_INTERRUPTED') {
return appText(
'Bridge 握手中断；当前对话已保留，下一次发送会继续同一会话。错误码：ACP_HTTP_HANDSHAKE_INTERRUPTED',
'Bridge handshake was interrupted; this conversation was kept, and the next send will continue the same session. Error code: ACP_HTTP_HANDSHAKE_INTERRUPTED',
);
}
final continuationUnavailable =
primaryCode == 'SESSION_CONTINUATION_UNAVAILABLE' ||
detailCode == 'SESSION_CONTINUATION_UNAVAILABLE' ||
@ -295,12 +304,25 @@ extension AppControllerDesktopRuntimeHelpers on AppController {
}
bool isAcpHttpConnectionClosedErrorInternal(Object error) {
return recoverableAcpHttpTransportCodeInternal(error) ==
'ACP_HTTP_CONNECTION_CLOSED';
}
String? recoverableAcpHttpTransportCodeInternal(Object error) {
final raw = error.toString().trim();
final primaryCode = gatewayExecutionPrimaryCodeInternal(error);
final detailCode = gatewayExecutionDetailCodeInternal(error);
return primaryCode == 'ACP_HTTP_CONNECTION_CLOSED' ||
if (primaryCode == 'ACP_HTTP_CONNECTION_CLOSED' ||
detailCode == 'ACP_HTTP_CONNECTION_CLOSED' ||
raw.contains('ACP_HTTP_CONNECTION_CLOSED');
raw.contains('ACP_HTTP_CONNECTION_CLOSED')) {
return 'ACP_HTTP_CONNECTION_CLOSED';
}
if (primaryCode == 'ACP_HTTP_HANDSHAKE_INTERRUPTED' ||
detailCode == 'ACP_HTTP_HANDSHAKE_INTERRUPTED' ||
raw.contains('ACP_HTTP_HANDSHAKE_INTERRUPTED')) {
return 'ACP_HTTP_HANDSHAKE_INTERRUPTED';
}
return null;
}
String formatAiGatewayHttpErrorInternal(int statusCode, String detail) {

19
lib/app/app_controller_desktop_thread_actions.dart
View File

@ -470,17 +470,20 @@ extension AppControllerDesktopThreadActions on AppController {
);
} catch (error) {
clearAiGatewayStreamingTextInternal(sessionKey);
final connectionClosed = isAcpHttpConnectionClosedErrorInternal(
error,
);
final recoverableTransportCode =
recoverableAcpHttpTransportCodeInternal(error);
final recoverableTransportInterrupted =
recoverableTransportCode != null;
upsertTaskThreadInternal(
sessionKey,
lifecycleStatus: connectionClosed ? 'interrupted' : 'ready',
lifecycleStatus: recoverableTransportInterrupted
? 'interrupted'
: 'ready',
lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
lastResultCode: connectionClosed
? 'ACP_HTTP_CONNECTION_CLOSED'
: 'error',
lastArtifactSyncStatus: connectionClosed ? 'interrupted' : null,
lastResultCode: recoverableTransportCode ?? 'error',
lastArtifactSyncStatus: recoverableTransportInterrupted
? 'interrupted'
: null,
updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
);
appendLocalSessionMessageInternal(

105
lib/runtime/gateway_acp_client.dart
View File

@ -5,6 +5,10 @@ import 'dart:io';
import 'acp_endpoint_paths.dart';
import 'runtime_models.dart';
const int gatewayAcpHttpHandshakeInterruptedRetryCount = 5;
const String gatewayAcpHttpHandshakeInterruptedCode =
'ACP_HTTP_HANDSHAKE_INTERRUPTED';
class GatewayAcpException implements Exception {
const GatewayAcpException(
this.message, {
@ -560,6 +564,43 @@ class GatewayAcpClient {
);
}
GatewayAcpException? lastHandshakeError;
for (
var attempt = 0;
attempt <= gatewayAcpHttpHandshakeInterruptedRetryCount;
attempt += 1
) {
try {
return await _requestViaHttpAttempt(
request,
endpoint: endpoint,
onNotification: onNotification,
authorizationOverride: authorizationOverride,
retryAttempt: attempt,
);
} on GatewayAcpException catch (error) {
if (error.code != gatewayAcpHttpHandshakeInterruptedCode ||
attempt == gatewayAcpHttpHandshakeInterruptedRetryCount) {
rethrow;
}
lastHandshakeError = error;
await Future<void>.delayed(Duration(milliseconds: 50 * (attempt + 1)));
}
}
throw lastHandshakeError ??
const GatewayAcpException(
'ACP HTTP handshake was interrupted before the response started',
code: gatewayAcpHttpHandshakeInterruptedCode,
);
}
Future<Map<String, dynamic>> _requestViaHttpAttempt(
_GatewayAcpRpcRequest request, {
required Uri endpoint,
required void Function(Map<String, dynamic>) onNotification,
required String authorizationOverride,
required int retryAttempt,
}) async {
final client = HttpClient()..connectionTimeout = const Duration(seconds: 8);
var statusCode = 0;
var contentType = '';
@ -652,6 +693,32 @@ class GatewayAcpClient {
};
} on GatewayAcpException {
rethrow;
} on HandshakeException catch (error) {
throw _handshakeInterruptedException(
endpoint: endpoint,
statusCode: statusCode,
contentType: contentType,
bodyRead: bodyRead,
retryAttempt: retryAttempt,
originalError: error,
);
} on SocketException catch (error) {
if (_looksLikeHandshakeInterruptedSocketError(
error.toString(),
endpoint: endpoint,
statusCode: statusCode,
bodyRead: bodyRead,
)) {
throw _handshakeInterruptedException(
endpoint: endpoint,
statusCode: statusCode,
contentType: contentType,
bodyRead: bodyRead,
retryAttempt: retryAttempt,
originalError: error,
);
}
rethrow;
} on HttpException catch (error) {
if (_looksLikeConnectionClosedBeforeResponse(error.toString())) {
throw GatewayAcpException(
@ -672,6 +739,44 @@ class GatewayAcpClient {
}
}
GatewayAcpException _handshakeInterruptedException({
required Uri endpoint,
required int statusCode,
required String contentType,
required bool bodyRead,
required int retryAttempt,
required Object originalError,
}) {
return GatewayAcpException(
'ACP HTTP handshake was interrupted before the response started',
code: gatewayAcpHttpHandshakeInterruptedCode,
details: <String, dynamic>{
'requestUrl': endpoint.toString(),
'statusCode': statusCode,
'contentType': contentType,
'bodyRead': bodyRead,
'retryAttempt': retryAttempt,
'maxRetryAttempts': gatewayAcpHttpHandshakeInterruptedRetryCount,
'originalError': originalError.toString(),
},
);
}
bool _looksLikeHandshakeInterruptedSocketError(
String raw, {
required Uri endpoint,
required int statusCode,
required bool bodyRead,
}) {
if (endpoint.scheme != 'https' || statusCode != 0 || bodyRead) {
return false;
}
final lowered = raw.toLowerCase();
return lowered.contains('connection reset') ||
lowered.contains('read failed') ||
lowered.contains('connection terminated during handshake');
}
bool _looksLikeConnectionClosedBeforeResponse(String raw) {
final lowered = raw.toLowerCase();
return lowered.contains('connection closed before full header') ||

22
lib/widgets/assistant_task_progress_bar.dart
View File

@ -125,17 +125,25 @@ AssistantTaskProgressState assistantTaskProgressState({
);
}
final result = lastResultCode.trim().toUpperCase();
if (status == 'interrupted' ||
syncStatus == 'interrupted' ||
result == 'ACP_HTTP_CONNECTION_CLOSED') {
if (status == 'interrupted' || syncStatus == 'interrupted') {
return AssistantTaskProgressState(
phase: AssistantTaskProgressPhase.interrupted,
label: appText(
'Bridge 响应中断，等待下一次发送续写同一会话。',
'Bridge response interrupted; the next send will continue this session.',
),
label: _interruptedTaskProgressLabel(result),
value: 0.48,
);
}
return const AssistantTaskProgressState.idle();
}
String _interruptedTaskProgressLabel(String result) {
if (result == 'ACP_HTTP_HANDSHAKE_INTERRUPTED') {
return appText(
'Bridge 握手中断，等待下一次发送续写同一会话。',
'Bridge handshake interrupted; the next send will continue this session.',
);
}
return appText(
'Bridge 响应中断，等待下一次发送续写同一会话。',
'Bridge response interrupted; the next send will continue this session.',
);
}

21
test/features/assistant/assistant_task_progress_bar_test.dart
View File

@ -114,6 +114,27 @@ void main() {
expect(indicator.value, 0.48);
});
testWidgets('shows interrupted state after ACP handshake interruption', (
tester,
) async {
await tester.pumpWidget(
_buildTestApp(
assistantTaskProgressState(
pending: false,
lifecycleStatus: 'interrupted',
lastResultCode: 'ACP_HTTP_HANDSHAKE_INTERRUPTED',
artifactSyncStatus: 'interrupted',
),
),
);
expect(find.text('Bridge 握手中断，等待下一次发送续写同一会话。'), findsOneWidget);
final indicator = tester.widget<LinearProgressIndicator>(
find.byKey(const Key('assistant-task-progress-indicator')),
);
expect(indicator.value, 0.48);
});
testWidgets('hides idle progress state', (tester) async {
await tester.pumpWidget(
_buildTestApp(const AssistantTaskProgressState.idle()),

26
test/runtime/assistant_connection_state_test.dart
View File

@ -182,6 +182,32 @@ void main() {
},
);
test(
'labels interrupted ACP HTTP handshakes as recoverable conversation state',
() async {
final controller = await _isolatedController();
addTearDown(controller.dispose);
final label = controller.gatewayExecutionErrorLabelInternal(
const GatewayAcpException(
'ACP HTTP handshake was interrupted before the response started',
code: gatewayAcpHttpHandshakeInterruptedCode,
),
target: AssistantExecutionTarget.gateway,
);
expect(
label,
'Bridge 握手中断；当前对话已保留，下一次发送会继续同一会话。错误码：ACP_HTTP_HANDSHAKE_INTERRUPTED',
);
expect(
label,
isNot(contains('Connection terminated during handshake')),
);
expect(label, isNot(contains('handshake was interrupted')));
},
);
test(
'labels unavailable session continuation without starting a new flow',
() async {

95
test/runtime/assistant_execution_target_test.dart
View File

@ -725,6 +725,99 @@ void main() {
},
);
test(
'sendChatMessage continues the same session after ACP HTTP handshake interruption',
() async {
final localWorkspace = await Directory.systemTemp.createTemp(
'xworkmate-acp-handshake-interrupt-artifacts-',
);
addTearDown(() async {
if (await localWorkspace.exists()) {
await localWorkspace.delete(recursive: true);
}
});
final fakeGoTaskService = _RecordingGoTaskServiceClient()
..updatesBeforeNextOutcome.add(
const GoTaskServiceUpdate(
sessionId: 'session-1',
threadId: 'session-1',
turnId: 'turn-1',
type: 'delta',
text: 'handshake partial output must not persist',
message: '',
pending: true,
error: false,
route: GoTaskServiceRoute.externalAcpSingle,
payload: <String, dynamic>{},
),
)
..outcomes.add(
const GatewayAcpException(
'ACP HTTP handshake was interrupted before the response started',
code: gatewayAcpHttpHandshakeInterruptedCode,
),
)
..outcomes.add(
GoTaskServiceResult(
success: true,
message: '全部 6 个文件已生成 ✅',
turnId: 'turn-2',
raw: <String, dynamic>{'artifacts': _generatedArtifactPayloads()},
errorMessage: '',
resolvedModel: '',
route: GoTaskServiceRoute.externalAcpSingle,
),
);
final controller = _connectedController(fakeGoTaskService);
addTearDown(controller.dispose);
controller.resolvedUserHomeDirectoryInternal = localWorkspace.path;
await controller.sessionsController.switchSession('session-1');
await controller.sendChatMessage('first turn');
expect(fakeGoTaskService.requests, hasLength(1));
expect(fakeGoTaskService.requests.single.resumeSession, isFalse);
final interruptedThread = controller.taskThreadForSessionInternal(
'session-1',
);
expect(interruptedThread?.lifecycleState.status, 'interrupted');
expect(
interruptedThread?.lifecycleState.lastResultCode,
gatewayAcpHttpHandshakeInterruptedCode,
);
expect(
controller.chatMessages.last.text,
'Bridge 握手中断；当前对话已保留，下一次发送会继续同一会话。错误码：ACP_HTTP_HANDSHAKE_INTERRUPTED',
);
expect(
controller.chatMessages.map((message) => message.text),
isNot(contains('handshake partial output must not persist')),
);
await controller.sendChatMessage('follow up');
expect(fakeGoTaskService.requests, hasLength(2));
expect(fakeGoTaskService.requests.last.resumeSession, isTrue);
expect(controller.chatMessages.last.text, '全部 6 个文件已生成 ✅');
final thread = controller.taskThreadForSessionInternal('session-1');
expect(thread?.lifecycleState.status, 'ready');
expect(thread?.lastArtifactSyncStatus, 'synced');
expect(thread?.lastArtifactSyncAtMs, greaterThan(0));
final workspacePath = controller.assistantWorkspacePathForSession(
'session-1',
);
for (final artifact in _generatedArtifactPayloads()) {
final relativePath = artifact['relativePath']! as String;
final content = artifact['content']! as String;
expect(
await File('$workspacePath/$relativePath').readAsString(),
content,
);
}
},
);
test(
'chatMessages does not duplicate persisted local turn messages',
() async {
@ -797,7 +890,7 @@ void main() {
final controller = _connectedController(fakeGoTaskService);
addTearDown(controller.dispose);
await controller.sessionsController.switchSession('task-a');
await controller.switchSession('task-a');
final taskAFuture = controller.sendChatMessage('task A');
await fakeGoTaskService.waitForRequestCount(1);
expect(fakeGoTaskService.requests.single.sessionId, 'task-a');

54
test/runtime/gateway_acp_client_auth_test.dart
View File

@ -476,6 +476,60 @@ void main() {
},
);
test(
'retries interrupted TLS handshakes before surfacing ACP diagnostics',
() async {
final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
var acceptedSockets = 0;
addTearDown(() => server.close());
server.listen((socket) {
acceptedSockets += 1;
socket.destroy();
});
final endpoint = Uri.parse('https://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',
gatewayAcpHttpHandshakeInterruptedCode,
)
.having(
(error) => error.message,
'message',
contains('handshake was interrupted'),
)
.having(
(error) => error.details,
'details',
allOf(
containsPair('requestUrl', '$endpoint/acp/rpc'),
containsPair(
'maxRetryAttempts',
gatewayAcpHttpHandshakeInterruptedRetryCount,
),
containsPair(
'retryAttempt',
gatewayAcpHttpHandshakeInterruptedRetryCount,
),
),
),
),
);
expect(
acceptedSockets,
gatewayAcpHttpHandshakeInterruptedRetryCount + 1,
);
},
);
test('desktop bridge auth resolver skips unrelated endpoints', () async {
final storeRoot = await Directory.systemTemp.createTemp(
'xworkmate-acp-auth-unrelated-',