From ded87aa63f9bf8a93f0d0f6ab25475af29404b86 Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Mon, 23 Mar 2026 14:17:09 +0800 Subject: [PATCH] refactor(desktop): route assistant execution through gateway ACP --- lib/app/app_controller_desktop.dart | 976 ++++++++++-------- lib/runtime/gateway_acp_client.dart | 845 +++++++++++++++ lib/runtime/runtime_coordinator.dart | 79 +- lib/runtime/single_agent_runner.dart | 497 ++------- .../no_direct_cli_execution_guard_suite.dart | 57 + test/runtime/runtime_coordinator_suite.dart | 14 +- 6 files changed, 1540 insertions(+), 928 deletions(-) create mode 100644 lib/runtime/gateway_acp_client.dart create mode 100644 test/runtime/no_direct_cli_execution_guard_suite.dart diff --git a/lib/app/app_controller_desktop.dart b/lib/app/app_controller_desktop.dart index 86d54a97..8bad8d42 100644 --- a/lib/app/app_controller_desktop.dart +++ b/lib/app/app_controller_desktop.dart @@ -19,13 +19,12 @@ import '../runtime/runtime_controllers.dart'; import '../runtime/runtime_models.dart'; import '../runtime/secure_config_store.dart'; import '../runtime/runtime_coordinator.dart'; +import '../runtime/gateway_acp_client.dart'; import '../runtime/codex_runtime.dart'; import '../runtime/codex_config_bridge.dart'; import '../runtime/code_agent_node_orchestrator.dart'; import '../runtime/mode_switcher.dart'; import '../runtime/agent_registry.dart'; -import '../runtime/multi_agent_broker.dart'; -import '../runtime/multi_agent_mounts.dart'; import '../runtime/multi_agent_orchestrator.dart'; import '../runtime/single_agent_runner.dart'; @@ -93,15 +92,14 @@ class AppController extends ChangeNotifier { (_isFlutterTestEnvironment ? const [] : _defaultGatewayOnlySkillScanRoots); + _gatewayAcpClient = GatewayAcpClient(endpointResolver: _resolveAcpEndpoint); _availableSingleAgentProvidersOverride = availableSingleAgentProvidersOverride; _arisBundleRepository = ArisBundleRepository(); _arisBridgeLocator = ArisBridgeLocator(); - _multiAgentMountManager = MultiAgentMountManager( - arisBundleRepository: _arisBundleRepository, - arisBridgeLocator: _arisBridgeLocator, - ); - _singleAgentRunner = singleAgentRunner ?? DefaultSingleAgentRunner(); + _singleAgentRunner = + singleAgentRunner ?? + DefaultSingleAgentRunner(acpClient: _gatewayAcpClient); _multiAgentOrchestrator = MultiAgentOrchestrator( config: _resolveMultiAgentConfig(_settingsController.snapshot), arisBundleRepository: _arisBundleRepository, @@ -132,14 +130,14 @@ class AppController extends ChangeNotifier { late final DerivedTasksController _tasksController; late final DesktopPlatformService _desktopPlatformService; late final List _gatewayOnlySkillScanRoots; + late final GatewayAcpClient _gatewayAcpClient; late final List? _availableSingleAgentProvidersOverride; late final ArisBundleRepository _arisBundleRepository; late final ArisBridgeLocator _arisBridgeLocator; - late final MultiAgentMountManager _multiAgentMountManager; late final SingleAgentRunner _singleAgentRunner; late final MultiAgentOrchestrator _multiAgentOrchestrator; - MultiAgentBrokerServer? _multiAgentBrokerServer; - MultiAgentBrokerClient? _multiAgentBrokerClient; + GatewayAcpCapabilities _acpCapabilities = + const GatewayAcpCapabilities.empty(); final Map> _assistantThreadMessages = >{}; final Map _assistantThreadRecords = @@ -155,7 +153,8 @@ class AppController extends ChangeNotifier { final Set _aiGatewayPendingSessionKeys = {}; final Set _aiGatewayAbortedSessionKeys = {}; final Set _singleAgentExternalCliPendingSessionKeys = {}; - final Set _activeMultiAgentBrokerSessions = {}; + final Map> _assistantThreadTurnQueues = + >{}; bool _multiAgentRunPending = false; int _localMessageCounter = 0; @@ -326,21 +325,13 @@ class AppController extends ChangeNotifier { bool _canUseSingleAgentProvider(SingleAgentProvider provider) { final override = _availableSingleAgentProvidersOverride; if (override != null) { - return provider != SingleAgentProvider.auto && override.contains(provider); + return provider != SingleAgentProvider.auto && + override.contains(provider); } if (provider == SingleAgentProvider.auto) { - return settings.multiAgent.mountTargets.any( - (item) => - item.available && - (item.targetId == 'codex' || - item.targetId == 'opencode' || - item.targetId == 'claude' || - item.targetId == 'gemini'), - ); + return _acpCapabilities.providers.isNotEmpty; } - return settings.multiAgent.mountTargets.any( - (item) => item.targetId == provider.providerId && item.available, - ); + return _acpCapabilities.providers.contains(provider); } SingleAgentProvider? _resolvedSingleAgentProvider( @@ -476,7 +467,9 @@ class AppController extends ChangeNotifier { SingleAgentProvider get currentSingleAgentProvider => singleAgentProviderForSession(currentSessionKey); - SingleAgentProvider? singleAgentResolvedProviderForSession(String sessionKey) { + SingleAgentProvider? singleAgentResolvedProviderForSession( + String sessionKey, + ) { final normalizedSessionKey = _normalizedAssistantSessionKey(sessionKey); return _resolvedSingleAgentProvider( singleAgentProviderForSession(normalizedSessionKey), @@ -590,18 +583,17 @@ class AppController extends ChangeNotifier { final target = assistantExecutionTargetForSession(normalizedSessionKey); if (target == AssistantExecutionTarget.singleAgent) { final provider = singleAgentProviderForSession(normalizedSessionKey); - final resolvedProvider = - singleAgentResolvedProviderForSession(normalizedSessionKey); + final resolvedProvider = singleAgentResolvedProviderForSession( + normalizedSessionKey, + ); final model = assistantModelForSession(normalizedSessionKey); - final fallbackReady = - singleAgentUsesAiChatFallbackForSession(normalizedSessionKey); + final fallbackReady = singleAgentUsesAiChatFallbackForSession( + normalizedSessionKey, + ); final host = _aiGatewayHostLabel(settings.aiGateway.baseUrl); final providerReady = resolvedProvider != null; final detail = providerReady - ? _joinConnectionParts([ - resolvedProvider.label, - model, - ]) + ? _joinConnectionParts([resolvedProvider.label, model]) : fallbackReady ? _joinConnectionParts([ appText('AI Chat fallback', 'AI Chat fallback'), @@ -614,8 +606,8 @@ class AppController extends ChangeNotifier { '${provider.label} is unavailable. Switch to Auto.', ) : singleAgentNeedsAiGatewayConfigurationForSession( - normalizedSessionKey, - ) + normalizedSessionKey, + ) ? appText( '没有可用的外部 CLI，请配置 AI Gateway fallback。', 'No external CLI is available. Configure AI Gateway fallback.', @@ -689,29 +681,7 @@ class AppController extends ChangeNotifier { } Future refreshMultiAgentMounts({bool sync = false}) async { - if (_disposed) { - return; - } - final resolved = _resolveMultiAgentConfig(settings); - final reconciled = await _multiAgentMountManager.reconcile( - config: sync ? resolved : resolved.copyWith(autoSync: false), - aiGatewayUrl: aiGatewayUrl, - configuredCodexCliPath: _resolvedCodexCliPath ?? settings.codexCliPath, - ); - if (_disposed) { - return; - } - if (jsonEncode(reconciled.toJson()) != - jsonEncode(settings.multiAgent.toJson())) { - await _settingsController.saveSnapshot( - settings.copyWith(multiAgent: reconciled), - ); - } - if (_disposed) { - return; - } - _multiAgentOrchestrator.updateConfig(reconciled); - _notifyIfActive(); + await _refreshAcpCapabilities(persistMountTargets: true); } Future runMultiAgentCollaboration({ @@ -723,125 +693,122 @@ class AppController extends ChangeNotifier { final sessionKey = currentSessionKey.trim().isEmpty ? 'main' : currentSessionKey; - final client = await _ensureMultiAgentBrokerClient(); - final aiGatewayApiKey = await loadAiGatewayApiKey(); - _multiAgentRunPending = true; - _appendLocalSessionMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'user', - text: rawPrompt, - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: null, - stopReason: null, - pending: false, - error: false, - ), - ); - _recomputeTasks(); - try { - final taskStream = settings.multiAgent.usesAris - ? (_activeMultiAgentBrokerSessions.contains(sessionKey) - ? client.sendSessionMessage( - sessionId: sessionKey, - taskPrompt: composedPrompt, - workingDirectory: - _resolveCodexWorkingDirectory() ?? - Directory.current.path, - attachments: attachments, - selectedSkills: selectedSkillLabels, - aiGatewayBaseUrl: aiGatewayUrl, - aiGatewayApiKey: aiGatewayApiKey, - ) - : client.startSession( - sessionId: sessionKey, - taskPrompt: composedPrompt, - workingDirectory: - _resolveCodexWorkingDirectory() ?? - Directory.current.path, - attachments: attachments, - selectedSkills: selectedSkillLabels, - aiGatewayBaseUrl: aiGatewayUrl, - aiGatewayApiKey: aiGatewayApiKey, - )) - : client.runTask( - taskPrompt: composedPrompt, - workingDirectory: - _resolveCodexWorkingDirectory() ?? Directory.current.path, - attachments: attachments, - selectedSkills: selectedSkillLabels, - aiGatewayBaseUrl: aiGatewayUrl, - aiGatewayApiKey: aiGatewayApiKey, + await _enqueueThreadTurn(sessionKey, () async { + final aiGatewayApiKey = await loadAiGatewayApiKey(); + _multiAgentRunPending = true; + _appendLocalSessionMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'user', + text: rawPrompt, + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: null, + stopReason: null, + pending: false, + error: false, + ), + ); + _recomputeTasks(); + try { + final taskStream = _gatewayAcpClient.runMultiAgent( + GatewayAcpMultiAgentRequest( + sessionId: sessionKey, + threadId: sessionKey, + prompt: composedPrompt, + workingDirectory: + _resolveCodexWorkingDirectory() ?? Directory.current.path, + attachments: attachments, + selectedSkills: selectedSkillLabels, + aiGatewayBaseUrl: aiGatewayUrl, + aiGatewayApiKey: aiGatewayApiKey, + resumeSession: true, + ), + ); + await for (final event in taskStream) { + if (event.type == 'result') { + final success = event.data['success'] == true; + final finalScore = event.data['finalScore']; + final iterations = event.data['iterations']; + _appendLocalSessionMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: success + ? appText( + '多 Agent 协作完成，评分 ${finalScore ?? '-'}，迭代 ${iterations ?? 0} 次。', + 'Multi-agent collaboration completed with score ${finalScore ?? '-'} after ${iterations ?? 0} iteration(s).', + ) + : appText( + '多 Agent 协作失败：${event.data['error'] ?? event.message}', + 'Multi-agent collaboration failed: ${event.data['error'] ?? event.message}', + ), + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: null, + stopReason: null, + pending: false, + error: !success, + ), ); - if (settings.multiAgent.usesAris) { - _activeMultiAgentBrokerSessions.add(sessionKey); - } - await for (final event in taskStream) { - if (event.type == 'result') { - final success = event.data['success'] == true; - final finalScore = event.data['finalScore']; - final iterations = event.data['iterations']; + continue; + } _appendLocalSessionMessage( sessionKey, GatewayChatMessage( id: _nextLocalMessageId(), role: 'assistant', - text: success - ? appText( - '多 Agent 协作完成，评分 ${finalScore ?? '-'}，迭代 ${iterations ?? 0} 次。', - 'Multi-agent collaboration completed with score ${finalScore ?? '-'} after ${iterations ?? 0} iteration(s).', - ) - : appText( - '多 Agent 协作失败：${event.data['error'] ?? event.message}', - 'Multi-agent collaboration failed: ${event.data['error'] ?? event.message}', - ), + text: event.message, timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), toolCallId: null, - toolName: null, + toolName: event.title, stopReason: null, - pending: false, - error: !success, + pending: event.pending, + error: event.error, ), ); - continue; } + } on GatewayAcpException catch (error) { _appendLocalSessionMessage( sessionKey, GatewayChatMessage( id: _nextLocalMessageId(), role: 'assistant', - text: event.message, + text: appText( + '多 Agent 协作不可用（Gateway ACP）：${error.message}', + 'Multi-agent collaboration is unavailable (Gateway ACP): ${error.message}', + ), timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), toolCallId: null, - toolName: event.title, + toolName: 'Multi-Agent', stopReason: null, - pending: event.pending, - error: event.error, + pending: false, + error: true, ), ); + } catch (error) { + _appendLocalSessionMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: error.toString(), + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: 'Multi-Agent', + stopReason: null, + pending: false, + error: true, + ), + ); + } finally { + _multiAgentRunPending = false; + _recomputeTasks(); + _notifyIfActive(); } - } catch (error) { - _appendLocalSessionMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'assistant', - text: error.toString(), - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: 'Multi-Agent', - stopReason: null, - pending: false, - error: true, - ), - ); - } finally { - _multiAgentRunPending = false; - _recomputeTasks(); - _notifyIfActive(); - } + }); } Future openOnlineWorkspace() async { @@ -887,10 +854,10 @@ class AppController extends ChangeNotifier { if (singleAgentUsesAiChatFallbackForSession(sessionKey)) { return aiGatewayConversationModelChoices; } - final selectedModel = _assistantThreadRecords[ - _normalizedAssistantSessionKey(sessionKey)] - ?.assistantModelId - .trim(); + final selectedModel = + _assistantThreadRecords[_normalizedAssistantSessionKey(sessionKey)] + ?.assistantModelId + .trim(); if (selectedModel?.isNotEmpty == true) { return [selectedModel!]; } @@ -1651,10 +1618,14 @@ class AppController extends ChangeNotifier { final sessionKey = _normalizedAssistantSessionKey( _sessionsController.currentSessionKey, ); - if (_activeMultiAgentBrokerSessions.contains(sessionKey)) { - await _multiAgentBrokerClient?.cancelSession(sessionKey); + try { + await _gatewayAcpClient.cancelSession( + sessionId: sessionKey, + threadId: sessionKey, + ); + } catch (_) { + // Best effort cancellation only. } - await _multiAgentOrchestrator.abort(); _multiAgentRunPending = false; _recomputeTasks(); _notifyIfActive(); @@ -2069,8 +2040,14 @@ class AppController extends ChangeNotifier { refreshAfterSave: false, ); if (archived) { - _activeMultiAgentBrokerSessions.remove(normalizedSessionKey); - unawaited(_multiAgentBrokerClient?.closeSession(normalizedSessionKey)); + unawaited( + _gatewayAcpClient + .closeSession( + sessionId: normalizedSessionKey, + threadId: normalizedSessionKey, + ) + .catchError((_) {}), + ); } _upsertAssistantThreadRecord( normalizedSessionKey, @@ -2270,7 +2247,7 @@ class AppController extends ChangeNotifier { _aiGatewayPendingSessionKeys.clear(); _aiGatewayAbortedSessionKeys.clear(); _singleAgentExternalCliPendingSessionKeys.clear(); - _activeMultiAgentBrokerSessions.clear(); + _assistantThreadTurnQueues.clear(); _multiAgentRunPending = false; setActiveAppLanguage(defaults.appLanguage); await _settingsController.resetSnapshot(defaults); @@ -2500,18 +2477,16 @@ class AppController extends ChangeNotifier { ); } + await _refreshAcpCapabilities(forceRefresh: true); final runtimeMode = effectiveCodeAgentRuntimeMode; - String? codexPath; - if (runtimeMode == CodeAgentRuntimeMode.externalCli) { - codexPath = await _resolveCodexCliPath(); - if (codexPath == null) { - throw StateError( - appText( - '未找到 Codex CLI。请先安装或填写可执行文件路径。', - 'Codex CLI not found. Install it or set a manual binary path.', - ), - ); - } + if (runtimeMode == CodeAgentRuntimeMode.externalCli && + !_canUseSingleAgentProvider(SingleAgentProvider.codex)) { + throw StateError( + appText( + 'Gateway ACP 未报告 Codex Provider 可用，请先检查 Agent Gateway / ACP Adapter 配置。', + 'Gateway ACP did not report a Codex provider. Check Agent Gateway / ACP Adapter settings first.', + ), + ); } await _runtimeCoordinator.configureCodexForGateway( @@ -2519,13 +2494,7 @@ class AppController extends ChangeNotifier { apiKey: apiKey, ); - await _runtimeCoordinator.startCodeAgentRuntime( - runtimeMode: runtimeMode, - codexPath: codexPath, - workingDirectory: _resolveCodexWorkingDirectory(), - ); - - _registerCodexExternalProvider(codexPath: codexPath); + _registerCodexExternalProvider(); _isCodexBridgeEnabled = true; _codexCooperationState = CodexCooperationState.bridgeOnly; await _ensureCodexGatewayRegistration(); @@ -2552,7 +2521,6 @@ class AppController extends ChangeNotifier { } else { _codeAgentBridgeRegistry.clearRegistration(); } - await _runtimeCoordinator.stopCodeAgentRuntime(); _isCodexBridgeEnabled = false; _codexCooperationState = CodexCooperationState.notStarted; _codexBridgeError = null; @@ -2589,7 +2557,7 @@ class AppController extends ChangeNotifier { _tasksController.dispose(); _store.dispose(); _desktopPlatformService.dispose(); - unawaited(_multiAgentBrokerServer?.stop() ?? Future.value()); + unawaited(_gatewayAcpClient.dispose()); super.dispose(); } @@ -2634,7 +2602,7 @@ class AppController extends ChangeNotifier { await _desktopPlatformService.initialize(settings.linuxDesktop); await _desktopPlatformService.setLaunchAtLogin(settings.launchAtLogin); _registerCodexExternalProvider(); - await _refreshCodexCliAvailability(); + await _refreshAcpCapabilities(persistMountTargets: true); if (_disposed) { return; } @@ -2675,10 +2643,6 @@ class AppController extends ChangeNotifier { // Keep the shell usable when auto-connect fails. } } - // Mount reconciliation may invoke multiple external CLIs. Keep startup - // responsive and let the mounts refresh in the background instead of - // blocking app initialization on those probes. - unawaited(refreshMultiAgentMounts(sync: settings.multiAgent.autoSync)); _settingsDraft = settings; _lastAppliedSettings = settings; _settingsDraftInitialized = true; @@ -2822,11 +2786,7 @@ class AppController extends ChangeNotifier { } if (previous.codexCliPath != current.codexCliPath || previous.codeAgentRuntimeMode != current.codeAgentRuntimeMode) { - _registerCodexExternalProvider(codexPath: current.codexCliPath); - await _refreshCodexCliAvailability(); - if (_disposed) { - return; - } + _registerCodexExternalProvider(); } if (previous.linuxDesktop.toJson().toString() != current.linuxDesktop.toJson().toString() || @@ -2840,7 +2800,7 @@ class AppController extends ChangeNotifier { if (refreshAfterSave) { _recomputeTasks(); } - unawaited(refreshMultiAgentMounts(sync: current.multiAgent.autoSync)); + unawaited(_refreshAcpCapabilities(persistMountTargets: true)); notifyListeners(); } @@ -3053,33 +3013,6 @@ class AppController extends ChangeNotifier { ); } - Future _ensureMultiAgentBrokerClient() async { - _multiAgentBrokerServer ??= MultiAgentBrokerServer(_multiAgentOrchestrator); - await _multiAgentBrokerServer!.start(); - final uri = _multiAgentBrokerServer!.wsUri; - if (uri == null) { - throw StateError('Multi-agent broker is unavailable'); - } - _runtimeCoordinator.registerExternalCodeAgent( - ExternalCodeAgentProvider( - id: 'aris-broker', - name: 'ARIS Broker', - command: 'xworkmate-multi-agent-broker', - transport: ExternalAgentTransport.websocketJsonRpc, - endpoint: uri.toString(), - capabilities: const [ - 'architect', - 'engineer', - 'tester', - 'multi-agent', - 'session-stream', - ], - ), - ); - _multiAgentBrokerClient = MultiAgentBrokerClient(uri); - return _multiAgentBrokerClient!; - } - Future _sendSingleAgentMessage( String message, { required String thinking, @@ -3094,203 +3027,14 @@ class AppController extends ChangeNotifier { if (trimmed.isEmpty && attachments.isEmpty) { return; } - - final userText = trimmed.isEmpty ? 'See attached.' : trimmed; - _appendAssistantThreadMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'user', - text: userText, - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: null, - stopReason: null, - pending: false, - error: false, - ), - ); - _aiGatewayPendingSessionKeys.add(sessionKey); - _recomputeTasks(); - _notifyIfActive(); - - try { - final selection = singleAgentProviderForSession(sessionKey); - final resolution = await _singleAgentRunner.resolveProvider( - selection: selection, - configuredCodexCliPath: configuredCodexCliPath, - ); - final provider = resolution.resolvedProvider; - if (provider == null) { - if (singleAgentUsesAiChatFallbackForSession(sessionKey)) { - _appendAssistantThreadMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'assistant', - text: _singleAgentFallbackLabel(resolution.fallbackReason), - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: 'AI Chat fallback', - stopReason: null, - pending: false, - error: false, - ), - ); - await _sendAiGatewayMessage( - message, - thinking: thinking, - attachments: attachments, - sessionKeyOverride: sessionKey, - appendUserMessage: false, - managePendingState: false, - ); - } else { - _appendAssistantThreadMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'assistant', - text: _singleAgentUnavailableLabel( - sessionKey, - resolution.fallbackReason, - ), - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: provider?.label ?? selection.label, - stopReason: null, - pending: false, - error: false, - ), - ); - } - return; - } - + await _enqueueThreadTurn(sessionKey, () async { + final userText = trimmed.isEmpty ? 'See attached.' : trimmed; _appendAssistantThreadMessage( sessionKey, GatewayChatMessage( id: _nextLocalMessageId(), - role: 'assistant', - text: appText( - '单机智能体已切换到 ${provider.label} 执行当前任务。', - 'Single Agent is using ${provider.label} for this task.', - ), - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: provider.label, - stopReason: null, - pending: false, - error: false, - ), - ); - _singleAgentExternalCliPendingSessionKeys.add(sessionKey); - - final result = await _singleAgentRunner.run( - SingleAgentRunRequest( - sessionId: sessionKey, - provider: provider, - prompt: message, - model: assistantModelForSession(sessionKey), - workingDirectory: - _resolveCodexWorkingDirectory() ?? Directory.current.path, - attachments: localAttachments, - selectedSkills: selectedSkillLabels, - aiGatewayBaseUrl: aiGatewayUrl, - aiGatewayApiKey: await loadAiGatewayApiKey(), - config: settings.multiAgent, - onOutput: (text) => _setAiGatewayStreamingText(sessionKey, text), - configuredCodexCliPath: configuredCodexCliPath, - ), - ); - _clearAiGatewayStreamingText(sessionKey); - if (result.aborted) { - final partial = result.output.trim(); - if (partial.isNotEmpty) { - _appendAssistantThreadMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'assistant', - text: partial, - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: null, - stopReason: 'aborted', - pending: false, - error: false, - ), - ); - } - return; - } - if (result.shouldFallbackToAiChat) { - if (singleAgentUsesAiChatFallbackForSession(sessionKey)) { - _appendAssistantThreadMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'assistant', - text: _singleAgentFallbackLabel( - result.fallbackReason ?? result.errorMessage, - ), - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: 'AI Chat fallback', - stopReason: null, - pending: false, - error: false, - ), - ); - await _sendAiGatewayMessage( - message, - thinking: thinking, - attachments: attachments, - sessionKeyOverride: sessionKey, - appendUserMessage: false, - managePendingState: false, - ); - } else { - _appendAssistantThreadMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'assistant', - text: _singleAgentUnavailableLabel( - sessionKey, - result.fallbackReason ?? result.errorMessage, - ), - timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), - toolCallId: null, - toolName: provider.label, - stopReason: null, - pending: false, - error: false, - ), - ); - } - return; - } - - if (!result.success) { - _appendAssistantThreadMessage( - sessionKey, - _assistantErrorMessage( - appText( - '单机智能体执行失败：${result.errorMessage}', - 'Single Agent execution failed: ${result.errorMessage}', - ), - ), - ); - return; - } - - _appendAssistantThreadMessage( - sessionKey, - GatewayChatMessage( - id: _nextLocalMessageId(), - role: 'assistant', - text: result.output, + role: 'user', + text: userText, timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), toolCallId: null, toolName: null, @@ -3299,19 +3043,209 @@ class AppController extends ChangeNotifier { error: false, ), ); - } catch (error) { - _clearAiGatewayStreamingText(sessionKey); - _appendAssistantThreadMessage( - sessionKey, - _assistantErrorMessage(error.toString()), - ); - } finally { - _singleAgentExternalCliPendingSessionKeys.remove(sessionKey); - _clearAiGatewayStreamingText(sessionKey); - _aiGatewayPendingSessionKeys.remove(sessionKey); + _aiGatewayPendingSessionKeys.add(sessionKey); _recomputeTasks(); _notifyIfActive(); - } + + try { + final selection = singleAgentProviderForSession(sessionKey); + final resolution = await _singleAgentRunner.resolveProvider( + selection: selection, + configuredCodexCliPath: configuredCodexCliPath, + ); + final provider = resolution.resolvedProvider; + if (provider == null) { + if (singleAgentUsesAiChatFallbackForSession(sessionKey)) { + _appendAssistantThreadMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: _singleAgentFallbackLabel(resolution.fallbackReason), + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: 'AI Chat fallback', + stopReason: null, + pending: false, + error: false, + ), + ); + await _sendAiGatewayMessage( + message, + thinking: thinking, + attachments: attachments, + sessionKeyOverride: sessionKey, + appendUserMessage: false, + managePendingState: false, + ); + } else { + _appendAssistantThreadMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: _singleAgentUnavailableLabel( + sessionKey, + resolution.fallbackReason, + ), + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: provider?.label ?? selection.label, + stopReason: null, + pending: false, + error: false, + ), + ); + } + return; + } + + _appendAssistantThreadMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: appText( + '单机智能体已切换到 ${provider.label} 执行当前任务。', + 'Single Agent is using ${provider.label} for this task.', + ), + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: provider.label, + stopReason: null, + pending: false, + error: false, + ), + ); + _singleAgentExternalCliPendingSessionKeys.add(sessionKey); + + final result = await _singleAgentRunner.run( + SingleAgentRunRequest( + sessionId: sessionKey, + provider: provider, + prompt: message, + model: assistantModelForSession(sessionKey), + workingDirectory: + _resolveCodexWorkingDirectory() ?? Directory.current.path, + attachments: localAttachments, + selectedSkills: selectedSkillLabels, + aiGatewayBaseUrl: aiGatewayUrl, + aiGatewayApiKey: await loadAiGatewayApiKey(), + config: settings.multiAgent, + onOutput: (text) => _appendAiGatewayStreamingText(sessionKey, text), + configuredCodexCliPath: configuredCodexCliPath, + ), + ); + _clearAiGatewayStreamingText(sessionKey); + if (result.aborted) { + final partial = result.output.trim(); + if (partial.isNotEmpty) { + _appendAssistantThreadMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: partial, + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: null, + stopReason: 'aborted', + pending: false, + error: false, + ), + ); + } + return; + } + if (result.shouldFallbackToAiChat) { + if (singleAgentUsesAiChatFallbackForSession(sessionKey)) { + _appendAssistantThreadMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: _singleAgentFallbackLabel( + result.fallbackReason ?? result.errorMessage, + ), + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: 'AI Chat fallback', + stopReason: null, + pending: false, + error: false, + ), + ); + await _sendAiGatewayMessage( + message, + thinking: thinking, + attachments: attachments, + sessionKeyOverride: sessionKey, + appendUserMessage: false, + managePendingState: false, + ); + } else { + _appendAssistantThreadMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: _singleAgentUnavailableLabel( + sessionKey, + result.fallbackReason ?? result.errorMessage, + ), + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: provider.label, + stopReason: null, + pending: false, + error: false, + ), + ); + } + return; + } + + if (!result.success) { + _appendAssistantThreadMessage( + sessionKey, + _assistantErrorMessage( + appText( + '单机智能体执行失败：${result.errorMessage}', + 'Single Agent execution failed: ${result.errorMessage}', + ), + ), + ); + return; + } + + _appendAssistantThreadMessage( + sessionKey, + GatewayChatMessage( + id: _nextLocalMessageId(), + role: 'assistant', + text: result.output, + timestampMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + toolCallId: null, + toolName: null, + stopReason: null, + pending: false, + error: false, + ), + ); + } catch (error) { + _clearAiGatewayStreamingText(sessionKey); + _appendAssistantThreadMessage( + sessionKey, + _assistantErrorMessage(error.toString()), + ); + } finally { + _singleAgentExternalCliPendingSessionKeys.remove(sessionKey); + _clearAiGatewayStreamingText(sessionKey); + _aiGatewayPendingSessionKeys.remove(sessionKey); + _recomputeTasks(); + _notifyIfActive(); + } + }); } Future _sendAiGatewayMessage( @@ -3701,7 +3635,9 @@ class AppController extends ChangeNotifier { 'This thread is pinned to ${selection.label}: $detail XWorkmate will not reroute to another external CLI automatically. Switch to Auto instead.', ); } - if (singleAgentNeedsAiGatewayConfigurationForSession(normalizedSessionKey)) { + if (singleAgentNeedsAiGatewayConfigurationForSession( + normalizedSessionKey, + )) { return detail.isEmpty ? appText( '当前没有可用的外部 CLI，也没有可用的 AI Chat fallback。请先安装外部 CLI，或配置 AI Gateway。', @@ -3858,9 +3794,8 @@ class AppController extends ChangeNotifier { return target.promptValue; } - Future> _scanGatewayOnlySkillCandidatesForSession( - String sessionKey, - ) async { + Future> + _scanGatewayOnlySkillCandidatesForSession(String sessionKey) async { final provider = singleAgentResolvedProviderForSession(sessionKey); if (provider == null) { return const []; @@ -4139,10 +4074,14 @@ class AppController extends ChangeNotifier { settings.assistantLastSessionKey == normalizedSessionKey) { return; } - await saveSettings( - settings.copyWith(assistantLastSessionKey: normalizedSessionKey), - refreshAfterSave: false, - ); + try { + await saveSettings( + settings.copyWith(assistantLastSessionKey: normalizedSessionKey), + refreshAfterSave: false, + ); + } catch (_) { + // Best effort only during teardown-sensitive transitions. + } } void _setAiGatewayStreamingText(String sessionKey, String text) { @@ -4155,6 +4094,16 @@ class AppController extends ChangeNotifier { _notifyIfActive(); } + void _appendAiGatewayStreamingText(String sessionKey, String delta) { + if (delta.isEmpty) { + return; + } + final key = _normalizedAssistantSessionKey(sessionKey); + final current = _aiGatewayStreamingTextBySession[key] ?? ''; + _aiGatewayStreamingTextBySession[key] = '$current$delta'; + _notifyIfActive(); + } + void _clearAiGatewayStreamingText(String sessionKey) { final key = _normalizedAssistantSessionKey(sessionKey); if (_aiGatewayStreamingTextBySession.remove(key) != null) { @@ -4167,6 +4116,30 @@ class AppController extends ChangeNotifier { return 'local-${DateTime.now().microsecondsSinceEpoch}-$_localMessageCounter'; } + Future _enqueueThreadTurn(String threadId, Future Function() task) { + final normalizedThreadId = _normalizedAssistantSessionKey(threadId); + final previous = + _assistantThreadTurnQueues[normalizedThreadId] ?? Future.value(); + final completer = Completer(); + late final Future next; + next = previous + .catchError((_) {}) + .then((_) async { + try { + completer.complete(await task()); + } catch (error, stackTrace) { + completer.completeError(error, stackTrace); + } + }) + .whenComplete(() { + if (identical(_assistantThreadTurnQueues[normalizedThreadId], next)) { + _assistantThreadTurnQueues.remove(normalizedThreadId); + } + }); + _assistantThreadTurnQueues[normalizedThreadId] = next; + return completer.future; + } + Uri? _normalizeAiGatewayBaseUrl(String raw) { final trimmed = raw.trim(); if (trimmed.isEmpty) { @@ -4404,19 +4377,81 @@ class AppController extends ChangeNotifier { return snapshot.copyWith(codexCliPath: normalizedPath); } - Future _refreshCodexCliAvailability() async { - _resolvedCodexCliPath = await _runtimeCoordinator.resolveCodexPath( - codexPath: settings.codexCliPath, - ); + Future _refreshAcpCapabilities({ + bool forceRefresh = false, + bool persistMountTargets = false, + }) async { + GatewayAcpCapabilities capabilities; + try { + capabilities = await _gatewayAcpClient.loadCapabilities( + forceRefresh: forceRefresh, + ); + } catch (_) { + capabilities = const GatewayAcpCapabilities.empty(); + } + _acpCapabilities = capabilities; + _resolvedCodexCliPath = + capabilities.providers.contains(SingleAgentProvider.codex) + ? appText( + '通过 Gateway ACP 能力协商检测到 Codex Provider', + 'Detected Codex provider via Gateway ACP capability negotiation', + ) + : null; + if (persistMountTargets && !_disposed) { + final currentConfig = settings.multiAgent; + final nextTargets = _mergeAcpCapabilitiesIntoMountTargets( + currentConfig.mountTargets, + capabilities, + ); + final nextConfig = currentConfig.copyWith(mountTargets: nextTargets); + if (jsonEncode(nextConfig.toJson()) != + jsonEncode(currentConfig.toJson())) { + await _settingsController.saveSnapshot( + settings.copyWith(multiAgent: nextConfig), + ); + _multiAgentOrchestrator.updateConfig(nextConfig); + } + } _notifyIfActive(); } - Future _resolveCodexCliPath() async { - if (_resolvedCodexCliPath != null) { - return _resolvedCodexCliPath; - } - await _refreshCodexCliAvailability(); - return _resolvedCodexCliPath; + List _mergeAcpCapabilitiesIntoMountTargets( + List current, + GatewayAcpCapabilities capabilities, + ) { + final source = current.isEmpty + ? ManagedMountTargetState.defaults() + : current; + final providers = capabilities.providers + .map((item) => item.providerId) + .toSet(); + return source + .map((item) { + final available = switch (item.targetId) { + 'codex' => providers.contains('codex'), + 'opencode' => providers.contains('opencode'), + 'claude' => providers.contains('claude'), + 'gemini' => providers.contains('gemini'), + 'aris' => capabilities.multiAgent, + 'openclaw' => capabilities.multiAgent || capabilities.singleAgent, + _ => false, + }; + return item.copyWith( + available: available, + discoveryState: available ? 'ready' : 'unavailable', + syncState: available ? item.syncState : 'idle', + detail: available + ? appText( + '来源：Gateway ACP capabilities', + 'Source: Gateway ACP capabilities', + ) + : appText( + 'Gateway ACP 未报告该能力。', + 'Gateway ACP did not report this capability.', + ), + ); + }) + .toList(growable: false); } String? _resolveCodexWorkingDirectory() { @@ -4428,20 +4463,27 @@ class AppController extends ChangeNotifier { return directory.existsSync() ? directory.path : null; } - void _registerCodexExternalProvider({String? codexPath}) { + void _registerCodexExternalProvider() { + final endpoint = _resolveAcpEndpoint()?.replace( + path: '/acp', + query: null, + fragment: null, + ); _runtimeCoordinator.registerExternalCodeAgent( ExternalCodeAgentProvider( id: 'codex', - name: 'Codex CLI', - command: (codexPath?.trim().isNotEmpty ?? false) - ? codexPath!.trim() - : 'codex', - defaultArgs: const ['app-server', '--listen', 'stdio://'], + name: 'Codex ACP', + command: 'xworkmate-agent-gateway', + transport: ExternalAgentTransport.websocketJsonRpc, + endpoint: endpoint?.toString() ?? '', + defaultArgs: const [], capabilities: const [ 'chat', 'code-edit', 'gateway-bridge', 'memory-sync', + 'single-agent', + 'multi-agent', ], ), ); @@ -4602,6 +4644,42 @@ class AppController extends ChangeNotifier { notifyListeners(); } + Uri? _resolveAcpEndpoint() { + final aiGatewayBase = _normalizeAiGatewayBaseUrl( + settings.aiGateway.baseUrl, + ); + if (aiGatewayBase != null) { + return aiGatewayBase; + } + final target = assistantExecutionTargetForSession( + _sessionsController.currentSessionKey, + ); + if (target == AssistantExecutionTarget.singleAgent) { + final remote = _gatewayProfileBaseUri( + settings.primaryRemoteGatewayProfile, + ); + if (remote != null) { + return remote; + } + return _gatewayProfileBaseUri(settings.primaryLocalGatewayProfile); + } + return _gatewayProfileBaseUri( + _gatewayProfileForAssistantExecutionTarget(target), + ); + } + + Uri? _gatewayProfileBaseUri(GatewayConnectionProfile profile) { + final host = profile.host.trim(); + if (host.isEmpty || profile.port <= 0) { + return null; + } + return Uri( + scheme: profile.tls ? 'https' : 'http', + host: host, + port: profile.port, + ); + } + RuntimeConnectionMode _modeFromHost(String host) { final trimmed = host.trim().toLowerCase(); if (_isLoopbackHost(trimmed)) { diff --git a/lib/runtime/gateway_acp_client.dart b/lib/runtime/gateway_acp_client.dart new file mode 100644 index 00000000..bb1232a8 --- /dev/null +++ b/lib/runtime/gateway_acp_client.dart @@ -0,0 +1,845 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'multi_agent_orchestrator.dart'; +import 'runtime_models.dart'; + +class GatewayAcpException implements Exception { + const GatewayAcpException(this.message, {this.code, this.details}); + + final String message; + final String? code; + final Object? details; + + @override + String toString() => code == null ? message : '$code: $message'; +} + +class GatewayAcpCapabilities { + const GatewayAcpCapabilities({ + required this.singleAgent, + required this.multiAgent, + required this.providers, + required this.raw, + }); + + const GatewayAcpCapabilities.empty() + : singleAgent = false, + multiAgent = false, + providers = const {}, + raw = const {}; + + final bool singleAgent; + final bool multiAgent; + final Set providers; + final Map raw; +} + +class GatewayAcpSessionUpdate { + const GatewayAcpSessionUpdate({ + required this.method, + required this.sessionId, + required this.threadId, + required this.turnId, + required this.type, + required this.textDelta, + required this.sequence, + required this.payload, + }); + + final String method; + final String sessionId; + final String threadId; + final String turnId; + final String type; + final String textDelta; + final int? sequence; + final Map payload; +} + +class GatewayAcpSingleAgentRequest { + const GatewayAcpSingleAgentRequest({ + required this.sessionId, + required this.threadId, + required this.provider, + required this.prompt, + required this.model, + required this.workingDirectory, + required this.attachments, + required this.selectedSkills, + required this.aiGatewayBaseUrl, + required this.aiGatewayApiKey, + required this.resumeSession, + }); + + final String sessionId; + final String threadId; + final SingleAgentProvider provider; + final String prompt; + final String model; + final String workingDirectory; + final List attachments; + final List selectedSkills; + final String aiGatewayBaseUrl; + final String aiGatewayApiKey; + final bool resumeSession; +} + +class GatewayAcpSingleAgentResult { + const GatewayAcpSingleAgentResult({ + required this.success, + required this.output, + required this.errorMessage, + required this.turnId, + required this.raw, + }); + + final bool success; + final String output; + final String errorMessage; + final String turnId; + final Map raw; +} + +class GatewayAcpMultiAgentRequest { + const GatewayAcpMultiAgentRequest({ + required this.sessionId, + required this.threadId, + required this.prompt, + required this.workingDirectory, + required this.attachments, + required this.selectedSkills, + required this.aiGatewayBaseUrl, + required this.aiGatewayApiKey, + required this.resumeSession, + }); + + final String sessionId; + final String threadId; + final String prompt; + final String workingDirectory; + final List attachments; + final List selectedSkills; + final String aiGatewayBaseUrl; + final String aiGatewayApiKey; + final bool resumeSession; +} + +class GatewayAcpClient { + GatewayAcpClient({required this.endpointResolver}); + + final Uri? Function() endpointResolver; + + int _requestCounter = 0; + GatewayAcpCapabilities _cachedCapabilities = + const GatewayAcpCapabilities.empty(); + DateTime? _capabilitiesRefreshedAt; + + Future loadCapabilities({ + bool forceRefresh = false, + }) async { + if (!forceRefresh && + _capabilitiesRefreshedAt != null && + DateTime.now().difference(_capabilitiesRefreshedAt!) < + const Duration(seconds: 15)) { + return _cachedCapabilities; + } + + final response = await _requestWithFallback( + _GatewayAcpRpcRequest( + id: _nextRequestId('capabilities'), + method: 'acp.capabilities', + params: const {}, + ), + onNotification: (_) {}, + ); + final result = asMap(response['result']); + final caps = asMap(result['capabilities']); + final providers = {}; + for (final raw in [ + ...asList(result['providers']), + ...asList(caps['providers']), + ]) { + if (raw == null) { + continue; + } + final provider = SingleAgentProviderCopy.fromJsonValue( + raw.toString().trim().toLowerCase(), + ); + if (provider != SingleAgentProvider.auto) { + providers.add(provider); + } + } + final singleAgent = + boolValue(result['singleAgent']) ?? + boolValue(caps['single_agent']) ?? + providers.isNotEmpty; + final multiAgent = + boolValue(result['multiAgent']) ?? + boolValue(caps['multi_agent']) ?? + true; + _cachedCapabilities = GatewayAcpCapabilities( + singleAgent: singleAgent, + multiAgent: multiAgent, + providers: providers, + raw: result, + ); + _capabilitiesRefreshedAt = DateTime.now(); + return _cachedCapabilities; + } + + Future runSingleAgent( + GatewayAcpSingleAgentRequest request, { + void Function(GatewayAcpSessionUpdate update)? onUpdate, + }) async { + final capabilities = await loadCapabilities(); + if (!capabilities.singleAgent || + !capabilities.providers.contains(request.provider)) { + throw GatewayAcpException( + 'Single-agent provider ${request.provider.providerId} is unavailable from ACP capabilities', + code: 'ACP_SINGLE_AGENT_UNAVAILABLE', + ); + } + final outputBuffer = StringBuffer(); + var lastSequence = -1; + final rpcRequest = _GatewayAcpRpcRequest( + id: _nextRequestId('single-agent'), + method: request.resumeSession ? 'session.message' : 'session.start', + params: { + 'sessionId': request.sessionId, + 'threadId': request.threadId, + 'mode': 'single-agent', + 'provider': request.provider.providerId, + 'taskPrompt': request.prompt, + 'model': request.model, + 'workingDirectory': request.workingDirectory, + 'attachments': request.attachments + .map( + (item) => { + 'name': item.name, + 'description': item.description, + 'path': item.path, + }, + ) + .toList(growable: false), + 'selectedSkills': request.selectedSkills, + 'aiGatewayBaseUrl': request.aiGatewayBaseUrl, + 'aiGatewayApiKey': request.aiGatewayApiKey, + }, + ); + final response = await _requestWithFallback( + rpcRequest, + onNotification: (notification) { + final update = _sessionUpdateFromNotification(notification); + if (update == null) { + return; + } + if (update.sessionId != request.sessionId) { + return; + } + if (update.sequence != null && update.sequence! <= lastSequence) { + return; + } + if (update.sequence != null) { + lastSequence = update.sequence!; + } + if (update.textDelta.isNotEmpty) { + outputBuffer.write(update.textDelta); + } + onUpdate?.call(update); + }, + ); + final result = asMap(response['result']); + final explicitOutput = _extractOutput(result); + final output = explicitOutput.isNotEmpty + ? explicitOutput + : outputBuffer.toString().trim(); + final success = boolValue(result['success']) ?? output.isNotEmpty; + return GatewayAcpSingleAgentResult( + success: success, + output: output, + errorMessage: stringValue(result['error']) ?? '', + turnId: stringValue(result['turnId']) ?? '', + raw: result, + ); + } + + Stream runMultiAgent( + GatewayAcpMultiAgentRequest request, + ) { + final controller = StreamController(); + unawaited(() async { + final capabilities = await loadCapabilities(); + if (!capabilities.multiAgent) { + throw const GatewayAcpException( + 'Multi-agent capability is unavailable from ACP', + code: 'ACP_MULTI_AGENT_UNAVAILABLE', + ); + } + final rpcRequest = _GatewayAcpRpcRequest( + id: _nextRequestId('multi-agent'), + method: request.resumeSession ? 'session.message' : 'session.start', + params: { + 'sessionId': request.sessionId, + 'threadId': request.threadId, + 'mode': 'multi-agent', + 'taskPrompt': request.prompt, + 'workingDirectory': request.workingDirectory, + 'attachments': request.attachments + .map( + (item) => { + 'name': item.name, + 'description': item.description, + 'path': item.path, + }, + ) + .toList(growable: false), + 'selectedSkills': request.selectedSkills, + 'aiGatewayBaseUrl': request.aiGatewayBaseUrl, + 'aiGatewayApiKey': request.aiGatewayApiKey, + }, + ); + var lastSequence = -1; + try { + final response = await _requestWithFallback( + rpcRequest, + onNotification: (notification) { + final event = _multiAgentEventFromNotification(notification); + if (event == null) { + return; + } + final seq = + (event.data['seq'] as num?)?.toInt() ?? + (event.data['sequence'] as num?)?.toInt(); + if (seq != null && seq <= lastSequence) { + return; + } + if (seq != null) { + lastSequence = seq; + } + if (!controller.isClosed) { + controller.add(event); + } + }, + ); + final result = asMap(response['result']); + if (!controller.isClosed) { + controller.add( + MultiAgentRunEvent( + type: 'result', + title: '', + message: stringValue(result['summary']) ?? '', + pending: false, + error: !(boolValue(result['success']) ?? false), + data: result, + ), + ); + } + } catch (error) { + if (!controller.isClosed) { + controller.add( + MultiAgentRunEvent( + type: 'result', + title: '', + message: error.toString(), + pending: false, + error: true, + data: {'error': error.toString()}, + ), + ); + } + } finally { + await controller.close(); + } + }()); + return controller.stream; + } + + Future cancelSession({ + required String sessionId, + required String threadId, + }) async { + await _requestWithFallback( + _GatewayAcpRpcRequest( + id: _nextRequestId('cancel'), + method: 'session.cancel', + params: {'sessionId': sessionId, 'threadId': threadId}, + ), + onNotification: (_) {}, + ); + } + + Future closeSession({ + required String sessionId, + required String threadId, + }) async { + await _requestWithFallback( + _GatewayAcpRpcRequest( + id: _nextRequestId('close'), + method: 'session.close', + params: {'sessionId': sessionId, 'threadId': threadId}, + ), + onNotification: (_) {}, + ); + } + + Future dispose() async {} + + Future> _requestWithFallback( + _GatewayAcpRpcRequest request, { + required void Function(Map) onNotification, + }) async { + try { + return await _requestViaWebSocket( + request, + onNotification: onNotification, + ); + } catch (_) { + return _requestViaHttp(request, onNotification: onNotification); + } + } + + Future> _requestViaWebSocket( + _GatewayAcpRpcRequest request, { + required void Function(Map) onNotification, + }) async { + final endpoint = _resolveWebSocketRpcEndpoint(); + if (endpoint == null) { + throw const GatewayAcpException( + 'Missing ACP endpoint', + code: 'ACP_ENDPOINT_MISSING', + ); + } + + final socket = await WebSocket.connect(endpoint.toString()).timeout( + const Duration(seconds: 6), + onTimeout: () => throw const GatewayAcpException( + 'ACP websocket connect timeout', + code: 'ACP_WS_CONNECT_TIMEOUT', + ), + ); + final completer = Completer>(); + late final StreamSubscription subscription; + subscription = socket.listen( + (raw) { + final json = _decodeMap(raw); + final id = stringValue(json['id']); + final method = stringValue(json['method']) ?? ''; + if (id == request.id && + (json.containsKey('result') || json.containsKey('error'))) { + if (!completer.isCompleted) { + completer.complete(json); + } + return; + } + if (method.isNotEmpty) { + onNotification(json); + } + }, + onError: (Object error, StackTrace stackTrace) { + if (!completer.isCompleted) { + completer.completeError( + GatewayAcpException(error.toString(), code: 'ACP_WS_RUNTIME_ERROR'), + ); + } + }, + onDone: () { + if (!completer.isCompleted) { + completer.completeError( + const GatewayAcpException( + 'ACP websocket closed before response', + code: 'ACP_WS_EARLY_CLOSE', + ), + ); + } + }, + cancelOnError: true, + ); + + socket.add( + jsonEncode({ + 'jsonrpc': '2.0', + 'id': request.id, + 'method': request.method, + 'params': request.params, + }), + ); + try { + final response = await completer.future.timeout( + const Duration(seconds: 120), + ); + _throwIfJsonRpcError(response); + return response; + } finally { + await subscription.cancel(); + await socket.close(); + } + } + + Future> _requestViaHttp( + _GatewayAcpRpcRequest request, { + required void Function(Map) onNotification, + }) async { + final endpoint = _resolveHttpRpcEndpoint(); + if (endpoint == null) { + throw const GatewayAcpException( + 'Missing ACP HTTP endpoint', + code: 'ACP_HTTP_ENDPOINT_MISSING', + ); + } + + final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); + try { + final httpRequest = await client.postUrl(endpoint); + httpRequest.headers.set( + HttpHeaders.contentTypeHeader, + 'application/json; charset=utf-8', + ); + httpRequest.headers.set( + HttpHeaders.acceptHeader, + 'text/event-stream, application/json', + ); + httpRequest.add( + utf8.encode( + jsonEncode({ + 'jsonrpc': '2.0', + 'id': request.id, + 'method': request.method, + 'params': request.params, + }), + ), + ); + final response = await httpRequest.close().timeout( + const Duration(seconds: 120), + ); + final contentType = + response.headers.contentType?.mimeType.toLowerCase() ?? + response.headers + .value(HttpHeaders.contentTypeHeader) + ?.toLowerCase() ?? + ''; + if (contentType.contains('text/event-stream')) { + return _consumeSseRpcResponse( + response: response, + requestId: request.id, + onNotification: onNotification, + ); + } + final body = await response.transform(utf8.decoder).join(); + final decoded = _decodeMap(body); + _throwIfJsonRpcError(decoded); + return decoded; + } finally { + client.close(force: true); + } + } + + Future> _consumeSseRpcResponse({ + required HttpClientResponse response, + required String requestId, + required void Function(Map) onNotification, + }) async { + final completer = Completer>(); + final eventLines = []; + + void consumeEventPayload(String payload) { + final trimmed = payload.trim(); + if (trimmed.isEmpty || trimmed == '[DONE]') { + return; + } + final json = _decodeMap(trimmed); + if (stringValue(json['id']) == requestId && + (json.containsKey('result') || json.containsKey('error'))) { + if (!completer.isCompleted) { + completer.complete(json); + } + return; + } + if ((stringValue(json['method']) ?? '').isNotEmpty) { + onNotification(json); + } + } + + await for (final line + in response.transform(utf8.decoder).transform(const LineSplitter())) { + if (line.isEmpty) { + if (eventLines.isNotEmpty) { + consumeEventPayload(eventLines.join('\n')); + eventLines.clear(); + } + continue; + } + if (line.startsWith('data:')) { + eventLines.add(line.substring(5).trimLeft()); + } + } + + if (eventLines.isNotEmpty) { + consumeEventPayload(eventLines.join('\n')); + } + if (!completer.isCompleted) { + throw const GatewayAcpException( + 'ACP SSE ended without JSON-RPC response', + code: 'ACP_SSE_NO_RESULT', + ); + } + final resolved = await completer.future; + _throwIfJsonRpcError(resolved); + return resolved; + } + + GatewayAcpSessionUpdate? _sessionUpdateFromNotification( + Map notification, + ) { + final method = stringValue(notification['method']) ?? ''; + if (method != 'session.update' && method != 'acp.session.update') { + return null; + } + final params = asMap(notification['params']); + return GatewayAcpSessionUpdate( + method: method, + sessionId: stringValue(params['sessionId']) ?? '', + threadId: stringValue(params['threadId']) ?? '', + turnId: stringValue(params['turnId']) ?? '', + type: + stringValue(params['type']) ?? + stringValue(params['event']) ?? + 'status', + textDelta: + stringValue(params['delta']) ?? + stringValue(params['text']) ?? + stringValue(asMap(params['message'])['content']) ?? + '', + sequence: intValue(params['seq']) ?? intValue(notification['seq']), + payload: params, + ); + } + + MultiAgentRunEvent? _multiAgentEventFromNotification( + Map notification, + ) { + final method = stringValue(notification['method']) ?? ''; + if (method == 'multi_agent.event' || method == 'acp.multi_agent.event') { + return MultiAgentRunEvent.fromJson(asMap(notification['params'])); + } + final update = _sessionUpdateFromNotification(notification); + if (update == null || update.payload['mode'] != 'multi-agent') { + return null; + } + return MultiAgentRunEvent( + type: update.type, + title: stringValue(update.payload['title']) ?? '', + message: update.textDelta.isNotEmpty + ? update.textDelta + : stringValue(update.payload['message']) ?? '', + pending: boolValue(update.payload['pending']) ?? false, + error: boolValue(update.payload['error']) ?? false, + role: stringValue(update.payload['role']), + iteration: intValue(update.payload['iteration']), + score: intValue(update.payload['score']), + data: update.payload, + ); + } + + String _extractOutput(Map result) { + final direct = stringValue(result['output']); + if ((direct ?? '').trim().isNotEmpty) { + return direct!.trim(); + } + final text = stringValue(result['text']); + if ((text ?? '').trim().isNotEmpty) { + return text!.trim(); + } + final summary = stringValue(result['summary']); + if ((summary ?? '').trim().isNotEmpty) { + return summary!.trim(); + } + final message = asMap(result['message']); + final messageContent = stringValue(message['content']); + if ((messageContent ?? '').trim().isNotEmpty) { + return messageContent!.trim(); + } + return ''; + } + + Map asMap(Object? raw) { + if (raw is Map) { + return raw; + } + if (raw is Map) { + return raw.cast(); + } + return const {}; + } + + List asList(Object? raw) { + if (raw is List) { + return raw; + } + if (raw is List) { + return raw.cast(); + } + return const []; + } + + String? stringValue(Object? raw) { + if (raw == null) { + return null; + } + final value = raw.toString().trim(); + return value.isEmpty ? null : value; + } + + 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; + } + + int? intValue(Object? raw) { + if (raw is int) { + return raw; + } + if (raw is num) { + return raw.toInt(); + } + return int.tryParse(raw?.toString().trim() ?? ''); + } + + void _throwIfJsonRpcError(Map envelope) { + final error = asMap(envelope['error']); + if (error.isEmpty) { + return; + } + throw GatewayAcpException( + stringValue(error['message']) ?? 'ACP JSON-RPC request failed', + code: stringValue(error['code']), + details: error['data'], + ); + } + + Map _decodeMap(dynamic raw) { + if (raw is Map) { + return raw; + } + if (raw is Map) { + return raw.cast(); + } + final text = raw is String ? raw : utf8.decode(raw as List); + final decoded = jsonDecode(_extractFirstJsonDocument(text)); + if (decoded is Map) { + return decoded; + } + if (decoded is Map) { + return decoded.cast(); + } + return const {}; + } + + Uri? _resolveWebSocketRpcEndpoint() { + final base = endpointResolver(); + if (base == null) { + return null; + } + final secure = base.scheme.toLowerCase() == 'https'; + return base.replace( + scheme: secure ? 'wss' : 'ws', + path: '/acp', + query: null, + fragment: null, + ); + } + + Uri? _resolveHttpRpcEndpoint() { + final base = endpointResolver(); + if (base == null) { + return null; + } + final scheme = base.scheme.toLowerCase(); + if (scheme != 'http' && scheme != 'https') { + return null; + } + return base.replace(path: '/acp/rpc', query: null, fragment: null); + } + + String _nextRequestId(String method) { + return '${DateTime.now().microsecondsSinceEpoch}-$method-${_requestCounter++}'; + } + + String _extractFirstJsonDocument(String text) { + final trimmed = text.trim(); + if (trimmed.isEmpty) { + throw const FormatException('Empty response body'); + } + final objectStart = trimmed.indexOf('{'); + final arrayStart = trimmed.indexOf('['); + var start = -1; + if (objectStart >= 0 && arrayStart >= 0) { + start = objectStart < arrayStart ? objectStart : arrayStart; + } else if (objectStart >= 0) { + start = objectStart; + } else if (arrayStart >= 0) { + start = arrayStart; + } + if (start < 0) { + throw const FormatException('Missing JSON document'); + } + + var depth = 0; + var inString = false; + var escaped = false; + for (var index = start; index < trimmed.length; index++) { + final char = trimmed[index]; + if (inString) { + if (escaped) { + escaped = false; + } else if (char == r'\') { + escaped = true; + } else if (char == '"') { + inString = false; + } + continue; + } + if (char == '"') { + inString = true; + continue; + } + if (char == '{' || char == '[') { + depth += 1; + } else if (char == '}' || char == ']') { + depth -= 1; + if (depth == 0) { + return trimmed.substring(start, index + 1); + } + } + } + throw const FormatException('Unterminated JSON document'); + } +} + +class _GatewayAcpRpcRequest { + const _GatewayAcpRpcRequest({ + required this.id, + required this.method, + required this.params, + }); + + final String id; + final String method; + final Map params; +} diff --git a/lib/runtime/runtime_coordinator.dart b/lib/runtime/runtime_coordinator.dart index b7cd54d2..bc39be6e 100644 --- a/lib/runtime/runtime_coordinator.dart +++ b/lib/runtime/runtime_coordinator.dart @@ -211,11 +211,6 @@ class RuntimeCoordinator extends ChangeNotifier { throw StateError('Failed to connect: ${result.error}'); } - // Step 2: Start code-agent runtime according to selected mode. - if (preferredMode != GatewayMode.offline) { - await _ensureCodeAgentRuntime(); - } - _state = CoordinatorState.ready; notifyListeners(); } catch (e) { @@ -248,10 +243,6 @@ class RuntimeCoordinator extends ChangeNotifier { throw StateError('No available connection mode: ${result.error}'); } - if (result.mode != GatewayMode.offline) { - await _ensureCodeAgentRuntime(); - } - _state = CoordinatorState.ready; notifyListeners(); } catch (e) { @@ -283,8 +274,7 @@ class RuntimeCoordinator extends ChangeNotifier { } return null; } - - return codex.findCodexBinary(); + return null; } /// Start the code-agent runtime without changing the Gateway connection state. @@ -297,49 +287,14 @@ class RuntimeCoordinator extends ChangeNotifier { _codexPath = codexPath?.trim(); _cwd = workingDirectory ?? _cwd ?? Directory.current.path; _lastError = null; - - if (runtimeMode == CodeAgentRuntimeMode.builtIn) { - if (codex.isConnected) { - await codex.stop(); - } - _state = CoordinatorState.ready; - notifyListeners(); - return; - } - - final resolvedCodexPath = await resolveCodexPath(codexPath: _codexPath); - if (resolvedCodexPath == null) { - _state = CoordinatorState.error; - _lastError = 'Codex CLI not found'; - notifyListeners(); - throw StateError('Codex CLI not found'); - } - - _codexPath = resolvedCodexPath; - if (codex.isConnected) { - _state = CoordinatorState.ready; - notifyListeners(); - return; - } - - _state = CoordinatorState.connecting; + _state = CoordinatorState.ready; notifyListeners(); - - try { - await codex.startStdio(codexPath: resolvedCodexPath, cwd: _cwd); - _state = CoordinatorState.ready; - notifyListeners(); - } catch (error) { - _state = CoordinatorState.error; - _lastError = error.toString(); - notifyListeners(); - rethrow; - } } Future stopCodeAgentRuntime() async { - await codex.stop(); - _state = CoordinatorState.disconnected; + _state = gateway.isConnected + ? CoordinatorState.ready + : CoordinatorState.disconnected; notifyListeners(); } @@ -404,7 +359,7 @@ class RuntimeCoordinator extends ChangeNotifier { _state = CoordinatorState.disconnected; notifyListeners(); - await Future.wait([codex.stop(), gateway.disconnect()]); + await gateway.disconnect(); } Future _switchMode(GatewayMode mode) { @@ -418,28 +373,6 @@ class RuntimeCoordinator extends ChangeNotifier { } } - Future _ensureCodeAgentRuntime() async { - if (_runtimeMode == CodeAgentRuntimeMode.builtIn) { - // Built-in mode: runtime is assumed internal, no external process needed. - return; - } - - final resolvedCodexPath = await resolveCodexPath(codexPath: _codexPath); - if (resolvedCodexPath == null) { - // Fall back to offline mode if external Codex CLI is unavailable. - await modeSwitcher.switchToOffline(); - return; - } - - _codexPath = resolvedCodexPath; - try { - await codex.startStdio(codexPath: resolvedCodexPath, cwd: _cwd); - } catch (_) { - // Continue without external code agent in offline mode. - await modeSwitcher.switchToOffline(); - } - } - static Set _normalizeCapabilitySet(Iterable capabilities) { return capabilities .map((item) => item.trim().toLowerCase()) diff --git a/lib/runtime/single_agent_runner.dart b/lib/runtime/single_agent_runner.dart index 66acdfab..88df18e5 100644 --- a/lib/runtime/single_agent_runner.dart +++ b/lib/runtime/single_agent_runner.dart @@ -1,6 +1,4 @@ -import 'dart:convert'; -import 'dart:io'; - +import 'gateway_acp_client.dart'; import 'multi_agent_orchestrator.dart'; import 'runtime_models.dart'; @@ -78,20 +76,8 @@ abstract class SingleAgentRunner { } class DefaultSingleAgentRunner implements SingleAgentRunner { - DefaultSingleAgentRunner({ - Future Function(String command)? binaryExistsResolver, - CliProcessStarter? processStarter, - }) : _binaryExistsResolver = binaryExistsResolver, - _processStarter = - processStarter ?? - ((executable, arguments, {environment, workingDirectory}) { - return Process.start( - executable, - arguments, - environment: environment, - workingDirectory: workingDirectory, - ); - }); + DefaultSingleAgentRunner({required GatewayAcpClient acpClient}) + : _acpClient = acpClient; static const List _autoOrder = [ SingleAgentProvider.codex, @@ -100,180 +86,111 @@ class DefaultSingleAgentRunner implements SingleAgentRunner { SingleAgentProvider.gemini, ]; - final Future Function(String command)? _binaryExistsResolver; - final CliProcessStarter _processStarter; - final Map _activeProcesses = {}; - final Set _abortedSessionIds = {}; + final GatewayAcpClient _acpClient; @override Future resolveProvider({ required SingleAgentProvider selection, required String configuredCodexCliPath, }) async { - if (selection != SingleAgentProvider.auto) { - final available = await _isProviderAvailable( - selection, - configuredCodexCliPath: configuredCodexCliPath, - ); - return SingleAgentProviderResolution( - selection: selection, - resolvedProvider: available ? selection : null, - fallbackReason: available - ? null - : '${selection.label} CLI is unavailable on this device.', - ); - } - - for (final provider in _autoOrder) { - if (await _isProviderAvailable( - provider, - configuredCodexCliPath: configuredCodexCliPath, - )) { + try { + final capabilities = await _acpClient.loadCapabilities(); + if (!capabilities.singleAgent) { return SingleAgentProviderResolution( selection: selection, - resolvedProvider: provider, - fallbackReason: null, + resolvedProvider: null, + fallbackReason: 'ACP single-agent capability is unavailable.', + ); + } + if (selection != SingleAgentProvider.auto) { + final available = capabilities.providers.contains(selection); + return SingleAgentProviderResolution( + selection: selection, + resolvedProvider: available ? selection : null, + fallbackReason: available + ? null + : '${selection.label} provider is unavailable from ACP adapter.', ); } - } - return const SingleAgentProviderResolution( - selection: SingleAgentProvider.auto, - resolvedProvider: null, - fallbackReason: 'No supported external CLI provider is available.', - ); + for (final provider in _autoOrder) { + if (capabilities.providers.contains(provider)) { + return SingleAgentProviderResolution( + selection: selection, + resolvedProvider: provider, + fallbackReason: null, + ); + } + } + return const SingleAgentProviderResolution( + selection: SingleAgentProvider.auto, + resolvedProvider: null, + fallbackReason: 'No ACP single-agent provider is currently available.', + ); + } catch (error) { + return SingleAgentProviderResolution( + selection: selection, + resolvedProvider: null, + fallbackReason: 'ACP capability negotiation failed: $error', + ); + } } @override Future run(SingleAgentRunRequest request) async { - final command = _resolveCommand( - request.provider, - configuredCodexCliPath: request.configuredCodexCliPath, - model: request.model, - ); - final args = _buildArgs( - provider: request.provider, - command: command, - model: request.model, - prompt: _augmentPrompt(request), - cwd: request.workingDirectory, - ); - final env = _buildEnvVars( - provider: request.provider, - aiGatewayBaseUrl: request.aiGatewayBaseUrl, - aiGatewayApiKey: request.aiGatewayApiKey, - config: request.config, - ); - try { - final process = await _processStarter( - command, - args, - environment: env, - workingDirectory: request.workingDirectory.trim().isEmpty - ? null - : request.workingDirectory, + final result = await _acpClient.runSingleAgent( + GatewayAcpSingleAgentRequest( + sessionId: request.sessionId, + threadId: request.sessionId, + provider: request.provider, + prompt: _augmentPrompt(request), + model: request.model, + workingDirectory: request.workingDirectory, + attachments: request.attachments, + selectedSkills: request.selectedSkills, + aiGatewayBaseUrl: request.aiGatewayBaseUrl, + aiGatewayApiKey: request.aiGatewayApiKey, + resumeSession: true, + ), + onUpdate: (update) { + if (update.textDelta.isNotEmpty) { + request.onOutput?.call(update.textDelta); + } + }, ); - _activeProcesses[request.sessionId] = process; - await process.stdin.close(); - final timeout = Duration(seconds: request.config.timeoutSeconds); - final stdoutBuffer = StringBuffer(); - final stderrBuffer = StringBuffer(); - final stdoutFuture = process.stdout - .transform(utf8.decoder) - .listen((chunk) { - if (chunk.isEmpty) { - return; - } - stdoutBuffer.write(chunk); - request.onOutput?.call(stdoutBuffer.toString()); - }) - .asFuture(); - final stderrFuture = process.stderr - .transform(utf8.decoder) - .listen((chunk) { - if (chunk.isEmpty) { - return; - } - stderrBuffer.write(chunk); - }) - .asFuture(); - final exitCode = await process.exitCode.timeout(timeout, onTimeout: () { - try { - process.kill(ProcessSignal.sigkill); - } catch (_) { - // Best effort only. - } - return -1; - }); - await Future.wait(>[ - stdoutFuture.timeout(timeout, onTimeout: () {}), - stderrFuture.timeout(timeout, onTimeout: () {}), - ]); - - final output = stdoutBuffer.toString().trim().isNotEmpty - ? stdoutBuffer.toString().trim() - : stderrBuffer.toString().trim(); - if (_abortedSessionIds.remove(request.sessionId)) { - return SingleAgentRunResult( - provider: request.provider, - output: output, - success: false, - errorMessage: 'aborted', - shouldFallbackToAiChat: false, - aborted: true, - ); - } - if (exitCode == 0 && output.isNotEmpty) { - return SingleAgentRunResult( - provider: request.provider, - output: output, - success: true, - errorMessage: '', - shouldFallbackToAiChat: false, - ); - } - - final fallbackReason = _isLaunchFailureExit( - exitCode, - stderrBuffer.toString(), - ) - ? '${request.provider.label} CLI could not be launched.' - : null; return SingleAgentRunResult( provider: request.provider, - output: output, - success: false, - errorMessage: stderrBuffer.toString().trim().isNotEmpty - ? stderrBuffer.toString().trim() - : 'CLI exited with code $exitCode', - shouldFallbackToAiChat: fallbackReason != null, - fallbackReason: fallbackReason, + output: result.output, + success: result.success, + errorMessage: result.errorMessage, + shouldFallbackToAiChat: !result.success && result.output.isEmpty, + fallbackReason: !result.success + ? 'ACP single-agent run failed: ${result.errorMessage}' + : null, ); - } catch (error) { - if (_abortedSessionIds.remove(request.sessionId)) { - return SingleAgentRunResult( - provider: request.provider, - output: '', - success: false, - errorMessage: 'aborted', - shouldFallbackToAiChat: false, - aborted: true, - ); - } - final fallbackReason = _isLaunchFailureError(error) - ? '${request.provider.label} CLI could not be launched.' - : null; + } on GatewayAcpException catch (error) { + final shouldFallback = _shouldFallbackToAiChat(error.code, error.message); return SingleAgentRunResult( provider: request.provider, output: '', success: false, errorMessage: error.toString(), - shouldFallbackToAiChat: fallbackReason != null, - fallbackReason: fallbackReason, + shouldFallbackToAiChat: shouldFallback, + fallbackReason: shouldFallback + ? '${request.provider.label} provider is unavailable from ACP adapter.' + : null, + ); + } catch (error) { + return SingleAgentRunResult( + provider: request.provider, + output: '', + success: false, + errorMessage: error.toString(), + shouldFallbackToAiChat: true, + fallbackReason: + '${request.provider.label} provider run failed before completion.', ); - } finally { - _activeProcesses.remove(request.sessionId); } } @@ -283,227 +200,29 @@ class DefaultSingleAgentRunner implements SingleAgentRunner { if (normalized.isEmpty) { return; } - _abortedSessionIds.add(normalized); - final process = _activeProcesses[normalized]; - if (process == null) { - return; - } try { - process.kill(ProcessSignal.sigterm); + await _acpClient.cancelSession( + sessionId: normalized, + threadId: normalized, + ); } catch (_) { // Best effort only. } } - Future _isProviderAvailable( - SingleAgentProvider provider, { - required String configuredCodexCliPath, - }) async { - if (provider == SingleAgentProvider.auto) { - return false; + bool _shouldFallbackToAiChat(String? code, String message) { + final normalizedCode = code?.trim().toUpperCase() ?? ''; + if (normalizedCode == 'ACP_ENDPOINT_MISSING' || + normalizedCode == 'ACP_HTTP_ENDPOINT_MISSING' || + normalizedCode == 'ACP_WS_CONNECT_TIMEOUT' || + normalizedCode == 'ACP_WS_RUNTIME_ERROR' || + normalizedCode == 'ACP_WS_EARLY_CLOSE') { + return true; } - if (provider == SingleAgentProvider.codex && - configuredCodexCliPath.trim().isNotEmpty) { - return File(configuredCodexCliPath.trim()).existsSync(); - } - return _binaryExists(_binaryName(provider)); - } - - Future _binaryExists(String command) async { - if (_binaryExistsResolver != null) { - return _binaryExistsResolver(command); - } - final check = await Process.run( - Platform.isWindows ? 'where' : 'which', - [command], - runInShell: true, - ); - return check.exitCode == 0 && '${check.stdout}'.trim().isNotEmpty; - } - - String _binaryName(SingleAgentProvider provider) { - return switch (provider) { - SingleAgentProvider.auto => 'auto', - SingleAgentProvider.codex => 'codex', - SingleAgentProvider.opencode => 'opencode', - SingleAgentProvider.claude => 'claude', - SingleAgentProvider.gemini => 'gemini', - }; - } - - String _resolveCommand( - SingleAgentProvider provider, { - required String configuredCodexCliPath, - required String model, - }) { - final useOllamaLaunch = _prefersOllamaLaunch( - provider: provider, - model: model, - ); - if (useOllamaLaunch) { - return 'ollama'; - } - if (provider == SingleAgentProvider.codex && - configuredCodexCliPath.trim().isNotEmpty) { - return configuredCodexCliPath.trim(); - } - return _binaryName(provider); - } - - List _buildArgs({ - required SingleAgentProvider provider, - required String command, - required String model, - required String prompt, - required String cwd, - }) { - final useOllamaLaunch = command == 'ollama'; - switch (provider) { - case SingleAgentProvider.claude: - if (useOllamaLaunch) { - return _buildOllamaLaunchArgs( - provider: provider, - model: model, - prompt: prompt, - cwd: cwd, - ); - } - return model.trim().isEmpty - ? ['-p', prompt] - : ['--model', model.trim(), '-p', prompt]; - case SingleAgentProvider.codex: - if (useOllamaLaunch) { - return _buildOllamaLaunchArgs( - provider: provider, - model: model, - prompt: prompt, - cwd: cwd, - ); - } - return [ - 'exec', - '--skip-git-repo-check', - '--color', - 'never', - if (cwd.trim().isNotEmpty) ...['-C', cwd.trim()], - if (model.trim().isNotEmpty) ...['-m', model.trim()], - prompt, - ]; - case SingleAgentProvider.gemini: - return model.trim().isEmpty - ? ['-p', prompt] - : ['--model', model.trim(), '-p', prompt]; - case SingleAgentProvider.opencode: - if (useOllamaLaunch) { - return _buildOllamaLaunchArgs( - provider: provider, - model: model, - prompt: prompt, - cwd: cwd, - ); - } - return [ - 'run', - '--format', - 'default', - if (cwd.trim().isNotEmpty) ...['--dir', cwd.trim()], - if (model.trim().isNotEmpty) ...['-m', model.trim()], - prompt, - ]; - case SingleAgentProvider.auto: - return const []; - } - } - - bool _prefersOllamaLaunch({ - required SingleAgentProvider provider, - required String model, - }) { - if (model.trim().isEmpty) { - return false; - } - return provider == SingleAgentProvider.codex || - provider == SingleAgentProvider.opencode || - provider == SingleAgentProvider.claude; - } - - List _buildOllamaLaunchArgs({ - required SingleAgentProvider provider, - required String model, - required String prompt, - required String cwd, - }) { - final tool = provider.providerId; - final args = ['launch', tool, '--model', model.trim()]; - if (provider == SingleAgentProvider.claude) { - args.add('--yes'); - args.addAll(['--', '-p', prompt]); - return args; - } - if (provider == SingleAgentProvider.codex) { - args.addAll([ - '--', - 'exec', - '--skip-git-repo-check', - '--color', - 'never', - if (cwd.trim().isNotEmpty) ...['-C', cwd.trim()], - prompt, - ]); - return args; - } - if (provider == SingleAgentProvider.opencode) { - args.addAll([ - '--', - 'run', - '--format', - 'default', - if (cwd.trim().isNotEmpty) ...['--dir', cwd.trim()], - prompt, - ]); - return args; - } - args.addAll(['--', '-p', prompt]); - return args; - } - - Map _buildEnvVars({ - required SingleAgentProvider provider, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - required MultiAgentConfig config, - }) { - final baseEnv = {...Platform.environment}; - if (config.aiGatewayInjectionPolicy != AiGatewayInjectionPolicy.disabled && - aiGatewayBaseUrl.trim().isNotEmpty && - aiGatewayApiKey.trim().isNotEmpty) { - baseEnv['OPENAI_BASE_URL'] = aiGatewayBaseUrl.trim(); - baseEnv['OPENAI_API_KEY'] = aiGatewayApiKey.trim(); - baseEnv['OLLAMA_BASE_URL'] = aiGatewayBaseUrl.trim(); - baseEnv['OLLAMA_HOST'] = aiGatewayBaseUrl.trim(); - if (provider == SingleAgentProvider.claude) { - baseEnv['ANTHROPIC_BASE_URL'] = aiGatewayBaseUrl.trim(); - baseEnv['ANTHROPIC_AUTH_TOKEN'] = aiGatewayApiKey.trim(); - baseEnv['ANTHROPIC_API_KEY'] = aiGatewayApiKey.trim(); - } - return baseEnv; - } - final ollamaEndpoint = config.ollamaEndpoint.trim(); - if (ollamaEndpoint.isNotEmpty) { - baseEnv['OLLAMA_BASE_URL'] = ollamaEndpoint; - baseEnv['OLLAMA_HOST'] = ollamaEndpoint; - baseEnv['OPENAI_API_KEY'] = 'ollama'; - baseEnv['OPENAI_BASE_URL'] = ollamaEndpoint.endsWith('/v1') - ? ollamaEndpoint - : '$ollamaEndpoint/v1'; - } - if (provider == SingleAgentProvider.claude || - provider == SingleAgentProvider.codex) { - baseEnv['ANTHROPIC_AUTH_TOKEN'] = 'ollama'; - baseEnv['ANTHROPIC_API_KEY'] = ''; - baseEnv['ANTHROPIC_BASE_URL'] = ollamaEndpoint; - } - return baseEnv; + final normalizedMessage = message.toLowerCase(); + return normalizedMessage.contains('timeout') || + normalizedMessage.contains('unavailable') || + normalizedMessage.contains('missing'); } String _augmentPrompt(SingleAgentRunRequest request) { @@ -515,24 +234,4 @@ class DefaultSingleAgentRunner implements SingleAgentRunner { .join('\n'); return 'User-selected local attachments:\n$attachmentLines\n\n${request.prompt}'; } - - bool _isLaunchFailureExit(int exitCode, String stderr) { - if (exitCode == 127 || exitCode == 9009 || exitCode == -1) { - return true; - } - final normalized = stderr.toLowerCase(); - return normalized.contains('not found') || - normalized.contains('no such file') || - normalized.contains('is not recognized'); - } - - bool _isLaunchFailureError(Object error) { - if (error is ProcessException) { - return true; - } - final normalized = error.toString().toLowerCase(); - return normalized.contains('not found') || - normalized.contains('no such file') || - normalized.contains('cannot find'); - } } diff --git a/test/runtime/no_direct_cli_execution_guard_suite.dart b/test/runtime/no_direct_cli_execution_guard_suite.dart new file mode 100644 index 00000000..dfbe0f81 --- /dev/null +++ b/test/runtime/no_direct_cli_execution_guard_suite.dart @@ -0,0 +1,57 @@ +@TestOn('vm') +library; + +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; + +void main() { + group('Desktop ACP guard', () { + test( + 'critical runtime client files must not execute external CLI directly', + () { + final blockedStartPattern = RegExp(r'\bProcess\.start\s*\('); + final blockedRunPattern = RegExp(r'\bProcess\.run\s*\('); + final allowedRunPatterns = [ + RegExp(r"Process\.run\(\s*'open'"), + RegExp(r"Process\.run\(\s*'cmd'"), + RegExp(r"Process\.run\(\s*'xdg-open'"), + ]; + const guardedFiles = [ + 'lib/app/app_controller_desktop.dart', + 'lib/runtime/single_agent_runner.dart', + 'lib/runtime/runtime_coordinator.dart', + 'lib/runtime/gateway_acp_client.dart', + ]; + + for (final relativePath in guardedFiles) { + final file = File(relativePath); + expect( + file.existsSync(), + isTrue, + reason: '$relativePath should exist', + ); + final content = file.readAsStringSync(); + expect( + blockedStartPattern.hasMatch(content), + isFalse, + reason: + '$relativePath contains forbidden local CLI execution: ${blockedStartPattern.pattern}', + ); + + for (final match in blockedRunPattern.allMatches(content)) { + final start = (match.start - 48).clamp(0, content.length); + final end = (match.end + 72).clamp(0, content.length); + final snippet = content.substring(start, end); + expect( + allowedRunPatterns.any((pattern) => pattern.hasMatch(snippet)), + isTrue, + reason: + '$relativePath contains non-whitelisted Process.run at offset ${match.start}', + ); + } + } + }, + ); + }); +} diff --git a/test/runtime/runtime_coordinator_suite.dart b/test/runtime/runtime_coordinator_suite.dart index 3036b870..52f1a732 100644 --- a/test/runtime/runtime_coordinator_suite.dart +++ b/test/runtime/runtime_coordinator_suite.dart @@ -187,7 +187,7 @@ void main() { ); test( - 'external mode resolves and starts codex process when binary exists', + 'external mode keeps gateway ready without starting local codex process', () async { codex.findResult = '/usr/local/bin/codex'; @@ -197,14 +197,14 @@ void main() { ); expect(coordinator.runtimeMode, CodeAgentRuntimeMode.externalCli); - expect(codex.findCalled, isTrue); - expect(codex.startCalled, isTrue); + expect(codex.findCalled, isFalse); + expect(codex.startCalled, isFalse); expect(modeSwitcher.currentMode, GatewayMode.remote); }, ); test( - 'external mode falls back to offline when codex binary missing', + 'external mode no longer forces offline when codex binary is missing', () async { codex.findResult = null; @@ -213,10 +213,10 @@ void main() { runtimeMode: CodeAgentRuntimeMode.externalCli, ); - expect(codex.findCalled, isTrue); + expect(codex.findCalled, isFalse); expect(codex.startCalled, isFalse); - expect(modeSwitcher.offlineSwitchCalled, isTrue); - expect(modeSwitcher.currentMode, GatewayMode.offline); + expect(modeSwitcher.offlineSwitchCalled, isFalse); + expect(modeSwitcher.currentMode, GatewayMode.remote); }, ); });