diff --git a/lib/app/app_controller_desktop_core.dart b/lib/app/app_controller_desktop_core.dart index 1596cf3b..97b8289d 100644 --- a/lib/app/app_controller_desktop_core.dart +++ b/lib/app/app_controller_desktop_core.dart @@ -31,7 +31,7 @@ import '../runtime/assistant_artifacts.dart'; import '../runtime/desktop_thread_artifact_service.dart'; import '../runtime/external_code_agent_acp_desktop_transport.dart'; import '../runtime/go_task_service_client.dart'; -import '../runtime/go_task_service_desktop_service.dart'; + import '../runtime/go_runtime_dispatch_desktop_client.dart'; import '../runtime/mode_switcher.dart'; import '../runtime/agent_registry.dart'; @@ -152,13 +152,10 @@ class AppController extends ChangeNotifier { ); goTaskServiceClientInternal = goTaskServiceClient ?? - DesktopGoTaskService( - gateway: runtimeCoordinatorInternal.gateway, - acpTransport: ExternalCodeAgentAcpDesktopTransport( - client: gatewayAcpClientInternal, - endpointResolver: resolveExternalAcpEndpointForTargetInternal, - taskEndpointResolver: resolveExternalAcpEndpointForRequestInternal, - ), + ExternalCodeAgentAcpDesktopTransport( + client: gatewayAcpClientInternal, + endpointResolver: resolveExternalAcpEndpointForTargetInternal, + taskEndpointResolver: resolveExternalAcpEndpointForRequestInternal, ); bridgeAgentProviderCatalogInternal = normalizeSingleAgentProviderList( initialBridgeProviderCatalog ?? const [], @@ -267,8 +264,7 @@ class AppController extends ChangeNotifier { final Map openClawGatewayActiveTurnsInternal = {}; - int get openClawGatewayActiveTasksInternal => - openClawGatewayActiveTurnsInternal.length; + int localMessageCounterInternal = 0; int assistantDraftSessionCounterInternal = 0; diff --git a/lib/app/app_controller_desktop_runtime_helpers.dart b/lib/app/app_controller_desktop_runtime_helpers.dart index 381ad308..30155446 100644 --- a/lib/app/app_controller_desktop_runtime_helpers.dart +++ b/lib/app/app_controller_desktop_runtime_helpers.dart @@ -939,7 +939,7 @@ extension AppControllerDesktopRuntimeHelpers on AppController { final authorization = await resolveBridgeArtifactAuthorizationHeaderInternal(uri); if (authorization == null || authorization.trim().isEmpty) { - return const _ArtifactBytesResult.skipped(); + return const _ArtifactBytesResult.failed(); } final bytes = await _downloadBridgeArtifactBytesInternal( uri, diff --git a/lib/app/app_controller_desktop_thread_actions.dart b/lib/app/app_controller_desktop_thread_actions.dart index 57a34024..43caac3e 100644 --- a/lib/app/app_controller_desktop_thread_actions.dart +++ b/lib/app/app_controller_desktop_thread_actions.dart @@ -1296,19 +1296,7 @@ extension AppControllerDesktopThreadActions on AppController { notifyIfActiveInternal(); } - void clearGatewayTaskArtifactStateInternal( - String sessionKey, { - required double completedAtMs, - required String syncStatus, - }) { - upsertTaskThreadInternal( - sessionKey, - lastArtifactSyncAtMs: completedAtMs, - lastArtifactSyncStatus: syncStatus, - lastTaskArtifactRelativePaths: const [], - updatedAtMs: completedAtMs, - ); - } + Future applyGatewayChatResultInternal({ required String sessionKey, @@ -1347,10 +1335,12 @@ extension AppControllerDesktopThreadActions on AppController { return; } if (!result.success) { - clearGatewayTaskArtifactStateInternal( + upsertTaskThreadInternal( sessionKey, - completedAtMs: completedAtMs, - syncStatus: 'failed', + lastArtifactSyncAtMs: completedAtMs, + lastArtifactSyncStatus: 'failed', + lastTaskArtifactRelativePaths: const [], + updatedAtMs: completedAtMs, ); appendLocalSessionMessageInternal( sessionKey, @@ -1370,10 +1360,12 @@ extension AppControllerDesktopThreadActions on AppController { return; } if (noDisplayableOutput) { - clearGatewayTaskArtifactStateInternal( + upsertTaskThreadInternal( sessionKey, - completedAtMs: completedAtMs, - syncStatus: 'failed', + lastArtifactSyncAtMs: completedAtMs, + lastArtifactSyncStatus: 'failed', + lastTaskArtifactRelativePaths: const [], + updatedAtMs: completedAtMs, ); appendLocalSessionMessageInternal( sessionKey, @@ -1554,7 +1546,10 @@ extension AppControllerDesktopThreadActions on AppController { bool gatewayResultCodeRequiresNewSessionInternal(String code) { final normalized = code.trim().toUpperCase(); - if (normalized.isEmpty) { + if (normalized.isEmpty || + normalized == 'SUCCESS' || + normalized == 'COMPLETED' || + normalized == 'READY') { return false; } if (normalized == 'RUNNING' || @@ -1566,6 +1561,7 @@ extension AppControllerDesktopThreadActions on AppController { normalized == 'BRIDGE_NOT_CONNECTED' || normalized == 'ACP_HTTP_401' || normalized == 'ACP_HTTP_403' || + normalized == 'OPENCLAW_GATEWAY_SOCKET_CLOSED' || normalized == 'OPENCLAW_GATEWAY_QUEUE_FULL' || normalized == 'OPENCLAW_AGENT_FAILED_BEFORE_REPLY' || normalized == 'OPENCLAW_NO_DISPLAYABLE_OUTPUT' || @@ -1574,7 +1570,7 @@ extension AppControllerDesktopThreadActions on AppController { normalized == 'ARTIFACT_MISSING') { return true; } - return false; + return true; } Future abortRun() async { diff --git a/lib/runtime/external_code_agent_acp_desktop_transport.dart b/lib/runtime/external_code_agent_acp_desktop_transport.dart index bcdc0ba4..c3790d2f 100644 --- a/lib/runtime/external_code_agent_acp_desktop_transport.dart +++ b/lib/runtime/external_code_agent_acp_desktop_transport.dart @@ -1,15 +1,13 @@ import 'dart:async'; import 'dart:io'; -import 'package:flutter/foundation.dart'; - import 'acp_endpoint_paths.dart'; import 'gateway_acp_client.dart'; import 'go_task_service_client.dart'; import 'runtime_models.dart'; class ExternalCodeAgentAcpDesktopTransport - implements ExternalCodeAgentAcpTransport { + implements GoTaskServiceClient { ExternalCodeAgentAcpDesktopTransport({ required GatewayAcpClient client, required Uri? Function(AssistantExecutionTarget target) endpointResolver, @@ -28,72 +26,7 @@ class ExternalCodeAgentAcpDesktopTransport final Duration _recoveryPollDelay; final int? _recoveryMaxAttempts; - @visibleForTesting - GatewayAcpClient get clientForTest => _client; - @override - Future loadExternalAcpCapabilities({ - required AssistantExecutionTarget target, - bool forceRefresh = false, - }) async { - final response = await _client.request( - method: 'acp.capabilities', - params: const {}, - endpointOverride: _endpointResolver(target), - ); - final result = _castMap(response['result']); - final caps = _castMap(result['capabilities']); - final providerCatalog = _parseProviderCatalog( - result['providerCatalog'] ?? caps['providerCatalog'], - defaultTarget: AssistantExecutionTarget.agent, - ); - final gatewayProviders = _parseProviderCatalog( - result['gatewayProviders'] ?? caps['gatewayProviders'], - defaultTarget: AssistantExecutionTarget.gateway, - ); - return ExternalCodeAgentAcpCapabilities( - singleAgent: - _boolValue(result['singleAgent']) ?? - _boolValue(caps['single_agent']) ?? - providerCatalog.isNotEmpty, - multiAgent: - _boolValue(result['multiAgent']) ?? - _boolValue(caps['multi_agent']) ?? - true, - availableExecutionTargets: _parseAvailableExecutionTargets( - result['availableExecutionTargets'] ?? - caps['availableExecutionTargets'], - singleAgent: - _boolValue(result['singleAgent']) ?? - _boolValue(caps['single_agent']) ?? - providerCatalog.isNotEmpty, - gatewayProviders: gatewayProviders, - ), - providerCatalog: providerCatalog, - gatewayProviders: gatewayProviders, - raw: result, - ); - } - - @override - Future resolveExternalAcpRouting({ - required String taskPrompt, - required String workingDirectory, - required ExternalCodeAgentAcpRoutingConfig routing, - }) async { - final response = await _client.request( - method: 'xworkmate.routing.resolve', - params: { - 'taskPrompt': taskPrompt, - 'workingDirectory': workingDirectory.trim(), - 'routing': routing.toJson(), - }, - endpointOverride: _endpointResolver(AssistantExecutionTarget.gateway), - ); - return ExternalCodeAgentAcpRoutingResolution( - raw: _castMap(response['result']), - ); - } @override Future executeTask( @@ -330,6 +263,7 @@ class ExternalCodeAgentAcpDesktopTransport @override Future cancelTask({ + required GoTaskServiceRoute route, required AssistantExecutionTarget target, required String sessionId, required String threadId, @@ -353,18 +287,7 @@ class ExternalCodeAgentAcpDesktopTransport ); } - @override - Future closeTask({ - required AssistantExecutionTarget target, - required String sessionId, - required String threadId, - }) async { - await _client.closeSession( - sessionId: sessionId, - threadId: threadId, - endpointOverride: _endpointResolver(target), - ); - } + @override Future dispose() => _client.dispose(); @@ -531,130 +454,6 @@ class ExternalCodeAgentAcpDesktopTransport return const {}; } - List _asList(Object? raw) { - if (raw is List) { - return raw; - } - if (raw is List) { - return raw.cast(); - } - return const []; - } - - bool? _boolValue(Object? raw) { - if (raw is bool) { - return raw; - } - if (raw is num) { - return raw != 0; - } - final text = raw?.toString().trim().toLowerCase(); - if (text == null || text.isEmpty) { - return null; - } - if (text == 'true' || text == '1' || text == 'yes') { - return true; - } - if (text == 'false' || text == '0' || text == 'no') { - return false; - } - return null; - } - - List _parseProviderCatalog( - Object? raw, { - required AssistantExecutionTarget defaultTarget, - }) { - final providers = []; - for (final item in _asList(raw)) { - final entry = _castMap(item); - final providerId = entry['providerId']?.toString().trim() ?? ''; - if (providerId.isEmpty) { - continue; - } - final label = entry['label']?.toString().trim(); - final providerDisplay = _castMap(entry['providerDisplay']); - final targets = _parseProviderTargets( - entry['targets'] ?? entry['executionTarget'], - defaultTarget: defaultTarget, - ); - final provider = SingleAgentProviderCopy.fromJsonValue( - providerId, - label: label?.isNotEmpty == true ? label : null, - badge: entry['badge']?.toString().trim().isNotEmpty == true - ? entry['badge']?.toString().trim() - : providerDisplay['badge']?.toString().trim(), - logoEmoji: entry['logoEmoji']?.toString().trim().isNotEmpty == true - ? entry['logoEmoji']?.toString().trim() - : providerDisplay['logoEmoji']?.toString().trim(), - supportedTargets: targets, - enabled: _boolValue(entry['enabled']) ?? true, - unavailableReason: - entry['unavailableReason']?.toString().trim().isNotEmpty == true - ? entry['unavailableReason']?.toString().trim() - : '', - ); - if (!provider.isUnspecified) { - providers.add(provider); - } - } - return normalizeSingleAgentProviderList(providers); - } - - List _parseAvailableExecutionTargets( - Object? raw, { - required bool singleAgent, - required List gatewayProviders, - }) { - final parsed = []; - for (final item in _asList(raw)) { - final normalized = item?.toString().trim().toLowerCase() ?? ''; - if (normalized == 'agent' || normalized == 'single-agent') { - if (!parsed.contains(AssistantExecutionTarget.agent)) { - parsed.add(AssistantExecutionTarget.agent); - } - } else if (normalized == 'gateway') { - if (!parsed.contains(AssistantExecutionTarget.gateway)) { - parsed.add(AssistantExecutionTarget.gateway); - } - } - } - if (parsed.isNotEmpty) { - return parsed; - } - if (singleAgent) { - parsed.add(AssistantExecutionTarget.agent); - } - if (gatewayProviders.isNotEmpty) { - parsed.add(AssistantExecutionTarget.gateway); - } - return parsed; - } - - List _parseProviderTargets( - Object? raw, { - required AssistantExecutionTarget defaultTarget, - }) { - final parsed = []; - final items = raw is List ? raw : [raw]; - for (final item in items) { - final normalized = item?.toString().trim().toLowerCase() ?? ''; - if (normalized == 'agent' || normalized == 'single-agent') { - if (!parsed.contains(AssistantExecutionTarget.agent)) { - parsed.add(AssistantExecutionTarget.agent); - } - } else if (normalized == 'gateway') { - if (!parsed.contains(AssistantExecutionTarget.gateway)) { - parsed.add(AssistantExecutionTarget.gateway); - } - } - } - if (parsed.isNotEmpty) { - return parsed; - } - return [defaultTarget]; - } - bool _socketExceptionLooksLikeConnectTimeout(SocketException error) { final lowered = error.toString().toLowerCase(); return lowered.contains('connection timed out') || diff --git a/lib/runtime/gateway_runtime_core.dart b/lib/runtime/gateway_runtime_core.dart index 16e9e9ce..3856d1c0 100644 --- a/lib/runtime/gateway_runtime_core.dart +++ b/lib/runtime/gateway_runtime_core.dart @@ -88,21 +88,7 @@ class GatewayRuntime extends ChangeNotifier with GatewayRuntimeHelpersInternal { notifyListeners(); } - @visibleForTesting - void addRuntimeLogForTest({ - required String level, - required String category, - required String message, - }) { - appendLogInternal(this, level, category, message); - } - @visibleForTesting - bool get usesSessionClient => sessionClientInternal != null; - - @visibleForTesting - GatewayRuntimeSessionClient? get sessionClientForTest => - sessionClientInternal; bool get canConnectBridgeSession => sessionClientInternal != null; diff --git a/lib/runtime/go_task_service_client.dart b/lib/runtime/go_task_service_client.dart index f5e7b63c..4f23b96c 100644 --- a/lib/runtime/go_task_service_client.dart +++ b/lib/runtime/go_task_service_client.dart @@ -680,57 +680,7 @@ String? goTaskServiceGatewayEntryState({ } } -abstract class ExternalCodeAgentAcpTransport { - Future loadExternalAcpCapabilities({ - required AssistantExecutionTarget target, - bool forceRefresh = false, - }); - - Future resolveExternalAcpRouting({ - required String taskPrompt, - required String workingDirectory, - required ExternalCodeAgentAcpRoutingConfig routing, - }); - - Future executeTask( - GoTaskServiceRequest request, { - 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({ - required AssistantExecutionTarget target, - required String sessionId, - required String threadId, - }); - - Future dispose(); -} - abstract class GoTaskServiceClient { - Future loadExternalAcpCapabilities({ - required AssistantExecutionTarget target, - bool forceRefresh = false, - }); - - Future resolveExternalAcpRouting({ - required String taskPrompt, - required String workingDirectory, - required ExternalCodeAgentAcpRoutingConfig routing, - }); - Future executeTask( GoTaskServiceRequest request, { required void Function(GoTaskServiceUpdate update) onUpdate, @@ -750,13 +700,6 @@ abstract class GoTaskServiceClient { OpenClawTaskAssociation? association, }); - Future closeTask({ - required GoTaskServiceRoute route, - required AssistantExecutionTarget target, - required String sessionId, - required String threadId, - }); - Future dispose(); } diff --git a/lib/runtime/go_task_service_desktop_service.dart b/lib/runtime/go_task_service_desktop_service.dart deleted file mode 100644 index cce57008..00000000 --- a/lib/runtime/go_task_service_desktop_service.dart +++ /dev/null @@ -1,84 +0,0 @@ -import 'gateway_runtime.dart'; -import 'go_task_service_client.dart'; -import 'runtime_models.dart'; -import 'package:flutter/foundation.dart'; - -class DesktopGoTaskService implements GoTaskServiceClient { - DesktopGoTaskService({ - required GatewayRuntime gateway, - required ExternalCodeAgentAcpTransport acpTransport, - }) : _acpTransport = acpTransport; - - final ExternalCodeAgentAcpTransport _acpTransport; - - @override - Future loadExternalAcpCapabilities({ - required AssistantExecutionTarget target, - bool forceRefresh = false, - }) => _acpTransport.loadExternalAcpCapabilities( - target: target, - forceRefresh: forceRefresh, - ); - - @override - Future resolveExternalAcpRouting({ - required String taskPrompt, - required String workingDirectory, - required ExternalCodeAgentAcpRoutingConfig routing, - }) => _acpTransport.resolveExternalAcpRouting( - taskPrompt: taskPrompt, - workingDirectory: workingDirectory, - routing: routing, - ); - - @override - Future executeTask( - GoTaskServiceRequest request, { - 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 - Future closeTask({ - required GoTaskServiceRoute route, - required AssistantExecutionTarget target, - required String sessionId, - required String threadId, - }) => _acpTransport.closeTask( - target: target, - sessionId: sessionId, - threadId: threadId, - ); - - @visibleForTesting - ExternalCodeAgentAcpTransport get acpTransportForTest => _acpTransport; - - @override - Future dispose() async { - await _acpTransport.dispose(); - } -} diff --git a/test/runtime/assistant_execution_target_test.dart b/test/runtime/assistant_execution_target_test.dart index 39856655..48a7414d 100644 --- a/test/runtime/assistant_execution_target_test.dart +++ b/test/runtime/assistant_execution_target_test.dart @@ -3274,7 +3274,7 @@ void main() { await fakeGoTaskService.waitForRequestCount(prompts.length); expect(fakeGoTaskService.requests, hasLength(prompts.length)); - expect(controller.openClawGatewayActiveTasksInternal, prompts.length); + expect(controller.openClawGatewayActiveTurnsInternal.length, prompts.length); expect(controller.openClawGatewayQueuedTurnsInternal, isEmpty); for (var index = 0; index < prompts.length; index += 1) { final sessionKey = 'openclaw-e2e-$index'; @@ -3332,7 +3332,7 @@ void main() { _openClawE2ECanonicalPrompts.length, ); expect( - controller.openClawGatewayActiveTasksInternal, + controller.openClawGatewayActiveTurnsInternal.length, _openClawE2ECanonicalPrompts.length, ); @@ -3855,7 +3855,7 @@ void main() { .status, 'running', ); - expect(controller.openClawGatewayActiveTasksInternal, 1); + expect(controller.openClawGatewayActiveTurnsInternal.length, 1); }, ); @@ -4602,12 +4602,12 @@ Future _waitForOpenClawActiveTaskCount( ) async { final deadline = DateTime.now().add(const Duration(seconds: 5)); while (DateTime.now().isBefore(deadline)) { - if (controller.openClawGatewayActiveTasksInternal == expected) { + if (controller.openClawGatewayActiveTurnsInternal.length == expected) { return; } await Future.delayed(const Duration(milliseconds: 10)); } - expect(controller.openClawGatewayActiveTasksInternal, expected); + expect(controller.openClawGatewayActiveTurnsInternal.length, expected); } Future> _startOpenClawActiveTasks( @@ -4701,19 +4701,7 @@ class _RecordingGoTaskServiceClient implements GoTaskServiceClient { final List taskOutcomes = []; Future Function(GoTaskServiceRequest request)? onExecuteTask; - @override - Future loadExternalAcpCapabilities({ - required AssistantExecutionTarget target, - bool forceRefresh = false, - }) async => const ExternalCodeAgentAcpCapabilities.empty(); - @override - Future resolveExternalAcpRouting({ - required String taskPrompt, - required String workingDirectory, - required ExternalCodeAgentAcpRoutingConfig routing, - }) async => - const ExternalCodeAgentAcpRoutingResolution(raw: {}); @override Future executeTask( @@ -4786,13 +4774,7 @@ class _RecordingGoTaskServiceClient implements GoTaskServiceClient { OpenClawTaskAssociation? association, }) async {} - @override - Future closeTask({ - required GoTaskServiceRoute route, - required AssistantExecutionTarget target, - required String sessionId, - required String threadId, - }) async {} + @override Future dispose() async {} @@ -4809,19 +4791,7 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient { final Map _updates = {}; - @override - Future loadExternalAcpCapabilities({ - required AssistantExecutionTarget target, - bool forceRefresh = false, - }) async => const ExternalCodeAgentAcpCapabilities.empty(); - @override - Future resolveExternalAcpRouting({ - required String taskPrompt, - required String workingDirectory, - required ExternalCodeAgentAcpRoutingConfig routing, - }) async => - const ExternalCodeAgentAcpRoutingResolution(raw: {}); @override Future executeTask( @@ -4927,13 +4897,7 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient { cancelledSessionIds.add(sessionId); } - @override - Future closeTask({ - required GoTaskServiceRoute route, - required AssistantExecutionTarget target, - required String sessionId, - required String threadId, - }) async {} + @override Future dispose() async {}