From 9d14a35b9dbba5fd407caee41ca88f350c7adff2 Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Wed, 8 Apr 2026 17:23:07 +0800 Subject: [PATCH] Fix OpenClaw status and Go task flow --- ...pp_controller_desktop_thread_sessions.dart | 18 +++- .../go_task_service_desktop_service.dart | 82 +++++++++++++++++++ lib/runtime/runtime_controllers_gateway.dart | 77 ----------------- .../runtime_models_runtime_payloads.dart | 7 +- ..._execution_target_switch_suite_thread.dart | 54 ++++++++++++ test/runtime/gateway_runtime_suite.dart | 22 +++-- .../go_task_service_desktop_service_test.dart | 45 ++++++++++ 7 files changed, 215 insertions(+), 90 deletions(-) diff --git a/lib/app/app_controller_desktop_thread_sessions.dart b/lib/app/app_controller_desktop_thread_sessions.dart index 89b8fc4b..ad90f712 100644 --- a/lib/app/app_controller_desktop_thread_sessions.dart +++ b/lib/app/app_controller_desktop_thread_sessions.dart @@ -451,17 +451,27 @@ extension AppControllerDesktopThreadSessions on AppController { ? connection.remoteAddress!.trim() : fallbackAddress) : fallbackAddress; - final status = matchesTarget + final rawStatus = matchesTarget ? connection.status : RuntimeConnectionStatus.offline; + final pairingRequired = matchesTarget && connection.pairingRequired; + final gatewayTokenMissing = matchesTarget && connection.gatewayTokenMissing; + final status = pairingRequired || gatewayTokenMissing + ? RuntimeConnectionStatus.error + : rawStatus; + final primaryLabel = pairingRequired + ? appText('需配对', 'Pairing Required') + : gatewayTokenMissing + ? appText('缺少令牌', 'Missing Token') + : status.label; return AssistantThreadConnectionState( executionTarget: target, status: status, - primaryLabel: status.label, + primaryLabel: primaryLabel, detailLabel: detail, ready: status == RuntimeConnectionStatus.connected, - pairingRequired: matchesTarget && connection.pairingRequired, - gatewayTokenMissing: matchesTarget && connection.gatewayTokenMissing, + pairingRequired: pairingRequired, + gatewayTokenMissing: gatewayTokenMissing, lastError: matchesTarget ? connection.lastError?.trim() : null, ); } diff --git a/lib/runtime/go_task_service_desktop_service.dart b/lib/runtime/go_task_service_desktop_service.dart index dab10b14..1cdb8c27 100644 --- a/lib/runtime/go_task_service_desktop_service.dart +++ b/lib/runtime/go_task_service_desktop_service.dart @@ -5,6 +5,11 @@ import 'go_task_service_client.dart'; import 'runtime_models.dart'; class DesktopGoTaskService implements GoTaskServiceClient { + static const Duration _openClawTaskRecoveryTimeout = Duration(seconds: 35); + static const Duration _openClawTaskRecoveryPollInterval = Duration( + milliseconds: 800, + ); + DesktopGoTaskService({ required GatewayRuntime gateway, required ExternalCodeAgentAcpTransport acpTransport, @@ -111,6 +116,7 @@ class DesktopGoTaskService implements GoTaskServiceClient { if (!_gateway.isConnected) { throw GatewayRuntimeException('gateway not connected'); } + final historyBaseline = await _gateway.loadHistory(request.sessionId); final runId = await _gateway.sendChat( sessionKey: request.sessionId, message: request.prompt, @@ -126,6 +132,13 @@ class DesktopGoTaskService implements GoTaskServiceClient { ); _pendingOpenClawTasksByRunId[runId] = pending; _openClawRunIdsBySession[request.sessionId] = runId; + final recovered = await _recoverOpenClawTaskFromHistory( + pending, + historyBaseline, + ); + if (recovered != null) { + return recovered; + } return pending.completer.future; } @@ -202,6 +215,75 @@ class DesktopGoTaskService implements GoTaskServiceClient { ), ); } + + Future _recoverOpenClawTaskFromHistory( + _PendingOpenClawTask pending, + List historyBaseline, + ) async { + final baselineAssistantFingerprint = _assistantMessageFingerprint( + historyBaseline, + ); + final deadline = DateTime.now().add(_openClawTaskRecoveryTimeout); + while (!pending.completer.isCompleted && DateTime.now().isBefore(deadline)) { + await Future.delayed(_openClawTaskRecoveryPollInterval); + if (pending.completer.isCompleted) { + return null; + } + final history = await _gateway.loadHistory(pending.request.sessionId); + final latestAssistant = _latestAssistantMessage(history); + if (latestAssistant == null) { + continue; + } + final fingerprint = _messageFingerprint(latestAssistant); + if (fingerprint == baselineAssistantFingerprint) { + continue; + } + final result = GoTaskServiceResult( + success: true, + message: latestAssistant.text.trim(), + turnId: pending.runId, + raw: { + 'recoveredFromHistory': true, + 'sessionId': pending.request.sessionId, + }, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.openClawTask, + ); + _pendingOpenClawTasksByRunId.remove(pending.runId); + _openClawRunIdsBySession.remove(pending.request.sessionId); + if (!pending.completer.isCompleted) { + pending.completer.complete(result); + } + return result; + } + return null; + } + + GatewayChatMessage? _latestAssistantMessage(List history) { + for (final message in history.reversed) { + if (message.role.trim().toLowerCase() != 'assistant') { + continue; + } + if (message.text.trim().isEmpty) { + continue; + } + return message; + } + return null; + } + + String _assistantMessageFingerprint(List history) { + final latest = _latestAssistantMessage(history); + if (latest == null) { + return ''; + } + return _messageFingerprint(latest); + } + + String _messageFingerprint(GatewayChatMessage message) { + return '${message.timestampMs ?? 0}|${message.text.trim()}'; + } } class _PendingOpenClawTask { diff --git a/lib/runtime/runtime_controllers_gateway.dart b/lib/runtime/runtime_controllers_gateway.dart index a923238d..9ca05248 100644 --- a/lib/runtime/runtime_controllers_gateway.dart +++ b/lib/runtime/runtime_controllers_gateway.dart @@ -178,8 +178,6 @@ class GatewayChatController extends ChangeNotifier { List messagesInternal = const []; String sessionKeyInternal = 'main'; bool loadingInternal = false; - bool sendingInternal = false; - bool abortingInternal = false; String? errorInternal; String? streamingAssistantTextInternal; final Set pendingRunsInternal = {}; @@ -187,8 +185,6 @@ class GatewayChatController extends ChangeNotifier { List get messages => messagesInternal; String get sessionKey => sessionKeyInternal; bool get loading => loadingInternal; - bool get sending => sendingInternal; - bool get aborting => abortingInternal; String? get error => errorInternal; String? get streamingAssistantText => streamingAssistantTextInternal; bool get hasPendingRun => pendingRunsInternal.isNotEmpty; @@ -219,79 +215,6 @@ class GatewayChatController extends ChangeNotifier { } } - Future sendMessage({ - required String sessionKey, - required String message, - required String thinking, - List attachments = - const [], - String? agentId, - Map? metadata, - }) async { - final trimmed = message.trim(); - if ((trimmed.isEmpty && attachments.isEmpty) || - !runtimeInternal.isConnected) { - return; - } - sessionKeyInternal = sessionKey.trim().isEmpty ? 'main' : sessionKey.trim(); - sendingInternal = true; - errorInternal = null; - streamingAssistantTextInternal = null; - messagesInternal = List.from(messagesInternal) - ..add( - GatewayChatMessage( - id: ephemeralIdInternal(), - role: 'user', - text: trimmed.isEmpty ? 'See attached.' : trimmed, - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: null, - stopReason: null, - pending: false, - error: false, - ), - ); - notifyListeners(); - try { - final runId = await runtimeInternal.sendChat( - sessionKey: sessionKeyInternal, - message: trimmed.isEmpty ? 'See attached.' : trimmed, - thinking: thinking, - attachments: attachments, - agentId: agentId, - metadata: metadata, - ); - pendingRunsInternal.add(runId); - } catch (error) { - errorInternal = error.toString(); - } finally { - sendingInternal = false; - notifyListeners(); - } - } - - Future abortRun() async { - if (pendingRunsInternal.isEmpty || !runtimeInternal.isConnected) { - return; - } - abortingInternal = true; - notifyListeners(); - try { - final runIds = pendingRunsInternal.toList(growable: false); - for (final runId in runIds) { - await runtimeInternal.abortChat( - sessionKey: sessionKeyInternal, - runId: runId, - ); - } - } catch (error) { - errorInternal = error.toString(); - } finally { - abortingInternal = false; - notifyListeners(); - } - } - void handleEvent(GatewayPushEvent event) { if (event.event == 'chat.run') { handleChatRunEventInternal(asMap(event.payload)); diff --git a/lib/runtime/runtime_models_runtime_payloads.dart b/lib/runtime/runtime_models_runtime_payloads.dart index fd70b90b..05e46928 100644 --- a/lib/runtime/runtime_models_runtime_payloads.dart +++ b/lib/runtime/runtime_models_runtime_payloads.dart @@ -146,10 +146,9 @@ class GatewayConnectionSnapshot { final detailCode = lastErrorDetailCode?.trim().toUpperCase(); final errorCode = lastErrorCode?.trim().toUpperCase(); final errorText = lastError?.toLowerCase() ?? ''; - return status != RuntimeConnectionStatus.connected && - (detailCode == 'PAIRING_REQUIRED' || - errorCode == 'NOT_PAIRED' || - errorText.contains('pairing required')); + return detailCode == 'PAIRING_REQUIRED' || + errorCode == 'NOT_PAIRED' || + errorText.contains('pairing required'); } bool get gatewayTokenMissing { diff --git a/test/runtime/app_controller_execution_target_switch_suite_thread.dart b/test/runtime/app_controller_execution_target_switch_suite_thread.dart index cef160ec..cae2633a 100644 --- a/test/runtime/app_controller_execution_target_switch_suite_thread.dart +++ b/test/runtime/app_controller_execution_target_switch_suite_thread.dart @@ -471,5 +471,59 @@ void registerExecutionTargetSwitchThreadTests() { ); }, ); + + test( + 'AppController surfaces pairing-required state on the active assistant thread even if transport still says connected', + () async { + SharedPreferences.setMockInitialValues({}); + final tempDirectory = await Directory.systemTemp.createTemp( + 'xworkmate-thread-pairing-state-', + ); + addTearDown(() async { + await deleteDirectoryWithRetryInternal(tempDirectory); + }); + final store = SecureConfigStore( + enableSecureStorage: false, + databasePathResolver: () async => '${tempDirectory.path}/settings.db', + fallbackDirectoryPathResolver: () async => tempDirectory.path, + ); + final gateway = FakeGatewayRuntimeInternal(store: store); + final controller = AppController( + store: store, + runtimeCoordinator: RuntimeCoordinator( + gateway: gateway, + codex: FakeCodexRuntimeInternal(), + ), + ); + addTearDown(controller.dispose); + + await waitForInternal(() => !controller.initializing); + await controller.setAssistantExecutionTarget( + AssistantExecutionTarget.local, + ); + + final localProfile = controller.settings.primaryLocalGatewayProfile; + gateway.fakeSnapshotInternal = gateway.fakeSnapshotInternal.copyWith( + status: RuntimeConnectionStatus.connected, + remoteAddress: '${localProfile.host}:${localProfile.port}', + lastError: 'NOT_PAIRED: pairing required', + lastErrorCode: 'NOT_PAIRED', + lastErrorDetailCode: 'PAIRING_REQUIRED', + ); + gateway.notifyListeners(); + await Future.delayed(Duration.zero); + + expect( + controller.currentAssistantConnectionState.pairingRequired, + isTrue, + ); + expect(controller.currentAssistantConnectionState.connected, isFalse); + expect(controller.assistantConnectionStatusLabel, '需配对'); + expect( + controller.assistantConnectionTargetLabel, + '${localProfile.host}:${localProfile.port}', + ); + }, + ); }); } diff --git a/test/runtime/gateway_runtime_suite.dart b/test/runtime/gateway_runtime_suite.dart index 19171df7..1984364d 100644 --- a/test/runtime/gateway_runtime_suite.dart +++ b/test/runtime/gateway_runtime_suite.dart @@ -297,11 +297,7 @@ void main() { addTearDown(controller.dispose); await controller.loadSession('agent:main:main'); - await controller.sendMessage( - sessionKey: 'agent:main:main', - message: 'hello', - thinking: 'low', - ); + controller.pendingRunsInternal.add('run-1'); expect(controller.hasPendingRun, isTrue); runtime.addAssistantMessage('HELLO'); @@ -585,6 +581,22 @@ void main() { ); }, ); + + test( + 'GatewayConnectionSnapshot keeps pairing-required visible even when status remains connected', + () { + final snapshot = GatewayConnectionSnapshot.initial( + mode: RuntimeConnectionMode.local, + ).copyWith( + status: RuntimeConnectionStatus.connected, + lastError: 'NOT_PAIRED: pairing required', + lastErrorCode: 'NOT_PAIRED', + lastErrorDetailCode: 'PAIRING_REQUIRED', + ); + + expect(snapshot.pairingRequired, isTrue); + }, + ); } class _FakeGatewayRuntimeSessionClient implements GatewayRuntimeSessionClient { diff --git a/test/runtime/go_task_service_desktop_service_test.dart b/test/runtime/go_task_service_desktop_service_test.dart index 677662b5..0615dea1 100644 --- a/test/runtime/go_task_service_desktop_service_test.dart +++ b/test/runtime/go_task_service_desktop_service_test.dart @@ -19,6 +19,7 @@ class _FakeGatewayRuntime extends GatewayRuntime { StreamController.broadcast(); final List> sendChatCalls = >[]; final List> abortChatCalls = >[]; + List history = const []; @override Stream get events => controller.stream; @@ -53,6 +54,14 @@ class _FakeGatewayRuntime extends GatewayRuntime { 'runId': runId, }); } + + @override + Future> loadHistory( + String sessionKey, { + int limit = 120, + }) async { + return history; + } } class DeviceIdentityStoreForTest extends DeviceIdentityStore { @@ -215,5 +224,41 @@ void main() { expect(singleResult.route, GoTaskServiceRoute.externalAcpSingle); expect(multiResult.route, GoTaskServiceRoute.externalAcpMulti); }); + + test( + 'recovers OpenClaw task completion from chat history when push events do not arrive', + () async { + final gateway = _FakeGatewayRuntime(); + final acp = _FakeExternalAcpTransport(); + final service = DesktopGoTaskService(gateway: gateway, acpTransport: acp); + + unawaited( + Future.delayed(const Duration(milliseconds: 1200), () { + gateway.history = [ + GatewayChatMessage( + id: 'assistant-1', + role: 'assistant', + text: 'RECOVERED_FROM_HISTORY', + timestampMs: 2, + toolCallId: null, + toolName: null, + stopReason: null, + pending: false, + error: false, + ), + ]; + }), + ); + + final result = await service.executeTask( + _request(target: AssistantExecutionTarget.local), + onUpdate: (_) {}, + ); + + expect(result.route, GoTaskServiceRoute.openClawTask); + expect(result.message, 'RECOVERED_FROM_HISTORY'); + expect(result.raw['recoveredFromHistory'], isTrue); + }, + ); }); }