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

Fix OpenClaw status and Go task flow

Browse Source
This commit is contained in:
Haitao Pan 2026-04-08 17:23:07 +08:00
parent 0d498f726e
commit 9d14a35b9d
7 changed files with 215 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
lib/app/app_controller_desktop_thread_sessions.dart
View File

@ -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,
);
}

82
lib/runtime/go_task_service_desktop_service.dart
View File

@ -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<GoTaskServiceResult?> _recoverOpenClawTaskFromHistory(
_PendingOpenClawTask pending,
List<GatewayChatMessage> historyBaseline,
) async {
final baselineAssistantFingerprint = _assistantMessageFingerprint(
historyBaseline,
);
final deadline = DateTime.now().add(_openClawTaskRecoveryTimeout);
while (!pending.completer.isCompleted && DateTime.now().isBefore(deadline)) {
await Future<void>.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: <String, dynamic>{
'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<GatewayChatMessage> 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<GatewayChatMessage> 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 {

77
lib/runtime/runtime_controllers_gateway.dart
View File

@ -178,8 +178,6 @@ class GatewayChatController extends ChangeNotifier {
List<GatewayChatMessage> messagesInternal = const <GatewayChatMessage>[];
String sessionKeyInternal = 'main';
bool loadingInternal = false;
bool sendingInternal = false;
bool abortingInternal = false;
String? errorInternal;
String? streamingAssistantTextInternal;
final Set<String> pendingRunsInternal = <String>{};
@ -187,8 +185,6 @@ class GatewayChatController extends ChangeNotifier {
List<GatewayChatMessage> 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<void> sendMessage({
required String sessionKey,
required String message,
required String thinking,
List<GatewayChatAttachmentPayload> attachments =
const <GatewayChatAttachmentPayload>[],
String? agentId,
Map<String, dynamic>? 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<GatewayChatMessage>.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<void> 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));

7
lib/runtime/runtime_models_runtime_payloads.dart
View File

@ -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 {

54
test/runtime/app_controller_execution_target_switch_suite_thread.dart
View File

@ -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(<String, Object>{});
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<void>.delayed(Duration.zero);
expect(
controller.currentAssistantConnectionState.pairingRequired,
isTrue,
);
expect(controller.currentAssistantConnectionState.connected, isFalse);
expect(controller.assistantConnectionStatusLabel, '需配对');
expect(
controller.assistantConnectionTargetLabel,
'${localProfile.host}:${localProfile.port}',
);
},
);
});
}

22
test/runtime/gateway_runtime_suite.dart
View File

@ -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 {

45
test/runtime/go_task_service_desktop_service_test.dart
View File

@ -19,6 +19,7 @@ class _FakeGatewayRuntime extends GatewayRuntime {
StreamController<GatewayPushEvent>.broadcast();
final List<Map<String, Object?>> sendChatCalls = <Map<String, Object?>>[];
final List<Map<String, String>> abortChatCalls = <Map<String, String>>[];
List<GatewayChatMessage> history = const <GatewayChatMessage>[];
@override
Stream<GatewayPushEvent> get events => controller.stream;
@ -53,6 +54,14 @@ class _FakeGatewayRuntime extends GatewayRuntime {
'runId': runId,
});
}
@override
Future<List<GatewayChatMessage>> 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<void>.delayed(const Duration(milliseconds: 1200), () {
gateway.history = <GatewayChatMessage>[
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);
},
);
});
}