diff --git a/lib/app/app_controller_desktop_skill_permissions.dart b/lib/app/app_controller_desktop_skill_permissions.dart index 221d9de7..180e712c 100644 --- a/lib/app/app_controller_desktop_skill_permissions.dart +++ b/lib/app/app_controller_desktop_skill_permissions.dart @@ -76,6 +76,8 @@ extension AppControllerDesktopSkillPermissions on AppController { double? lastArtifactSyncAtMs, String? lastArtifactSyncStatus, List? lastTaskArtifactRelativePaths, + OpenClawTaskAssociation? openClawTaskAssociation, + bool clearOpenClawTaskAssociation = false, }) { final normalizedSessionKey = normalizedAssistantSessionKeyInternal( sessionKey, @@ -225,6 +227,8 @@ extension AppControllerDesktopSkillPermissions on AppController { lastArtifactSyncAtMs: lastArtifactSyncAtMs, lastArtifactSyncStatus: lastArtifactSyncStatus, lastTaskArtifactRelativePaths: lastTaskArtifactRelativePaths, + openClawTaskAssociation: openClawTaskAssociation, + clearOpenClawTaskAssociation: clearOpenClawTaskAssociation, ); final nextStatus = lifecycleStatus ?? @@ -263,6 +267,7 @@ extension AppControllerDesktopSkillPermissions on AppController { executionBinding: nextExecutionBinding, contextState: nextContextState, lifecycleState: nextLifecycleState, + openClawTaskAssociation: nextContextState.openClawTaskAssociation, updatedAtMs: updatedAtMs ?? existing?.updatedAtMs ?? diff --git a/lib/app/app_controller_desktop_thread_actions.dart b/lib/app/app_controller_desktop_thread_actions.dart index 25cbdf00..3ecd40a1 100644 --- a/lib/app/app_controller_desktop_thread_actions.dart +++ b/lib/app/app_controller_desktop_thread_actions.dart @@ -3,6 +3,7 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'dart:math' as math; import 'package:flutter/material.dart'; import 'app_metadata.dart'; import 'app_capabilities.dart'; @@ -64,7 +65,11 @@ extension AppControllerDesktopThreadActions on AppController { bool assistantSessionHasPendingRun(String sessionKey) { final normalized = normalizedAssistantSessionKeyInternal(sessionKey); + final association = taskThreadForSessionInternal( + normalized, + )?.openClawTaskAssociation; return aiGatewayPendingSessionKeysInternal.contains(normalized) || + (association != null && !association.isTerminal) || openClawGatewayQueuedTurnsBySessionInternal[normalized]?.any( (turn) => !turn.cancelled, ) == @@ -558,6 +563,7 @@ extension AppControllerDesktopThreadActions on AppController { appendGatewayUserTurnInternal(sessionKey, message); } markGatewayChatRunInternal(sessionKey); + var handedOffToBridgeTask = false; try { final result = await goTaskServiceClientInternal.executeTask( GoTaskServiceRequest( @@ -590,6 +596,22 @@ extension AppControllerDesktopThreadActions on AppController { clearAiGatewayStreamingTextInternal(sessionKey); return; } + final association = result.openClawTaskAssociation; + if (association != null) { + handedOffToBridgeTask = true; + persistOpenClawTaskAssociationInternal( + sessionKey: sessionKey, + association: association, + ); + unawaited( + pollOpenClawTaskAssociationInternal( + sessionKey: sessionKey, + target: target, + association: association, + ), + ); + return; + } await applyGatewayChatResultInternal( sessionKey: sessionKey, target: target, @@ -610,13 +632,155 @@ extension AppControllerDesktopThreadActions on AppController { error: error, ); } finally { - aiGatewayPendingSessionKeysInternal.remove(sessionKey); - clearAiGatewayStreamingTextInternal(sessionKey); + if (!handedOffToBridgeTask) { + aiGatewayPendingSessionKeysInternal.remove(sessionKey); + clearAiGatewayStreamingTextInternal(sessionKey); + } recomputeTasksInternal(); notifyIfActiveInternal(); } } + void persistOpenClawTaskAssociationInternal({ + required String sessionKey, + required OpenClawTaskAssociation association, + }) { + final nowMs = DateTime.now().millisecondsSinceEpoch.toDouble(); + aiGatewayPendingSessionKeysInternal.add(sessionKey); + upsertTaskThreadInternal( + sessionKey, + lifecycleStatus: 'running', + lastResultCode: 'running', + lastRunAtMs: association.startedAtMs > 0 + ? association.startedAtMs + : nowMs, + lastArtifactSyncAtMs: nowMs, + lastArtifactSyncStatus: 'running', + openClawTaskAssociation: association, + updatedAtMs: nowMs, + ); + recomputeTasksInternal(); + notifyIfActiveInternal(); + unawaited(flushAssistantThreadPersistenceInternal()); + } + + Future pollOpenClawTaskAssociationInternal({ + required String sessionKey, + required AssistantExecutionTarget target, + required OpenClawTaskAssociation association, + }) async { + var current = association; + final pollDelay = _openClawAssociationPollDelayInternal(current); + final maxAttempts = math.max( + 1, + ((math.max(1, current.runtimeBudgetMinutes) * 60) / + math.max(1, pollDelay.inSeconds)) + .ceil(), + ); + for (var attempt = 0; attempt < maxAttempts; attempt += 1) { + if (disposedInternal) { + return; + } + if (!aiGatewayPendingSessionKeysInternal.contains(sessionKey)) { + return; + } + if (attempt > 0) { + await Future.delayed(pollDelay); + } + try { + final result = await goTaskServiceClientInternal.getTask( + route: GoTaskServiceRoute.externalAcpSingle, + target: target, + association: current, + ); + final nextAssociation = + result.openClawTaskAssociation ?? + current.copyWith( + status: result.status.trim().isEmpty + ? current.status + : result.status.trim(), + ); + current = nextAssociation; + if (result.isOpenClawRunningTaskHandle) { + persistOpenClawTaskAssociationInternal( + sessionKey: sessionKey, + association: nextAssociation, + ); + continue; + } + if (aiGatewayPendingSessionKeysInternal.contains(sessionKey)) { + await applyGatewayChatResultInternal( + sessionKey: sessionKey, + target: target, + result: result, + ); + } + aiGatewayPendingSessionKeysInternal.remove(sessionKey); + clearAiGatewayStreamingTextInternal(sessionKey); + recomputeTasksInternal(); + notifyIfActiveInternal(); + return; + } catch (_) { + continue; + } + } + final nowMs = DateTime.now().millisecondsSinceEpoch.toDouble(); + upsertTaskThreadInternal( + sessionKey, + lifecycleStatus: 'ready', + lastRunAtMs: nowMs, + lastResultCode: 'TASK_SLA_EXPIRED', + lastArtifactSyncAtMs: nowMs, + lastArtifactSyncStatus: 'failed', + openClawTaskAssociation: current.copyWith(status: 'failed'), + updatedAtMs: nowMs, + ); + aiGatewayPendingSessionKeysInternal.remove(sessionKey); + clearAiGatewayStreamingTextInternal(sessionKey); + recomputeTasksInternal(); + notifyIfActiveInternal(); + } + + Duration _openClawAssociationPollDelayInternal( + OpenClawTaskAssociation association, + ) { + final budget = association.runtimeBudgetMinutes; + if (budget >= 60) { + return const Duration(seconds: 5); + } + if (budget >= 30) { + return const Duration(seconds: 3); + } + return const Duration(seconds: 2); + } + + void resumeOpenClawTaskAssociationsInternal({String? onlySessionKey}) { + final normalizedOnly = onlySessionKey == null + ? '' + : normalizedAssistantSessionKeyInternal(onlySessionKey); + for (final record in taskThreadRepositoryInternal.snapshot()) { + if (normalizedOnly.isNotEmpty && record.threadId != normalizedOnly) { + continue; + } + final association = record.openClawTaskAssociation; + if (association == null || association.isTerminal) { + continue; + } + aiGatewayPendingSessionKeysInternal.add(record.threadId); + unawaited( + pollOpenClawTaskAssociationInternal( + sessionKey: record.threadId, + target: assistantExecutionTargetFromExecutionMode( + record.executionBinding.executionMode, + ), + association: association, + ), + ); + } + recomputeTasksInternal(); + notifyIfActiveInternal(); + } + String messageWithSelectedSkillsContextInternal({ required String message, required List selectedSkillLabels, @@ -844,6 +1008,7 @@ extension AppControllerDesktopThreadActions on AppController { lastArtifactSyncAtMs: nowMs, lastArtifactSyncStatus: 'failed', lastTaskArtifactRelativePaths: const [], + clearOpenClawTaskAssociation: true, updatedAtMs: nowMs, ); recomputeTasksInternal(); @@ -1017,6 +1182,7 @@ extension AppControllerDesktopThreadActions on AppController { lifecycleStatus: 'ready', lastRunAtMs: completedAtMs, lastResultCode: terminalResultCode, + clearOpenClawTaskAssociation: true, updatedAtMs: completedAtMs, ); if (isOpenClawNoExportedArtifactsGuardResultInternal(result)) { @@ -1102,6 +1268,7 @@ extension AppControllerDesktopThreadActions on AppController { lastArtifactSyncAtMs: completedAtMs, lastArtifactSyncStatus: 'failed', lastTaskArtifactRelativePaths: const [], + clearOpenClawTaskAssociation: true, updatedAtMs: completedAtMs, ); appendLocalSessionMessageInternal( @@ -1254,6 +1421,9 @@ extension AppControllerDesktopThreadActions on AppController { target: assistantExecutionTargetForSession(sessionKey), sessionId: sessionKey, threadId: sessionKey, + association: taskThreadForSessionInternal( + sessionKey, + )?.openClawTaskAssociation, ); } catch (_) { // Best effort cancellation only. @@ -1287,6 +1457,9 @@ extension AppControllerDesktopThreadActions on AppController { target: assistantExecutionTargetForSession(sessionKey), sessionId: sessionKey, threadId: sessionKey, + association: taskThreadForSessionInternal( + sessionKey, + )?.openClawTaskAssociation, ); } catch (_) { // Best effort cancellation only. Local state must still leave pending. diff --git a/lib/app/app_controller_desktop_thread_storage.dart b/lib/app/app_controller_desktop_thread_storage.dart index 09e6198b..5cd84db7 100644 --- a/lib/app/app_controller_desktop_thread_storage.dart +++ b/lib/app/app_controller_desktop_thread_storage.dart @@ -145,6 +145,7 @@ extension AppControllerDesktopThreadStorage on AppController { normalized, persistSelection: false, ); + resumeOpenClawTaskAssociationsInternal(onlySessionKey: normalized); } void handleRuntimeEventInternal(GatewayPushEvent event) { diff --git a/lib/features/assistant/assistant_page_state_closure.dart b/lib/features/assistant/assistant_page_state_closure.dart index 6135bdea..d6327e4e 100644 --- a/lib/features/assistant/assistant_page_state_closure.dart +++ b/lib/features/assistant/assistant_page_state_closure.dart @@ -125,11 +125,7 @@ extension AssistantPageStateClosureInternal on AssistantPageStateInternal { lastResultCode: thread?.lifecycleState.lastResultCode ?? '', artifactSyncStatus: thread?.lastArtifactSyncStatus ?? '', runtimeBudgetMinutes: - gatewayAcpTaskRuntimeBudgetMinutesForParams({ - 'taskPrompt': currentTask.preview, - 'requestedExecutionTarget': - currentTask.executionTarget.promptValue, - }), + thread?.openClawTaskAssociation?.runtimeBudgetMinutes ?? 10, ); return SurfaceCard( diff --git a/lib/runtime/external_code_agent_acp_desktop_transport.dart b/lib/runtime/external_code_agent_acp_desktop_transport.dart index e14afb6b..e8b639a4 100644 --- a/lib/runtime/external_code_agent_acp_desktop_transport.dart +++ b/lib/runtime/external_code_agent_acp_desktop_transport.dart @@ -1,6 +1,5 @@ import 'dart:async'; import 'dart:io'; -import 'dart:math' as math; import 'package:flutter/foundation.dart'; @@ -104,6 +103,7 @@ class ExternalCodeAgentAcpDesktopTransport var streamedText = ''; String? completedMessage; Map? completedResultSnapshot; + Map? runningTaskSnapshot; try { final endpointOverride = _taskEndpointResolver == null ? _endpointResolver(request.target) @@ -136,6 +136,11 @@ class ExternalCodeAgentAcpDesktopTransport update, ); } + if (update.payload['status']?.toString().trim().toLowerCase() == + 'running' && + (update.payload['runId']?.toString().trim().isNotEmpty == true)) { + runningTaskSnapshot = {...update.payload}; + } onUpdate(update); }, ); @@ -155,6 +160,7 @@ class ExternalCodeAgentAcpDesktopTransport streamedText: streamedText, completedMessage: completedMessage, fallbackAvailable: completedResultSnapshot != null, + runningTaskSnapshot: runningTaskSnapshot, ); if (recovered != null) { return recovered; @@ -203,11 +209,27 @@ class ExternalCodeAgentAcpDesktopTransport required String streamedText, required String? completedMessage, bool fallbackAvailable = false, + Map? runningTaskSnapshot, }) async { final endpoint = _sessionSnapshotEndpoint(taskEndpoint); if (endpoint == null) { return null; } + final association = OpenClawTaskAssociation.fromJsonOrNull( + runningTaskSnapshot, + ); + if (association != null) { + return goTaskServiceResultFromAcpResponse( + { + 'jsonrpc': '2.0', + 'id': 'recovered-from-running-task-handle', + 'result': runningTaskSnapshot, + }, + route: request.route, + streamedText: streamedText, + completedMessage: completedMessage, + ); + } final attempts = _recoveryAttemptsForRequest(request); for (var attempt = 0; attempt < attempts; attempt += 1) { if (attempt > 0) { @@ -216,7 +238,7 @@ class ExternalCodeAgentAcpDesktopTransport Map response; try { response = await _client.request( - method: 'xworkmate.sessions.get', + method: 'xworkmate.tasks.get', params: { 'sessionId': request.sessionId, 'threadId': request.threadId, @@ -232,11 +254,7 @@ class ExternalCodeAgentAcpDesktopTransport continue; } final snapshot = _castMap(response['result']); - final task = _castMap(snapshot['task']); - final status = (task['state'] ?? snapshot['status'] ?? '') - .toString() - .trim() - .toLowerCase(); + final status = (snapshot['status'] ?? '').toString().trim().toLowerCase(); final terminal = status == 'completed' || status == 'failed' || @@ -245,7 +263,19 @@ class ExternalCodeAgentAcpDesktopTransport if (!terminal) { continue; } - final result = _recoveredResultFromSessionSnapshot(snapshot); + if (status == 'failed' || status == 'cancelled' || status == 'canceled') { + return goTaskServiceResultFromAcpResponse( + { + 'jsonrpc': '2.0', + 'id': 'recovered-from-terminal-task-snapshot', + 'result': _failureResultFromTaskSnapshot(snapshot, status), + }, + route: request.route, + streamedText: streamedText, + completedMessage: completedMessage, + ); + } + final result = _recoveredResultFromTaskSnapshot(snapshot); if (result.isNotEmpty) { return goTaskServiceResultFromAcpResponse( { @@ -258,34 +288,37 @@ class ExternalCodeAgentAcpDesktopTransport completedMessage: completedMessage, ); } - if (status == 'failed' || status == 'cancelled' || status == 'canceled') { - return goTaskServiceResultFromAcpResponse( - { - 'jsonrpc': '2.0', - 'id': 'recovered-from-terminal-session-snapshot', - 'result': _failureResultFromSessionSnapshot(snapshot, status), - }, - route: request.route, - streamedText: streamedText, - completedMessage: completedMessage, - ); - } } return null; } + @override + Future getTask({ + required AssistantExecutionTarget target, + required OpenClawTaskAssociation association, + required GoTaskServiceRoute route, + }) async { + final endpoint = _sessionSnapshotEndpoint(_endpointResolver(target)); + if (endpoint == null) { + throw const GatewayAcpException( + 'xworkmate-bridge is not connected', + code: 'BRIDGE_NOT_CONNECTED', + ); + } + final response = await _client.request( + method: 'xworkmate.tasks.get', + params: association.toTaskGetParams(), + endpointOverride: endpoint, + ); + return goTaskServiceResultFromAcpResponse(response, route: route); + } + int _recoveryAttemptsForRequest(GoTaskServiceRequest request) { final configured = _recoveryMaxAttempts; if (configured != null) { return configured <= 0 ? 1 : configured; } - final pollMicros = math.max(1, _recoveryPollDelay.inMicroseconds); - final budgetMicros = Duration( - minutes: gatewayAcpTaskRuntimeBudgetMinutesForParams( - request.toExternalAcpParams(), - ), - ).inMicroseconds; - return math.max(1, (budgetMicros / pollMicros).ceil()); + return 1; } Uri? _sessionSnapshotEndpoint(Uri? taskEndpoint) { @@ -303,7 +336,19 @@ class ExternalCodeAgentAcpDesktopTransport required AssistantExecutionTarget target, required String sessionId, required String threadId, + OpenClawTaskAssociation? association, }) async { + if (association != null) { + final endpoint = _sessionSnapshotEndpoint(_endpointResolver(target)); + if (endpoint != null) { + await _client.request( + method: 'xworkmate.tasks.cancel', + params: association.toTaskGetParams(), + endpointOverride: endpoint, + ); + return; + } + } await _client.cancelSession( sessionId: sessionId, threadId: threadId, @@ -359,13 +404,16 @@ class ExternalCodeAgentAcpDesktopTransport return snapshot; } - Map _recoveredResultFromSessionSnapshot( + Map _recoveredResultFromTaskSnapshot( Map snapshot, ) { - final result = {..._castMap(snapshot['result'])}; + final result = { + ..._castMap(snapshot['result']), + ...snapshot, + }; final artifactRecord = _castMap(snapshot['artifacts']); - final artifactItems = _listValue(artifactRecord['items']); - if (artifactItems.isNotEmpty && !_hasArtifactList(result)) { + final artifactItems = artifactRecord['items']; + if (artifactItems is List && result['artifacts'] == artifactRecord) { result['artifacts'] = artifactItems; } for (final entry in { @@ -382,31 +430,23 @@ class ExternalCodeAgentAcpDesktopTransport return result; } - Map _failureResultFromSessionSnapshot( + Map _failureResultFromTaskSnapshot( Map snapshot, String status, ) { - final task = _castMap(snapshot['task']); final error = _castMap(snapshot['error']); final message = _firstNonEmptyDisplayText( - {...error, ...snapshot, 'taskMessage': task['message']}, - const [ - 'message', - 'error', - 'errorMessage', - 'reason', - 'taskMessage', - 'code', - ], + {...error, ...snapshot}, + const ['message', 'error', 'errorMessage', 'reason', 'code'], ); final code = _firstNonEmptyDisplayText( - {...error, ...snapshot, 'taskCode': task['code']}, - const ['code', 'errorCode', 'taskCode'], + {...error, ...snapshot}, + const ['code', 'errorCode'], ); final result = { 'success': false, 'status': status, - 'turnId': task['turnId']?.toString().trim() ?? '', + 'turnId': snapshot['turnId']?.toString().trim() ?? '', 'error': message.isNotEmpty ? message : 'Bridge session ended: $status', 'message': message.isNotEmpty ? message : 'Bridge session ended: $status', }; @@ -416,29 +456,6 @@ class ExternalCodeAgentAcpDesktopTransport return result; } - bool _hasArtifactList(Map result) { - for (final key in const ['artifacts', 'files', 'attachments']) { - if (_listValue(result[key]).isNotEmpty) { - return true; - } - final recordItems = _listValue(_castMap(result[key])['items']); - if (recordItems.isNotEmpty) { - return true; - } - } - for (final key in const ['payload', 'result', 'data']) { - final nested = _castMap(result[key]); - if (nested.isNotEmpty && _hasArtifactList(nested)) { - return true; - } - } - return false; - } - - List _listValue(Object? value) { - return value is List ? value : const []; - } - String _firstNonEmptyDisplayText( Map values, List keys, diff --git a/lib/runtime/gateway_acp_client.dart b/lib/runtime/gateway_acp_client.dart index cd27047b..16c6182f 100644 --- a/lib/runtime/gateway_acp_client.dart +++ b/lib/runtime/gateway_acp_client.dart @@ -1457,130 +1457,12 @@ class GatewayAcpRuntimeSessionClient implements GatewayRuntimeSessionClient { } } -bool _isOpenClawTaskSubmitMethod(String method) { - final normalized = method.trim(); - return normalized == 'session.start' || normalized == 'session.message'; -} - Duration gatewayAcpHttpResponseTimeoutFor( Uri endpoint, String method, [ Map params = const {}, ]) { - if (!_isOpenClawTaskSubmitMethod(method)) { - return const Duration(seconds: 120); - } - return Duration(minutes: gatewayAcpTaskRuntimeBudgetMinutesForParams(params)); -} - -int gatewayAcpTaskRuntimeBudgetMinutesForParams(Map params) { - if (_looksLikeLongArtifactTask(params)) { - return 30; - } - if (_looksLikeGatewayTask(params)) { - return 10; - } - return 2; -} - -bool _looksLikeGatewayTask(Map params) { - final target = _paramText(params, const [ - 'requestedExecutionTarget', - 'executionTarget', - ]).toLowerCase(); - if (target == AssistantExecutionTarget.gateway.promptValue) { - return true; - } - final providerText = _paramText(params, const [ - 'provider', - 'gatewayProvider', - 'preferredGatewayProviderId', - ]).toLowerCase(); - if (providerText.contains('openclaw')) { - return true; - } - final routing = params['routing']; - if (routing is Map) { - final preferred = routing['preferredGatewayProviderId'] - ?.toString() - .trim() - .toLowerCase(); - return preferred == kCanonicalGatewayProviderId || - preferred?.contains('openclaw') == true; - } - return false; -} - -bool _looksLikeLongArtifactTask(Map params) { - final prompt = _paramText(params, const [ - 'taskPrompt', - 'prompt', - 'message', - ]); - final lower = prompt.toLowerCase(); - final attachments = - _paramListLength(params['attachments']) + - _paramListLength(params['inlineAttachments']); - if (attachments >= 2 || prompt.length >= 1200) { - return true; - } - const markers = [ - '生成文件', - '同步生成文件', - '产物', - '附件', - '图片提示词', - '完整调研ppt', - 'markdown格式', - '输出markdown', - '输出 完整', - 'ppt', - 'pptx', - 'powerpoint', - 'xls', - '.xls', - 'word格式', - 'word 格式', - 'word', - 'docx', - '.docx', - 'png', - '.png', - 'mp4', - '.mp4', - 'jpg', - '.jpg', - 'markdown格式', - 'markdown', - '.md', - 'javascript', - '.js', - 'image prompt', - 'artifacts', - 'downloadurl', - ]; - return markers.any(lower.contains); -} - -String _paramText(Map params, List keys) { - for (final key in keys) { - final value = params[key]; - if (value == null) { - continue; - } - final text = value.toString().trim(); - if (text.isNotEmpty) { - return text; - } - } - return ''; -} - -int _paramListLength(Object? value) { - if (value is List) { - return value.length; - } - return 0; + return const Duration(seconds: 120); } class _GatewayAcpRpcRequest { diff --git a/lib/runtime/go_task_service_client.dart b/lib/runtime/go_task_service_client.dart index d19cad45..14621d6a 100644 --- a/lib/runtime/go_task_service_client.dart +++ b/lib/runtime/go_task_service_client.dart @@ -494,6 +494,27 @@ class GoTaskServiceResult { String get code => raw['code']?.toString().trim() ?? ''; + bool get isOpenClawRunningTaskHandle { + final normalizedStatus = status.trim().toLowerCase(); + final runId = raw['runId']?.toString().trim() ?? ''; + final artifactScope = raw['artifactScope']?.toString().trim() ?? ''; + final provider = + raw['resolvedGatewayProviderId']?.toString().trim().toLowerCase() ?? + raw['gatewayProviderId']?.toString().trim().toLowerCase() ?? + ''; + return normalizedStatus == 'running' && + runId.isNotEmpty && + artifactScope.isNotEmpty && + provider.contains('openclaw'); + } + + OpenClawTaskAssociation? get openClawTaskAssociation { + if (!isOpenClawRunningTaskHandle) { + return null; + } + return OpenClawTaskAssociation.fromJsonOrNull(raw); + } + String get resolvedExecutionTarget => raw['resolvedExecutionTarget']?.toString().trim() ?? ''; @@ -651,10 +672,17 @@ abstract class ExternalCodeAgentAcpTransport { required void Function(GoTaskServiceUpdate update) onUpdate, }); + Future getTask({ + required AssistantExecutionTarget target, + required OpenClawTaskAssociation association, + required GoTaskServiceRoute route, + }); + Future cancelTask({ required AssistantExecutionTarget target, required String sessionId, required String threadId, + OpenClawTaskAssociation? association, }); Future closeTask({ @@ -683,11 +711,18 @@ abstract class GoTaskServiceClient { required void Function(GoTaskServiceUpdate update) onUpdate, }); + Future getTask({ + required AssistantExecutionTarget target, + required OpenClawTaskAssociation association, + required GoTaskServiceRoute route, + }); + Future cancelTask({ required GoTaskServiceRoute route, required AssistantExecutionTarget target, required String sessionId, required String threadId, + OpenClawTaskAssociation? association, }); Future closeTask({ diff --git a/lib/runtime/go_task_service_desktop_service.dart b/lib/runtime/go_task_service_desktop_service.dart index 77ceaef1..cce57008 100644 --- a/lib/runtime/go_task_service_desktop_service.dart +++ b/lib/runtime/go_task_service_desktop_service.dart @@ -37,16 +37,29 @@ class DesktopGoTaskService implements GoTaskServiceClient { required void Function(GoTaskServiceUpdate update) onUpdate, }) => _acpTransport.executeTask(request, onUpdate: onUpdate); + @override + Future getTask({ + required AssistantExecutionTarget target, + required OpenClawTaskAssociation association, + required GoTaskServiceRoute route, + }) => _acpTransport.getTask( + target: target, + association: association, + route: route, + ); + @override Future cancelTask({ required GoTaskServiceRoute route, required AssistantExecutionTarget target, required String sessionId, required String threadId, + OpenClawTaskAssociation? association, }) => _acpTransport.cancelTask( target: target, sessionId: sessionId, threadId: threadId, + association: association, ); @override diff --git a/lib/runtime/runtime_models_runtime_payloads.dart b/lib/runtime/runtime_models_runtime_payloads.dart index 7c3b2212..eac01bf3 100644 --- a/lib/runtime/runtime_models_runtime_payloads.dart +++ b/lib/runtime/runtime_models_runtime_payloads.dart @@ -677,6 +677,7 @@ class ThreadContextState { this.lastArtifactSyncAtMs, this.lastArtifactSyncStatus, this.lastTaskArtifactRelativePaths = const [], + this.openClawTaskAssociation, }); final List messages; @@ -694,6 +695,7 @@ class ThreadContextState { final double? lastArtifactSyncAtMs; final String? lastArtifactSyncStatus; final List lastTaskArtifactRelativePaths; + final OpenClawTaskAssociation? openClawTaskAssociation; ThreadContextState copyWith({ List? messages, @@ -712,6 +714,8 @@ class ThreadContextState { double? lastArtifactSyncAtMs, String? lastArtifactSyncStatus, List? lastTaskArtifactRelativePaths, + OpenClawTaskAssociation? openClawTaskAssociation, + bool clearOpenClawTaskAssociation = false, }) { return ThreadContextState( messages: messages ?? this.messages, @@ -738,6 +742,9 @@ class ThreadContextState { lastTaskArtifactRelativePaths: lastTaskArtifactRelativePaths == null ? this.lastTaskArtifactRelativePaths : _stringListFromJson(lastTaskArtifactRelativePaths), + openClawTaskAssociation: clearOpenClawTaskAssociation + ? null + : (openClawTaskAssociation ?? this.openClawTaskAssociation), ); } @@ -758,6 +765,7 @@ class ThreadContextState { 'lastArtifactSyncAtMs': lastArtifactSyncAtMs, 'lastArtifactSyncStatus': lastArtifactSyncStatus, 'lastTaskArtifactRelativePaths': lastTaskArtifactRelativePaths, + 'openClawTaskAssociation': openClawTaskAssociation?.toJson(), }; } @@ -823,6 +831,163 @@ class ThreadContextState { lastTaskArtifactRelativePaths: _stringListFromJson( json['lastTaskArtifactRelativePaths'], ), + openClawTaskAssociation: OpenClawTaskAssociation.fromJsonOrNull( + json['openClawTaskAssociation'], + ), + ); + } +} + +class OpenClawTaskAssociation { + const OpenClawTaskAssociation({ + required this.sessionId, + required this.threadId, + required this.turnId, + required this.runId, + required this.artifactScope, + required this.artifactDirectory, + required this.gatewayProviderId, + required this.runtimeBudgetMinutes, + required this.startedAtMs, + required this.status, + this.taskLoadClass = '', + this.sessionKey = '', + this.requiredArtifactExtensions = const [], + this.expectedArtifactExtensions = const [], + }); + + final String sessionId; + final String threadId; + final String turnId; + final String runId; + final String artifactScope; + final String artifactDirectory; + final String gatewayProviderId; + final int runtimeBudgetMinutes; + final double startedAtMs; + final String status; + final String taskLoadClass; + final String sessionKey; + final List requiredArtifactExtensions; + final List expectedArtifactExtensions; + + bool get isTerminal { + final normalized = status.trim().toLowerCase(); + return normalized == 'completed' || + normalized == 'failed' || + normalized == 'cancelled' || + normalized == 'canceled'; + } + + OpenClawTaskAssociation copyWith({String? status}) { + return OpenClawTaskAssociation( + sessionId: sessionId, + threadId: threadId, + turnId: turnId, + runId: runId, + artifactScope: artifactScope, + artifactDirectory: artifactDirectory, + gatewayProviderId: gatewayProviderId, + runtimeBudgetMinutes: runtimeBudgetMinutes, + startedAtMs: startedAtMs, + status: status ?? this.status, + taskLoadClass: taskLoadClass, + sessionKey: sessionKey, + requiredArtifactExtensions: requiredArtifactExtensions, + expectedArtifactExtensions: expectedArtifactExtensions, + ); + } + + Map toJson() { + return { + 'sessionId': sessionId, + 'threadId': threadId, + 'turnId': turnId, + 'runId': runId, + 'artifactScope': artifactScope, + 'artifactDirectory': artifactDirectory, + 'gatewayProviderId': gatewayProviderId, + 'runtimeBudgetMinutes': runtimeBudgetMinutes, + 'startedAtMs': startedAtMs, + 'status': status, + 'taskLoadClass': taskLoadClass, + 'sessionKey': sessionKey, + 'requiredArtifactExtensions': requiredArtifactExtensions, + 'expectedArtifactExtensions': expectedArtifactExtensions, + }; + } + + Map toTaskGetParams() { + return { + 'sessionId': sessionId, + 'threadId': threadId, + 'turnId': turnId, + 'runId': runId, + 'artifactScope': artifactScope, + 'artifactDirectory': artifactDirectory, + 'gatewayProviderId': gatewayProviderId, + 'runtimeBudgetMinutes': runtimeBudgetMinutes, + 'taskLoadClass': taskLoadClass, + 'sessionKey': sessionKey, + 'requiredArtifactExtensions': requiredArtifactExtensions, + 'expectedArtifactExtensions': expectedArtifactExtensions, + }; + } + + static OpenClawTaskAssociation? fromJsonOrNull(Object? value) { + if (value is! Map) { + return null; + } + final json = value.cast(); + final runId = json['runId']?.toString().trim() ?? ''; + final artifactScope = json['artifactScope']?.toString().trim() ?? ''; + if (runId.isEmpty || artifactScope.isEmpty) { + return null; + } + int asInt(Object? raw) { + if (raw is int) { + return raw; + } + if (raw is num) { + return raw.toInt(); + } + return int.tryParse(raw?.toString() ?? '') ?? 60; + } + + double asDouble(Object? raw) { + if (raw is num) { + return raw.toDouble(); + } + return double.tryParse(raw?.toString() ?? '') ?? 0; + } + + return OpenClawTaskAssociation( + sessionId: json['sessionId']?.toString().trim() ?? '', + threadId: json['threadId']?.toString().trim() ?? '', + turnId: json['turnId']?.toString().trim() ?? '', + runId: runId, + artifactScope: artifactScope, + artifactDirectory: json['artifactDirectory']?.toString().trim() ?? '', + gatewayProviderId: + json['gatewayProviderId']?.toString().trim().isNotEmpty == true + ? json['gatewayProviderId'].toString().trim() + : (json['resolvedGatewayProviderId']?.toString().trim().isNotEmpty == + true + ? json['resolvedGatewayProviderId'].toString().trim() + : 'openclaw'), + runtimeBudgetMinutes: asInt(json['runtimeBudgetMinutes']), + startedAtMs: asDouble(json['startedAtMs']), + status: json['status']?.toString().trim().isNotEmpty == true + ? json['status'].toString().trim() + : 'running', + taskLoadClass: json['taskLoadClass']?.toString().trim() ?? '', + sessionKey: json['sessionKey']?.toString().trim() ?? '', + requiredArtifactExtensions: _stringListFromJson( + json['requiredArtifactExtensions'], + ), + expectedArtifactExtensions: _stringListFromJson( + json['expectedArtifactExtensions'], + ), ); } } @@ -931,6 +1096,7 @@ class TaskThread { double? lastArtifactSyncAtMs, String? lastArtifactSyncStatus, List? lastTaskArtifactRelativePaths, + OpenClawTaskAssociation? openClawTaskAssociation, }) : threadId = _resolveThreadId(threadId), title = title ?? '', ownerScope = @@ -977,6 +1143,7 @@ class TaskThread { lastTaskArtifactRelativePaths: _stringListFromJson( lastTaskArtifactRelativePaths, ), + openClawTaskAssociation: openClawTaskAssociation, ), lifecycleState = lifecycleState ?? @@ -1015,6 +1182,8 @@ class TaskThread { String? get lastArtifactSyncStatus => contextState.lastArtifactSyncStatus; List get lastTaskArtifactRelativePaths => contextState.lastTaskArtifactRelativePaths; + OpenClawTaskAssociation? get openClawTaskAssociation => + contextState.openClawTaskAssociation; String get latestResolvedRuntimeModel => contextState.latestResolvedRuntimeModel; String get latestResolvedProviderId => contextState.latestResolvedProviderId; @@ -1058,6 +1227,8 @@ class TaskThread { double? lastArtifactSyncAtMs, String? lastArtifactSyncStatus, List? lastTaskArtifactRelativePaths, + OpenClawTaskAssociation? openClawTaskAssociation, + bool clearOpenClawTaskAssociation = false, }) { return TaskThread( threadId: threadId ?? this.threadId, @@ -1081,6 +1252,8 @@ class TaskThread { lastArtifactSyncAtMs: lastArtifactSyncAtMs, lastArtifactSyncStatus: lastArtifactSyncStatus, lastTaskArtifactRelativePaths: lastTaskArtifactRelativePaths, + openClawTaskAssociation: openClawTaskAssociation, + clearOpenClawTaskAssociation: clearOpenClawTaskAssociation, ), lifecycleState: (lifecycleState ?? this.lifecycleState).copyWith( archived: archived, @@ -1197,6 +1370,7 @@ class TaskThread { 'lastArtifactSyncAtMs': json['lastArtifactSyncAtMs'], 'lastArtifactSyncStatus': json['lastArtifactSyncStatus'], 'lastTaskArtifactRelativePaths': json['lastTaskArtifactRelativePaths'], + 'openClawTaskAssociation': json['openClawTaskAssociation'], }; } diff --git a/test/runtime/assistant_execution_target_test.dart b/test/runtime/assistant_execution_target_test.dart index 983cdd2e..5e9cefef 100644 --- a/test/runtime/assistant_execution_target_test.dart +++ b/test/runtime/assistant_execution_target_test.dart @@ -3615,7 +3615,7 @@ void main() { 'openclaw-second-task', ); await secondSubmitFuture; - expect(controller.openClawGatewayActiveTasksInternal, 0); + await _waitForOpenClawActiveTaskCount(controller, 0); expect( controller.chatMessages.map((message) => message.text), contains('second task completed'), @@ -4083,6 +4083,20 @@ Future _selectGatewaySession( ); } +Future _waitForOpenClawActiveTaskCount( + AppController controller, + int expected, +) async { + final deadline = DateTime.now().add(const Duration(seconds: 5)); + while (DateTime.now().isBefore(deadline)) { + if (controller.openClawGatewayActiveTasksInternal == expected) { + return; + } + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(controller.openClawGatewayActiveTasksInternal, expected); +} + Future> _startOpenClawActiveTasks( AppController controller, _BlockingGoTaskServiceClient fakeGoTaskService, { @@ -4217,12 +4231,36 @@ class _RecordingGoTaskServiceClient implements GoTaskServiceClient { ); } + @override + Future getTask({ + required AssistantExecutionTarget target, + required OpenClawTaskAssociation association, + required GoTaskServiceRoute route, + }) async { + return GoTaskServiceResult( + success: true, + message: 'ok', + turnId: association.turnId, + raw: { + 'success': true, + 'status': 'completed', + 'turnId': association.turnId, + 'runId': association.runId, + 'output': 'ok', + }, + errorMessage: '', + resolvedModel: '', + route: route, + ); + } + @override Future cancelTask({ required GoTaskServiceRoute route, required AssistantExecutionTarget target, required String sessionId, required String threadId, + OpenClawTaskAssociation? association, }) async {} @override @@ -4275,6 +4313,29 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient { return completer.future; } + @override + Future getTask({ + required AssistantExecutionTarget target, + required OpenClawTaskAssociation association, + required GoTaskServiceRoute route, + }) async { + return GoTaskServiceResult( + success: true, + message: 'cleanup', + turnId: association.turnId, + raw: { + 'success': true, + 'status': 'completed', + 'turnId': association.turnId, + 'runId': association.runId, + 'output': 'cleanup', + }, + errorMessage: '', + resolvedModel: '', + route: route, + ); + } + Future waitForRequestCount(int count) async { final deadline = DateTime.now().add(const Duration(seconds: 15)); while (requests.length < count && DateTime.now().isBefore(deadline)) { @@ -4338,6 +4399,7 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient { required AssistantExecutionTarget target, required String sessionId, required String threadId, + OpenClawTaskAssociation? association, }) async { cancelledSessionIds.add(sessionId); } diff --git a/test/runtime/gateway_acp_client_auth_test.dart b/test/runtime/gateway_acp_client_auth_test.dart index 0ae9b1ab..04ffd22b 100644 --- a/test/runtime/gateway_acp_client_auth_test.dart +++ b/test/runtime/gateway_acp_client_auth_test.dart @@ -253,38 +253,6 @@ void main() { }); }); - group('Gateway ACP task runtime budget', () { - test( - 'treats explicit document and media formats as long artifact tasks', - () { - expect( - gatewayAcpTaskRuntimeBudgetMinutesForParams(const { - 'taskPrompt': '围绕上述话题输出一篇800-1000字左右的文章 markdown格式', - }), - 30, - ); - expect( - gatewayAcpTaskRuntimeBudgetMinutesForParams(const { - 'taskPrompt': 'save the final article as article.docx', - }), - 30, - ); - expect( - gatewayAcpTaskRuntimeBudgetMinutesForParams(const { - 'taskPrompt': '生成封面图 png 和短视频 mp4', - }), - 30, - ); - expect( - gatewayAcpTaskRuntimeBudgetMinutesForParams(const { - 'taskPrompt': '导出 jpg、pptx 和 xls 文件', - }), - 30, - ); - }, - ); - }); - group('GatewayAcpClient authorization', () { test('normalizes raw resolver token into bearer header for HTTP', () async { final capture = await _startAcpHttpServer(); @@ -672,7 +640,7 @@ void main() { ); test( - 'recovers OpenClaw task result from bridge session snapshot after SSE connection close', + 'recovers OpenClaw task result from bridge task snapshot after SSE connection close', () async { final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); addTearDown(() => server.close(force: true)); @@ -701,7 +669,7 @@ void main() { socket.destroy(); return; } - if (method == 'xworkmate.sessions.get') { + if (method == 'xworkmate.tasks.get') { request.response.headers.contentType = ContentType.json; request.response.write( jsonEncode({ @@ -717,7 +685,7 @@ void main() { }, 'result': { 'success': true, - 'output': 'recovered from bridge session snapshot', + 'output': 'recovered from bridge task snapshot', 'turnId': 'turn-recovered', }, 'artifacts': { @@ -772,7 +740,7 @@ void main() { ); expect(result.success, isTrue); - expect(result.message, 'recovered from bridge session snapshot'); + expect(result.message, 'recovered from bridge task snapshot'); expect(result.artifacts.single.relativePath, 'exports/snapshot.md'); expect(result.remoteWorkingDirectory, '/remote/openclaw/workspace'); expect(result.remoteWorkspaceRefKind, WorkspaceRefKind.remotePath); @@ -818,7 +786,7 @@ void main() { socket.destroy(); return; } - if (method == 'xworkmate.sessions.get') { + if (method == 'xworkmate.tasks.get') { request.response.headers.contentType = ContentType.json; request.response.write( jsonEncode({ @@ -890,13 +858,13 @@ void main() { expect(result.artifacts.single.relativePath, 'reports/final.md'); expect(requestMethods, [ 'session.start', - 'xworkmate.sessions.get', + 'xworkmate.tasks.get', ]); }, ); test( - 'recovers OpenClaw follow-up from bridge session snapshot after SSE ends without final envelope', + 'recovers OpenClaw follow-up from bridge task snapshot after SSE ends without final envelope', () async { final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); addTearDown(() => server.close(force: true)); @@ -921,7 +889,7 @@ void main() { await request.response.close(); return; } - if (method == 'xworkmate.sessions.get') { + if (method == 'xworkmate.tasks.get') { request.response.headers.contentType = ContentType.json; request.response.write( jsonEncode({ @@ -1014,7 +982,7 @@ void main() { socket.destroy(); return; } - if (method == 'xworkmate.sessions.get') { + if (method == 'xworkmate.tasks.get') { snapshotPolls += 1; final completed = snapshotPolls >= 3; request.response.headers.contentType = ContentType.json; @@ -1080,96 +1048,6 @@ void main() { }, ); - test( - 'uses long task budget for default OpenClaw SSE recovery polling', - () async { - final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); - addTearDown(() => server.close(force: true)); - var snapshotPolls = 0; - server.listen((request) async { - final body = await utf8.decoder.bind(request).join(); - final decoded = jsonDecode(body) as Map; - final method = decoded['method']?.toString() ?? ''; - final id = decoded['id']?.toString() ?? 'request-id'; - if (method == 'session.start') { - final event = jsonEncode({ - 'jsonrpc': '2.0', - 'method': 'xworkmate.bridge.accepted', - 'params': {'sessionId': 'unit-fixture-task-d'}, - }); - request.response.headers.set( - HttpHeaders.contentTypeHeader, - 'text/event-stream', - ); - request.response.write('data: $event\n\n'); - await request.response.close(); - return; - } - if (method == 'xworkmate.sessions.get') { - snapshotPolls += 1; - final completed = snapshotPolls >= 301; - request.response.headers.contentType = ContentType.json; - request.response.write( - jsonEncode({ - 'jsonrpc': '2.0', - 'id': id, - 'result': { - 'status': completed ? 'completed' : 'running', - 'sessionId': 'unit-fixture-task-d', - 'threadId': 'unit-fixture-task-d', - 'task': { - 'state': completed ? 'completed' : 'running', - 'turnId': 'turn-recovered-long', - }, - if (completed) - 'result': { - 'success': true, - 'output': 'recovered after long polling window', - 'turnId': 'turn-recovered-long', - }, - }, - }), - ); - await request.response.close(); - return; - } - request.response.statusCode = HttpStatus.badRequest; - await request.response.close(); - }); - final endpoint = Uri.parse('http://127.0.0.1:${server.port}'); - final transport = ExternalCodeAgentAcpDesktopTransport( - client: GatewayAcpClient(endpointResolver: () => endpoint), - endpointResolver: (_) => endpoint, - taskEndpointResolver: (_) => endpoint, - recoveryPollDelay: const Duration(microseconds: 1), - ); - addTearDown(transport.dispose); - - final result = await transport.executeTask( - const GoTaskServiceRequest( - sessionId: 'unit-fixture-task-d', - threadId: 'unit-fixture-task-d', - target: AssistantExecutionTarget.gateway, - provider: SingleAgentProvider.openclaw, - prompt: '生成封面图 png 和短视频 mp4', - workingDirectory: '/tmp/workspace', - model: '', - thinking: 'off', - selectedSkills: [], - inlineAttachments: [], - localAttachments: [], - agentId: '', - metadata: {}, - ), - onUpdate: (_) {}, - ); - - expect(snapshotPolls, 301); - expect(result.success, isTrue); - expect(result.message, 'recovered after long polling window'); - }, - ); - test( 'recovers terminal failed OpenClaw snapshot without displayable result', () async { @@ -1198,7 +1076,7 @@ void main() { socket.destroy(); return; } - if (method == 'xworkmate.sessions.get') { + if (method == 'xworkmate.tasks.get') { request.response.headers.contentType = ContentType.json; request.response.write( jsonEncode({ @@ -1957,7 +1835,7 @@ void main() { }, ); - test('task submit uses dynamic HTTP response timeout budgets', () { + test('task submit uses short HTTP response timeout', () { final openClawEndpoint = Uri.parse( 'https://xworkmate-bridge.svc.plus/acp/rpc', ); @@ -1971,7 +1849,7 @@ void main() { 'session.start', const {'requestedExecutionTarget': 'gateway'}, ), - const Duration(minutes: 10), + const Duration(seconds: 120), ); expect( gatewayAcpHttpResponseTimeoutFor( @@ -1982,11 +1860,11 @@ void main() { 'requestedExecutionTarget': 'gateway', }, ), - const Duration(minutes: 30), + const Duration(seconds: 120), ); expect( gatewayAcpHttpResponseTimeoutFor(acpEndpoint, 'session.start'), - const Duration(minutes: 2), + const Duration(seconds: 120), ); expect( gatewayAcpHttpResponseTimeoutFor(openClawEndpoint, 'acp.capabilities'),