diff --git a/lib/runtime/direct_single_agent_app_server_client.dart b/lib/runtime/direct_single_agent_app_server_client.dart index 6324e81d..60d4b00e 100644 --- a/lib/runtime/direct_single_agent_app_server_client.dart +++ b/lib/runtime/direct_single_agent_app_server_client.dart @@ -4,4 +4,7 @@ import 'dart:io'; import 'runtime_models.dart'; +part 'direct_single_agent_app_server_client_protocol.part.dart'; +part 'direct_single_agent_app_server_client_transport.part.dart'; +part 'direct_single_agent_app_server_client_helpers.part.dart'; part 'direct_single_agent_app_server_client_core.part.dart'; diff --git a/lib/runtime/direct_single_agent_app_server_client_core.part.dart b/lib/runtime/direct_single_agent_app_server_client_core.part.dart index 3f1eb996..30956f60 100644 --- a/lib/runtime/direct_single_agent_app_server_client_core.part.dart +++ b/lib/runtime/direct_single_agent_app_server_client_core.part.dart @@ -1,152 +1,5 @@ part of 'direct_single_agent_app_server_client.dart'; -class DirectSingleAgentCapabilities { - const DirectSingleAgentCapabilities({ - required this.available, - required this.supportedProviders, - required this.endpoint, - this.errorMessage, - }); - - const DirectSingleAgentCapabilities.unavailable({ - required this.endpoint, - this.errorMessage, - }) : available = false, - supportedProviders = const []; - - final bool available; - final List supportedProviders; - final String endpoint; - final String? errorMessage; - - bool get supportsCodex => supportsProvider(SingleAgentProvider.codex); - - bool supportsProvider(SingleAgentProvider provider) => - supportedProviders.contains(provider); -} - -class DirectSingleAgentRunResult { - const DirectSingleAgentRunResult({ - required this.success, - required this.output, - required this.errorMessage, - this.aborted = false, - this.resolvedModel = '', - this.resolvedWorkingDirectory = '', - this.resolvedWorkspaceRefKind, - }); - - final bool success; - final String output; - final String errorMessage; - final bool aborted; - final String resolvedModel; - final String resolvedWorkingDirectory; - final WorkspaceRefKind? resolvedWorkspaceRefKind; -} - -class DirectSingleAgentRunRequest { - const DirectSingleAgentRunRequest({ - required this.sessionId, - required this.provider, - required this.prompt, - required this.model, - required this.workingDirectory, - required this.gatewayToken, - this.selectedSkills = const [], - this.onOutput, - }); - - final String sessionId; - final SingleAgentProvider provider; - final String prompt; - final String model; - final String workingDirectory; - final String gatewayToken; - final List selectedSkills; - final void Function(String text)? onOutput; -} - -enum DirectSingleAgentEndpointMode { - wsLocal, - wss, - httpLocal, - https, - unsupported, -} - -enum _DirectSingleAgentTransportKind { websocketAppServer, restSessionApi } - -class DirectSingleAgentEndpointDescriptor { - const DirectSingleAgentEndpointDescriptor({ - required this.mode, - required this.baseUri, - this.websocketUri, - }); - - final DirectSingleAgentEndpointMode mode; - final Uri? baseUri; - final Uri? websocketUri; - - bool get isSupported => mode != DirectSingleAgentEndpointMode.unsupported; - - bool get prefersWebSocket => - mode == DirectSingleAgentEndpointMode.wsLocal || - mode == DirectSingleAgentEndpointMode.wss; - - bool get allowsRest => - mode == DirectSingleAgentEndpointMode.httpLocal || - mode == DirectSingleAgentEndpointMode.https; - - static DirectSingleAgentEndpointDescriptor describe(Uri? endpoint) { - if (endpoint == null) { - return const DirectSingleAgentEndpointDescriptor( - mode: DirectSingleAgentEndpointMode.unsupported, - baseUri: null, - ); - } - final scheme = endpoint.scheme.toLowerCase(); - final normalizedBase = endpoint.replace( - path: '', - query: null, - fragment: null, - ); - final isLocal = _isLocalHost(endpoint.host); - if (scheme == 'ws' && isLocal) { - return DirectSingleAgentEndpointDescriptor( - mode: DirectSingleAgentEndpointMode.wsLocal, - baseUri: normalizedBase, - websocketUri: normalizedBase, - ); - } - if (scheme == 'wss') { - return DirectSingleAgentEndpointDescriptor( - mode: DirectSingleAgentEndpointMode.wss, - baseUri: normalizedBase, - websocketUri: normalizedBase, - ); - } - if (scheme == 'http' && isLocal) { - return DirectSingleAgentEndpointDescriptor( - mode: DirectSingleAgentEndpointMode.httpLocal, - baseUri: normalizedBase, - websocketUri: normalizedBase.replace(scheme: 'ws'), - ); - } - if (scheme == 'https') { - return DirectSingleAgentEndpointDescriptor( - mode: DirectSingleAgentEndpointMode.https, - baseUri: normalizedBase, - websocketUri: normalizedBase.replace(scheme: 'wss'), - ); - } - return DirectSingleAgentEndpointDescriptor( - mode: DirectSingleAgentEndpointMode.unsupported, - baseUri: normalizedBase, - ); - } -} - class DirectSingleAgentAppServerClient { DirectSingleAgentAppServerClient({required this.endpointResolver}); @@ -353,1064 +206,3 @@ class DirectSingleAgentAppServerClient { ); } } - -class _ResolvedSingleAgentTransport { - const _ResolvedSingleAgentTransport({ - required this.kind, - required this.endpoint, - required this.workspaceRefKind, - this.websocket, - this.rest, - }); - - final _DirectSingleAgentTransportKind kind; - final Uri endpoint; - final WorkspaceRefKind workspaceRefKind; - final _DirectSingleAgentWebSocketTransport? websocket; - final _DirectSingleAgentRestTransport? rest; -} - -class _ResolvedDirectThread { - const _ResolvedDirectThread({ - required this.threadId, - this.workingDirectory = '', - }); - - final String threadId; - final String workingDirectory; -} - -class _DirectSingleAgentWebSocketTransport { - final Map _activeConnections = - {}; - final Map _threadIds = {}; - final Map _threadWorkingDirectories = {}; - final Set _abortedSessions = {}; - - Future probe(Uri endpoint, {required String gatewayToken}) async { - _DirectAppServerConnection? connection; - try { - connection = await _DirectAppServerConnection.connect( - endpoint, - gatewayToken: gatewayToken, - ); - await connection.initialize(); - } finally { - await connection?.close(); - } - } - - Future run( - DirectSingleAgentRunRequest request, { - required Uri endpoint, - required WorkspaceRefKind workspaceRefKind, - }) async { - final normalizedSessionId = request.sessionId.trim(); - if (normalizedSessionId.isEmpty) { - return const DirectSingleAgentRunResult( - success: false, - output: '', - errorMessage: 'Single-agent session id is missing.', - ); - } - - _abortedSessions.remove(normalizedSessionId); - final connection = await _DirectAppServerConnection.connect( - endpoint, - gatewayToken: request.gatewayToken, - ); - _activeConnections[normalizedSessionId] = connection; - - try { - await connection.initialize(); - final resolvedThread = await _ensureThread( - connection, - sessionId: normalizedSessionId, - workingDirectory: request.workingDirectory, - model: request.model, - ); - final threadId = resolvedThread.threadId; - final resolvedWorkingDirectory = resolvedThread.workingDirectory.trim(); - - final output = StringBuffer(); - String resolvedModel = ''; - final completion = Completer(); - late final StreamSubscription> subscription; - subscription = connection.notifications.listen( - (notification) { - final method = notification['method']?.toString().trim() ?? ''; - final params = _asMap(notification['params']); - if (params['threadId']?.toString() != threadId) { - return; - } - if (method == 'item/agentMessage/delta') { - final delta = params['delta']?.toString() ?? ''; - if (delta.isNotEmpty) { - output.write(delta); - request.onOutput?.call(delta); - } - return; - } - if (method == 'turn/completed' && !completion.isCompleted) { - completion.complete( - DirectSingleAgentRunResult( - success: true, - output: output.toString(), - errorMessage: '', - resolvedModel: resolvedModel, - resolvedWorkingDirectory: resolvedWorkingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ), - ); - return; - } - if ((method == 'turn/failed' || method == 'turn/error') && - !completion.isCompleted) { - final aborted = - _abortedSessions.contains(normalizedSessionId) || - (params['message']?.toString().toLowerCase().contains( - 'abort', - ) ?? - false); - completion.complete( - DirectSingleAgentRunResult( - success: false, - output: output.toString(), - aborted: aborted, - resolvedModel: resolvedModel, - resolvedWorkingDirectory: resolvedWorkingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - errorMessage: - params['message']?.toString() ?? - params['error']?.toString() ?? - 'Single-agent app-server turn failed.', - ), - ); - } - }, - onError: (Object error, StackTrace stackTrace) { - if (!completion.isCompleted) { - completion.complete( - DirectSingleAgentRunResult( - success: false, - output: output.toString(), - errorMessage: error.toString(), - aborted: _abortedSessions.contains(normalizedSessionId), - resolvedModel: resolvedModel, - resolvedWorkingDirectory: resolvedWorkingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ), - ); - } - }, - onDone: () { - if (!completion.isCompleted) { - completion.complete( - DirectSingleAgentRunResult( - success: false, - output: output.toString(), - errorMessage: _abortedSessions.contains(normalizedSessionId) - ? 'Single-agent app-server run aborted.' - : 'Single-agent app-server connection closed before completion.', - aborted: _abortedSessions.contains(normalizedSessionId), - resolvedModel: resolvedModel, - resolvedWorkingDirectory: resolvedWorkingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ), - ); - } - }, - ); - - try { - final input = >[ - {'type': 'text', 'text': request.prompt}, - for (final skill in request.selectedSkills) - if (skill.label.trim().isNotEmpty && - skill.sourcePath.trim().isNotEmpty) - { - 'type': 'skill', - 'name': skill.label.trim(), - 'path': skill.sourcePath.trim(), - }, - ]; - final started = await connection.request( - 'turn/start', - params: {'threadId': threadId, 'input': input}, - ); - resolvedModel = _extractModel(started) ?? resolvedModel; - return await completion.future.timeout( - const Duration(minutes: 10), - onTimeout: () => DirectSingleAgentRunResult( - success: false, - output: output.toString(), - errorMessage: 'Single-agent app-server request timed out.', - aborted: _abortedSessions.contains(normalizedSessionId), - resolvedModel: resolvedModel, - resolvedWorkingDirectory: resolvedWorkingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ), - ); - } finally { - await subscription.cancel(); - } - } catch (error) { - return DirectSingleAgentRunResult( - success: false, - output: '', - errorMessage: error.toString(), - aborted: _abortedSessions.contains(normalizedSessionId), - resolvedModel: '', - resolvedWorkingDirectory: request.workingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ); - } finally { - _activeConnections.remove(normalizedSessionId); - await connection.close(); - _abortedSessions.remove(normalizedSessionId); - } - } - - Future abort(String sessionId) async { - final normalizedSessionId = sessionId.trim(); - if (normalizedSessionId.isEmpty) { - return; - } - _abortedSessions.add(normalizedSessionId); - final connection = _activeConnections[normalizedSessionId]; - final threadId = _threadIds[normalizedSessionId]; - if (connection == null || threadId == null || threadId.isEmpty) { - return; - } - try { - await connection.request( - 'turn/interrupt', - params: {'threadId': threadId}, - ); - } catch (_) { - // Best effort only. - } - await connection.close(); - } - - Future dispose() async { - final connections = _activeConnections.values.toList(growable: false); - _activeConnections.clear(); - for (final connection in connections) { - await connection.close(); - } - } - - Future<_ResolvedDirectThread> _ensureThread( - _DirectAppServerConnection connection, { - required String sessionId, - required String workingDirectory, - required String model, - }) async { - final normalizedWorkingDirectory = workingDirectory.trim(); - final existingThreadId = _threadIds[sessionId]?.trim() ?? ''; - final existingWorkingDirectory = - _threadWorkingDirectories[sessionId]?.trim() ?? ''; - final canReuseExistingThread = - existingThreadId.isNotEmpty && - (normalizedWorkingDirectory.isEmpty || - (existingWorkingDirectory.isNotEmpty && - existingWorkingDirectory == normalizedWorkingDirectory)); - if (existingThreadId.isNotEmpty) { - if (!canReuseExistingThread) { - _threadIds.remove(sessionId); - _threadWorkingDirectories.remove(sessionId); - } - } - if (canReuseExistingThread) { - try { - final resumed = await connection.request( - 'thread/resume', - params: { - 'threadId': existingThreadId, - if (normalizedWorkingDirectory.isNotEmpty) - 'cwd': normalizedWorkingDirectory, - }, - ); - final resumedId = _extractThreadId(resumed) ?? existingThreadId; - final resumedWorkingDirectory = - _extractThreadPath(resumed)?.trim() ?? normalizedWorkingDirectory; - _threadIds[sessionId] = resumedId; - if (resumedWorkingDirectory.isNotEmpty) { - _threadWorkingDirectories[sessionId] = resumedWorkingDirectory; - } - return _ResolvedDirectThread( - threadId: resumedId, - workingDirectory: resumedWorkingDirectory, - ); - } catch (_) { - _threadIds.remove(sessionId); - _threadWorkingDirectories.remove(sessionId); - } - } - - final created = await connection.request( - 'thread/start', - params: { - if (normalizedWorkingDirectory.isNotEmpty) - 'cwd': normalizedWorkingDirectory, - if (model.trim().isNotEmpty) 'model': model.trim(), - }, - ); - final threadId = _extractThreadId(created) ?? ''; - if (threadId.isEmpty) { - throw StateError('Single-agent app-server returned an empty thread id.'); - } - final createdWorkingDirectory = - _extractThreadPath(created)?.trim() ?? normalizedWorkingDirectory; - _threadIds[sessionId] = threadId; - if (createdWorkingDirectory.isNotEmpty) { - _threadWorkingDirectories[sessionId] = createdWorkingDirectory; - } - return _ResolvedDirectThread( - threadId: threadId, - workingDirectory: createdWorkingDirectory, - ); - } -} - -class _DirectSingleAgentRestTransport { - final Map _restSessionIds = {}; - final Map _restSessionWorkingDirectories = {}; - final Set _abortedSessions = {}; - - Future probe(Uri base, {required String gatewayToken}) async { - await _fetchJson( - _buildRestUri(base, '/global/health'), - gatewayToken: gatewayToken, - ); - } - - Future run( - DirectSingleAgentRunRequest request, { - required Uri base, - required WorkspaceRefKind workspaceRefKind, - }) async { - final normalizedSessionId = request.sessionId.trim(); - if (normalizedSessionId.isEmpty) { - return const DirectSingleAgentRunResult( - success: false, - output: '', - errorMessage: 'Single-agent session id is missing.', - ); - } - - _abortedSessions.remove(normalizedSessionId); - final remoteSessionId = await _ensureRestSession( - base, - sessionId: normalizedSessionId, - workingDirectory: request.workingDirectory, - gatewayToken: request.gatewayToken, - ); - - final output = StringBuffer(); - final completion = Completer(); - String? activeAssistantMessageId; - String? lastAssistantText; - var busySeen = false; - - void completeFailure(String message) { - if (completion.isCompleted) { - return; - } - completion.complete( - DirectSingleAgentRunResult( - success: false, - output: output.toString(), - errorMessage: message, - aborted: _abortedSessions.contains(normalizedSessionId), - resolvedModel: request.model, - resolvedWorkingDirectory: request.workingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ), - ); - } - - final eventClient = HttpClient() - ..connectionTimeout = const Duration(seconds: 8); - late final HttpClientRequest eventRequest; - late final HttpClientResponse eventResponse; - StreamSubscription? lineSubscription; - - void completeSuccess() { - if (completion.isCompleted) { - return; - } - final resolvedOutput = output.toString().trim().isNotEmpty - ? output.toString() - : (lastAssistantText ?? ''); - if (resolvedOutput.trim().isEmpty) { - completeFailure( - 'OpenCode REST session completed without assistant content.', - ); - return; - } - completion.complete( - DirectSingleAgentRunResult( - success: true, - output: resolvedOutput, - errorMessage: '', - resolvedModel: request.model, - resolvedWorkingDirectory: request.workingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ), - ); - } - - try { - final eventUri = _buildRestUri(base, '/global/event'); - eventRequest = await eventClient.getUrl(eventUri); - eventRequest.headers.set(HttpHeaders.acceptHeader, 'text/event-stream'); - final normalizedToken = request.gatewayToken.trim(); - if (normalizedToken.isNotEmpty) { - eventRequest.headers.set( - HttpHeaders.authorizationHeader, - 'Bearer $normalizedToken', - ); - } - eventResponse = await eventRequest.close(); - lineSubscription = eventResponse - .transform(utf8.decoder) - .transform(const LineSplitter()) - .listen( - (line) { - if (!line.startsWith('data: ')) { - return; - } - final event = _decodeMap(line.substring(6)); - final payload = _asMap(event['payload']); - final type = payload['type']?.toString().trim() ?? ''; - final properties = _asMap(payload['properties']); - if (properties['sessionID']?.toString().trim() != - remoteSessionId) { - return; - } - if (type == 'session.status') { - final status = _asMap(properties['status']); - final statusType = status['type']?.toString().trim() ?? ''; - if (statusType == 'busy') { - busySeen = true; - } - if (statusType == 'idle' && busySeen) { - completeSuccess(); - } - return; - } - if (type == 'session.idle' && busySeen) { - completeSuccess(); - return; - } - if (type == 'session.error' && !completion.isCompleted) { - final error = _asMap(properties['error']); - completeFailure( - error['message']?.toString() ?? - error['name']?.toString() ?? - 'OpenCode session failed.', - ); - return; - } - if (type == 'message.updated') { - final info = _asMap(properties['info']); - if (info['role']?.toString().trim() == 'assistant') { - activeAssistantMessageId = info['id']?.toString().trim(); - } - return; - } - if (type == 'message.part.delta') { - final part = _asMap(properties['part']); - if (activeAssistantMessageId != null && - part['messageID']?.toString().trim() == - activeAssistantMessageId) { - final delta = - properties['text']?.toString() ?? - properties['delta']?.toString() ?? - ''; - if (delta.isNotEmpty) { - output.write(delta); - request.onOutput?.call(delta); - } - } - return; - } - if (type == 'message.part.updated') { - final part = _asMap(properties['part']); - if (activeAssistantMessageId != null && - part['messageID']?.toString().trim() == - activeAssistantMessageId && - part['type']?.toString().trim() == 'text') { - lastAssistantText = part['text']?.toString(); - if ((lastAssistantText?.trim().isNotEmpty ?? false)) { - completeSuccess(); - } - } - } - }, - onError: (Object error, StackTrace stackTrace) {}, - onDone: () {}, - cancelOnError: true, - ); - - await _postJson( - _buildRestUri( - base, - '/session/$remoteSessionId/message', - queryParameters: { - 'directory': request.workingDirectory, - }, - ), - body: { - 'agent': 'build', - 'parts': >[ - {'type': 'text', 'text': request.prompt}, - ], - }, - gatewayToken: request.gatewayToken, - ); - unawaited( - _pollRestAssistantMessage( - base, - remoteSessionId: remoteSessionId, - workingDirectory: request.workingDirectory, - gatewayToken: request.gatewayToken, - onResolved: (text) { - if (text.trim().isNotEmpty) { - lastAssistantText = text; - if (output.toString().trim().isEmpty) { - output.write(text); - request.onOutput?.call(text); - } - completeSuccess(); - } - }, - onError: completeFailure, - ), - ); - - return await completion.future.timeout( - const Duration(minutes: 10), - onTimeout: () => DirectSingleAgentRunResult( - success: false, - output: output.toString(), - errorMessage: 'OpenCode REST request timed out.', - aborted: _abortedSessions.contains(normalizedSessionId), - resolvedModel: request.model, - resolvedWorkingDirectory: request.workingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ), - ); - } catch (error) { - return DirectSingleAgentRunResult( - success: false, - output: output.toString(), - errorMessage: error.toString(), - aborted: _abortedSessions.contains(normalizedSessionId), - resolvedModel: request.model, - resolvedWorkingDirectory: request.workingDirectory, - resolvedWorkspaceRefKind: workspaceRefKind, - ); - } finally { - unawaited(lineSubscription?.cancel()); - eventClient.close(force: true); - _abortedSessions.remove(normalizedSessionId); - } - } - - Future abort( - String sessionId, { - required List candidateBases, - }) async { - final normalizedSessionId = sessionId.trim(); - if (normalizedSessionId.isEmpty) { - return; - } - _abortedSessions.add(normalizedSessionId); - final restSessionId = _restSessionIds[normalizedSessionId]?.trim() ?? ''; - if (restSessionId.isEmpty) { - return; - } - for (final base in candidateBases) { - try { - await _postJson( - _buildRestUri(base, '/session/$restSessionId/abort'), - body: null, - gatewayToken: '', - ); - } catch (_) { - // Best effort only. - } - break; - } - } - - Future _ensureRestSession( - Uri base, { - required String sessionId, - required String workingDirectory, - required String gatewayToken, - }) async { - final normalizedWorkingDirectory = workingDirectory.trim(); - final existing = _restSessionIds[sessionId]?.trim() ?? ''; - if (existing.isNotEmpty) { - final existingWorkingDirectory = - _restSessionWorkingDirectories[sessionId]?.trim() ?? ''; - final canReuseExistingSession = - normalizedWorkingDirectory.isEmpty || - (existingWorkingDirectory.isNotEmpty && - existingWorkingDirectory == normalizedWorkingDirectory); - if (canReuseExistingSession) { - return existing; - } - _restSessionIds.remove(sessionId); - _restSessionWorkingDirectories.remove(sessionId); - } - final created = await _postJson( - _buildRestUri( - base, - '/session', - queryParameters: { - 'directory': normalizedWorkingDirectory, - }, - ), - body: {'title': sessionId}, - gatewayToken: gatewayToken, - ); - final createdId = created['id']?.toString().trim() ?? ''; - if (createdId.isEmpty) { - throw StateError('OpenCode REST endpoint returned an empty session id.'); - } - _restSessionIds[sessionId] = createdId; - if (normalizedWorkingDirectory.isNotEmpty) { - _restSessionWorkingDirectories[sessionId] = normalizedWorkingDirectory; - } - return createdId; - } - - Future _pollRestAssistantMessage( - Uri base, { - required String remoteSessionId, - required String workingDirectory, - required String gatewayToken, - required void Function(String text) onResolved, - required void Function(String message) onError, - }) async { - String? previousText; - var stableCount = 0; - for (var attempt = 0; attempt < 100; attempt++) { - try { - final items = await _fetchJsonList( - _buildRestUri( - base, - '/session/$remoteSessionId/message', - queryParameters: { - 'directory': workingDirectory, - 'limit': '20', - }, - ), - gatewayToken: gatewayToken, - ); - final text = _latestAssistantTextFromRestMessages(items); - if (text.trim().isNotEmpty) { - if (text == previousText) { - stableCount += 1; - } else { - previousText = text; - stableCount = 1; - } - if (stableCount >= 2) { - onResolved(text); - return; - } - } - } catch (error) { - onError(error.toString()); - return; - } - await Future.delayed(const Duration(milliseconds: 200)); - } - onError('OpenCode REST session completed without assistant content.'); - } - - String _latestAssistantTextFromRestMessages(List items) { - for (final raw in items.reversed) { - final item = _asMap(raw); - final info = _asMap(item['info']); - if (info['role']?.toString().trim() != 'assistant') { - continue; - } - final parts = item['parts']; - if (parts is! List) { - continue; - } - for (final rawPart in parts) { - final part = _asMap(rawPart); - if (part['type']?.toString().trim() == 'text') { - final text = part['text']?.toString() ?? ''; - if (text.trim().isNotEmpty) { - return text; - } - } - } - } - return ''; - } -} - -class _DirectAppServerConnection { - _DirectAppServerConnection(this._socket); - - final WebSocket _socket; - final StreamController> _notifications = - StreamController>.broadcast(); - final Map>> _pendingRequests = - >>{}; - int _requestCounter = 0; - bool _initialized = false; - StreamSubscription? _subscription; - - Stream> get notifications => _notifications.stream; - - static Future<_DirectAppServerConnection> connect( - Uri endpoint, { - String gatewayToken = '', - }) async { - final headers = {}; - final normalizedToken = gatewayToken.trim(); - if (normalizedToken.isNotEmpty) { - headers[HttpHeaders.authorizationHeader] = 'Bearer $normalizedToken'; - } - final socket = - await WebSocket.connect( - endpoint.toString(), - headers: headers.isEmpty ? null : headers, - ).timeout( - const Duration(seconds: 8), - onTimeout: () => throw TimeoutException( - 'Single-agent app-server websocket connect timed out.', - ), - ); - final connection = _DirectAppServerConnection(socket); - connection._attach(); - return connection; - } - - Future initialize() async { - if (_initialized) { - return; - } - await request( - 'initialize', - params: const { - 'clientInfo': {'name': 'xworkmate', 'version': '0'}, - 'capabilities': { - 'optOutNotificationMethods': [], - }, - }, - ); - await notify('initialized', params: const {}); - _initialized = true; - } - - Future> request( - String method, { - Map params = const {}, - Duration timeout = const Duration(seconds: 60), - }) async { - final id = '${DateTime.now().microsecondsSinceEpoch}-${_requestCounter++}'; - final completer = Completer>(); - _pendingRequests[id] = completer; - _socket.add( - jsonEncode({ - 'jsonrpc': '2.0', - 'id': id, - 'method': method, - 'params': params, - }), - ); - return completer.future.timeout( - timeout, - onTimeout: () { - _pendingRequests.remove(id); - throw TimeoutException( - 'Single-agent app-server request $method timed out.', - ); - }, - ); - } - - Future notify( - String method, { - required Map params, - }) async { - _socket.add( - jsonEncode({ - 'jsonrpc': '2.0', - 'method': method, - 'params': params, - }), - ); - } - - void _attach() { - _subscription = _socket.listen( - (dynamic raw) { - final message = _decodeMap(raw); - final id = message['id']?.toString(); - if (id != null && message.containsKey('result')) { - final completer = _pendingRequests.remove(id); - if (completer != null && !completer.isCompleted) { - completer.complete(_asMap(message['result'])); - } - return; - } - if (id != null && message.containsKey('error')) { - final completer = _pendingRequests.remove(id); - if (completer != null && !completer.isCompleted) { - final error = _asMap(message['error']); - completer.completeError( - StateError( - error['message']?.toString() ?? - 'Single-agent app-server request failed.', - ), - ); - } - return; - } - if (message.containsKey('method')) { - _notifications.add(message); - } - }, - onError: (Object error, StackTrace stackTrace) { - for (final completer in _pendingRequests.values) { - if (!completer.isCompleted) { - completer.completeError(error); - } - } - _pendingRequests.clear(); - _notifications.addError(error, stackTrace); - }, - onDone: () { - final error = StateError( - 'Single-agent app-server websocket closed unexpectedly.', - ); - for (final completer in _pendingRequests.values) { - if (!completer.isCompleted) { - completer.completeError(error); - } - } - _pendingRequests.clear(); - if (!_notifications.isClosed) { - unawaited(_notifications.close()); - } - }, - cancelOnError: true, - ); - } - - Future close() async { - await _subscription?.cancel(); - _subscription = null; - for (final completer in _pendingRequests.values) { - if (!completer.isCompleted) { - completer.completeError( - StateError('Single-agent app-server connection closed.'), - ); - } - } - _pendingRequests.clear(); - if (!_notifications.isClosed) { - await _notifications.close(); - } - try { - await _socket.close(); - } catch (_) { - // Best effort only. - } - } -} - -Uri _buildRestUri( - Uri base, - String path, { - Map? queryParameters, -}) { - final normalizedPath = path.startsWith('/') ? path : '/$path'; - return base.replace( - path: normalizedPath, - queryParameters: queryParameters, - fragment: null, - ); -} - -Future> _fetchJson( - Uri uri, { - required String gatewayToken, -}) async { - final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); - try { - final request = await client.getUrl(uri); - final normalizedToken = gatewayToken.trim(); - if (normalizedToken.isNotEmpty) { - request.headers.set( - HttpHeaders.authorizationHeader, - 'Bearer $normalizedToken', - ); - } - final response = await request.close(); - final body = await response.transform(utf8.decoder).join(); - return _decodeMap(body); - } finally { - client.close(force: true); - } -} - -Future> _postJson( - Uri uri, { - required Object? body, - required String gatewayToken, -}) async { - final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); - try { - final request = await client.postUrl(uri); - request.headers.set( - HttpHeaders.contentTypeHeader, - 'application/json; charset=utf-8', - ); - final normalizedToken = gatewayToken.trim(); - if (normalizedToken.isNotEmpty) { - request.headers.set( - HttpHeaders.authorizationHeader, - 'Bearer $normalizedToken', - ); - } - if (body != null) { - request.add(utf8.encode(jsonEncode(body))); - } - final response = await request.close(); - final text = await response.transform(utf8.decoder).join(); - if (text.trim().isEmpty) { - return const {}; - } - return _decodeMap(text); - } finally { - client.close(force: true); - } -} - -Future> _fetchJsonList( - Uri uri, { - required String gatewayToken, -}) async { - final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); - try { - final request = await client.getUrl(uri); - final normalizedToken = gatewayToken.trim(); - if (normalizedToken.isNotEmpty) { - request.headers.set( - HttpHeaders.authorizationHeader, - 'Bearer $normalizedToken', - ); - } - final response = await request.close(); - final body = await response.transform(utf8.decoder).join(); - final decoded = jsonDecode(body); - if (decoded is List) { - return decoded; - } - if (decoded is List) { - return decoded.cast(); - } - return const []; - } finally { - client.close(force: true); - } -} - -String? _extractThreadId(Map payload) { - final topLevelId = payload['id']?.toString().trim() ?? ''; - if (topLevelId.isNotEmpty) { - return topLevelId; - } - final thread = _asMap(payload['thread']); - final nestedId = thread['id']?.toString().trim() ?? ''; - if (nestedId.isNotEmpty) { - return nestedId; - } - return null; -} - -String? _extractModel(Map payload) { - final model = payload['model']?.toString().trim() ?? ''; - if (model.isNotEmpty) { - return model; - } - return null; -} - -String? _extractThreadPath(Map payload) { - final directPath = payload['path']?.toString().trim() ?? ''; - if (directPath.isNotEmpty) { - return directPath; - } - final thread = _asMap(payload['thread']); - final nestedPath = thread['path']?.toString().trim() ?? ''; - if (nestedPath.isNotEmpty) { - return nestedPath; - } - return null; -} - -Map _decodeMap(Object raw) { - if (raw is Map) { - return raw; - } - if (raw is Map) { - return raw.cast(); - } - final decoded = jsonDecode(raw.toString()); - if (decoded is Map) { - return decoded; - } - if (decoded is Map) { - return decoded.cast(); - } - return const {}; -} - -Map _asMap(Object? value) { - if (value is Map) { - return value; - } - if (value is Map) { - return value.cast(); - } - return const {}; -} - -bool _isLocalHost(String host) { - final normalized = host.trim().toLowerCase(); - if (normalized.isEmpty || - normalized == 'localhost' || - normalized == '127.0.0.1' || - normalized == '::1') { - return true; - } - final address = InternetAddress.tryParse(normalized); - return address?.isLoopback ?? false; -} - -WorkspaceRefKind _workspaceRefKindForEndpointMode( - DirectSingleAgentEndpointMode mode, -) { - return switch (mode) { - DirectSingleAgentEndpointMode.wsLocal || - DirectSingleAgentEndpointMode.httpLocal => WorkspaceRefKind.localPath, - DirectSingleAgentEndpointMode.wss || - DirectSingleAgentEndpointMode.https => WorkspaceRefKind.remotePath, - DirectSingleAgentEndpointMode.unsupported => WorkspaceRefKind.localPath, - }; -} diff --git a/lib/runtime/direct_single_agent_app_server_client_helpers.part.dart b/lib/runtime/direct_single_agent_app_server_client_helpers.part.dart new file mode 100644 index 00000000..278518ba --- /dev/null +++ b/lib/runtime/direct_single_agent_app_server_client_helpers.part.dart @@ -0,0 +1,183 @@ +part of 'direct_single_agent_app_server_client.dart'; + +Uri _buildRestUri( + Uri base, + String path, { + Map? queryParameters, +}) { + final normalizedPath = path.startsWith('/') ? path : '/$path'; + return base.replace( + path: normalizedPath, + queryParameters: queryParameters, + fragment: null, + ); +} + +Future> _fetchJson( + Uri uri, { + required String gatewayToken, +}) async { + final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); + try { + final request = await client.getUrl(uri); + final normalizedToken = gatewayToken.trim(); + if (normalizedToken.isNotEmpty) { + request.headers.set( + HttpHeaders.authorizationHeader, + 'Bearer $normalizedToken', + ); + } + final response = await request.close(); + final body = await response.transform(utf8.decoder).join(); + return _decodeMap(body); + } finally { + client.close(force: true); + } +} + +Future> _postJson( + Uri uri, { + required Object? body, + required String gatewayToken, +}) async { + final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); + try { + final request = await client.postUrl(uri); + request.headers.set( + HttpHeaders.contentTypeHeader, + 'application/json; charset=utf-8', + ); + final normalizedToken = gatewayToken.trim(); + if (normalizedToken.isNotEmpty) { + request.headers.set( + HttpHeaders.authorizationHeader, + 'Bearer $normalizedToken', + ); + } + if (body != null) { + request.add(utf8.encode(jsonEncode(body))); + } + final response = await request.close(); + final text = await response.transform(utf8.decoder).join(); + if (text.trim().isEmpty) { + return const {}; + } + return _decodeMap(text); + } finally { + client.close(force: true); + } +} + +Future> _fetchJsonList( + Uri uri, { + required String gatewayToken, +}) async { + final client = HttpClient()..connectionTimeout = const Duration(seconds: 8); + try { + final request = await client.getUrl(uri); + final normalizedToken = gatewayToken.trim(); + if (normalizedToken.isNotEmpty) { + request.headers.set( + HttpHeaders.authorizationHeader, + 'Bearer $normalizedToken', + ); + } + final response = await request.close(); + final body = await response.transform(utf8.decoder).join(); + final decoded = jsonDecode(body); + if (decoded is List) { + return decoded; + } + if (decoded is List) { + return decoded.cast(); + } + return const []; + } finally { + client.close(force: true); + } +} + +String? _extractThreadId(Map payload) { + final topLevelId = payload['id']?.toString().trim() ?? ''; + if (topLevelId.isNotEmpty) { + return topLevelId; + } + final thread = _asMap(payload['thread']); + final nestedId = thread['id']?.toString().trim() ?? ''; + if (nestedId.isNotEmpty) { + return nestedId; + } + return null; +} + +String? _extractModel(Map payload) { + final model = payload['model']?.toString().trim() ?? ''; + if (model.isNotEmpty) { + return model; + } + return null; +} + +String? _extractThreadPath(Map payload) { + final directPath = payload['path']?.toString().trim() ?? ''; + if (directPath.isNotEmpty) { + return directPath; + } + final thread = _asMap(payload['thread']); + final nestedPath = thread['path']?.toString().trim() ?? ''; + if (nestedPath.isNotEmpty) { + return nestedPath; + } + return null; +} + +Map _decodeMap(Object raw) { + if (raw is Map) { + return raw; + } + if (raw is Map) { + return raw.cast(); + } + final decoded = jsonDecode(raw.toString()); + if (decoded is Map) { + return decoded; + } + if (decoded is Map) { + return decoded.cast(); + } + return const {}; +} + +Map _asMap(Object? value) { + if (value is Map) { + return value; + } + if (value is Map) { + return value.cast(); + } + return const {}; +} + +bool _isLocalHost(String host) { + final normalized = host.trim().toLowerCase(); + if (normalized.isEmpty || + normalized == 'localhost' || + normalized == '127.0.0.1' || + normalized == '::1') { + return true; + } + final address = InternetAddress.tryParse(normalized); + return address?.isLoopback ?? false; +} + +WorkspaceRefKind _workspaceRefKindForEndpointMode( + DirectSingleAgentEndpointMode mode, +) { + return switch (mode) { + DirectSingleAgentEndpointMode.wsLocal || + DirectSingleAgentEndpointMode.httpLocal => WorkspaceRefKind.localPath, + DirectSingleAgentEndpointMode.wss || + DirectSingleAgentEndpointMode.https => WorkspaceRefKind.remotePath, + DirectSingleAgentEndpointMode.unsupported => WorkspaceRefKind.localPath, + }; +} diff --git a/lib/runtime/direct_single_agent_app_server_client_protocol.part.dart b/lib/runtime/direct_single_agent_app_server_client_protocol.part.dart new file mode 100644 index 00000000..8eea4d62 --- /dev/null +++ b/lib/runtime/direct_single_agent_app_server_client_protocol.part.dart @@ -0,0 +1,149 @@ +part of 'direct_single_agent_app_server_client.dart'; + +class DirectSingleAgentCapabilities { + const DirectSingleAgentCapabilities({ + required this.available, + required this.supportedProviders, + required this.endpoint, + this.errorMessage, + }); + + const DirectSingleAgentCapabilities.unavailable({ + required this.endpoint, + this.errorMessage, + }) : available = false, + supportedProviders = const []; + + final bool available; + final List supportedProviders; + final String endpoint; + final String? errorMessage; + + bool get supportsCodex => supportsProvider(SingleAgentProvider.codex); + + bool supportsProvider(SingleAgentProvider provider) => + supportedProviders.contains(provider); +} + +class DirectSingleAgentRunResult { + const DirectSingleAgentRunResult({ + required this.success, + required this.output, + required this.errorMessage, + this.aborted = false, + this.resolvedModel = '', + this.resolvedWorkingDirectory = '', + this.resolvedWorkspaceRefKind, + }); + + final bool success; + final String output; + final String errorMessage; + final bool aborted; + final String resolvedModel; + final String resolvedWorkingDirectory; + final WorkspaceRefKind? resolvedWorkspaceRefKind; +} + +class DirectSingleAgentRunRequest { + const DirectSingleAgentRunRequest({ + required this.sessionId, + required this.provider, + required this.prompt, + required this.model, + required this.workingDirectory, + required this.gatewayToken, + this.selectedSkills = const [], + this.onOutput, + }); + + final String sessionId; + final SingleAgentProvider provider; + final String prompt; + final String model; + final String workingDirectory; + final String gatewayToken; + final List selectedSkills; + final void Function(String text)? onOutput; +} + +enum DirectSingleAgentEndpointMode { + wsLocal, + wss, + httpLocal, + https, + unsupported, +} + +enum _DirectSingleAgentTransportKind { websocketAppServer, restSessionApi } + +class DirectSingleAgentEndpointDescriptor { + const DirectSingleAgentEndpointDescriptor({ + required this.mode, + required this.baseUri, + this.websocketUri, + }); + + final DirectSingleAgentEndpointMode mode; + final Uri? baseUri; + final Uri? websocketUri; + + bool get isSupported => mode != DirectSingleAgentEndpointMode.unsupported; + + bool get prefersWebSocket => + mode == DirectSingleAgentEndpointMode.wsLocal || + mode == DirectSingleAgentEndpointMode.wss; + + bool get allowsRest => + mode == DirectSingleAgentEndpointMode.httpLocal || + mode == DirectSingleAgentEndpointMode.https; + + static DirectSingleAgentEndpointDescriptor describe(Uri? endpoint) { + if (endpoint == null) { + return const DirectSingleAgentEndpointDescriptor( + mode: DirectSingleAgentEndpointMode.unsupported, + baseUri: null, + ); + } + final scheme = endpoint.scheme.toLowerCase(); + final normalizedBase = endpoint.replace( + path: '', + query: null, + fragment: null, + ); + final isLocal = _isLocalHost(endpoint.host); + if (scheme == 'ws' && isLocal) { + return DirectSingleAgentEndpointDescriptor( + mode: DirectSingleAgentEndpointMode.wsLocal, + baseUri: normalizedBase, + websocketUri: normalizedBase, + ); + } + if (scheme == 'wss') { + return DirectSingleAgentEndpointDescriptor( + mode: DirectSingleAgentEndpointMode.wss, + baseUri: normalizedBase, + websocketUri: normalizedBase, + ); + } + if (scheme == 'http' && isLocal) { + return DirectSingleAgentEndpointDescriptor( + mode: DirectSingleAgentEndpointMode.httpLocal, + baseUri: normalizedBase, + websocketUri: normalizedBase.replace(scheme: 'ws'), + ); + } + if (scheme == 'https') { + return DirectSingleAgentEndpointDescriptor( + mode: DirectSingleAgentEndpointMode.https, + baseUri: normalizedBase, + websocketUri: normalizedBase.replace(scheme: 'wss'), + ); + } + return DirectSingleAgentEndpointDescriptor( + mode: DirectSingleAgentEndpointMode.unsupported, + baseUri: normalizedBase, + ); + } +} + diff --git a/lib/runtime/direct_single_agent_app_server_client_transport.part.dart b/lib/runtime/direct_single_agent_app_server_client_transport.part.dart new file mode 100644 index 00000000..7373d798 --- /dev/null +++ b/lib/runtime/direct_single_agent_app_server_client_transport.part.dart @@ -0,0 +1,881 @@ +part of 'direct_single_agent_app_server_client.dart'; + +class _ResolvedSingleAgentTransport { + const _ResolvedSingleAgentTransport({ + required this.kind, + required this.endpoint, + required this.workspaceRefKind, + this.websocket, + this.rest, + }); + + final _DirectSingleAgentTransportKind kind; + final Uri endpoint; + final WorkspaceRefKind workspaceRefKind; + final _DirectSingleAgentWebSocketTransport? websocket; + final _DirectSingleAgentRestTransport? rest; +} + +class _ResolvedDirectThread { + const _ResolvedDirectThread({ + required this.threadId, + this.workingDirectory = '', + }); + + final String threadId; + final String workingDirectory; +} + +class _DirectSingleAgentWebSocketTransport { + final Map _activeConnections = + {}; + final Map _threadIds = {}; + final Map _threadWorkingDirectories = {}; + final Set _abortedSessions = {}; + + Future probe(Uri endpoint, {required String gatewayToken}) async { + _DirectAppServerConnection? connection; + try { + connection = await _DirectAppServerConnection.connect( + endpoint, + gatewayToken: gatewayToken, + ); + await connection.initialize(); + } finally { + await connection?.close(); + } + } + + Future run( + DirectSingleAgentRunRequest request, { + required Uri endpoint, + required WorkspaceRefKind workspaceRefKind, + }) async { + final normalizedSessionId = request.sessionId.trim(); + if (normalizedSessionId.isEmpty) { + return const DirectSingleAgentRunResult( + success: false, + output: '', + errorMessage: 'Single-agent session id is missing.', + ); + } + + _abortedSessions.remove(normalizedSessionId); + final connection = await _DirectAppServerConnection.connect( + endpoint, + gatewayToken: request.gatewayToken, + ); + _activeConnections[normalizedSessionId] = connection; + + try { + await connection.initialize(); + final resolvedThread = await _ensureThread( + connection, + sessionId: normalizedSessionId, + workingDirectory: request.workingDirectory, + model: request.model, + ); + final threadId = resolvedThread.threadId; + final resolvedWorkingDirectory = resolvedThread.workingDirectory.trim(); + + final output = StringBuffer(); + String resolvedModel = ''; + final completion = Completer(); + late final StreamSubscription> subscription; + subscription = connection.notifications.listen( + (notification) { + final method = notification['method']?.toString().trim() ?? ''; + final params = _asMap(notification['params']); + if (params['threadId']?.toString() != threadId) { + return; + } + if (method == 'item/agentMessage/delta') { + final delta = params['delta']?.toString() ?? ''; + if (delta.isNotEmpty) { + output.write(delta); + request.onOutput?.call(delta); + } + return; + } + if (method == 'turn/completed' && !completion.isCompleted) { + completion.complete( + DirectSingleAgentRunResult( + success: true, + output: output.toString(), + errorMessage: '', + resolvedModel: resolvedModel, + resolvedWorkingDirectory: resolvedWorkingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ), + ); + return; + } + if ((method == 'turn/failed' || method == 'turn/error') && + !completion.isCompleted) { + final aborted = + _abortedSessions.contains(normalizedSessionId) || + (params['message']?.toString().toLowerCase().contains( + 'abort', + ) ?? + false); + completion.complete( + DirectSingleAgentRunResult( + success: false, + output: output.toString(), + aborted: aborted, + resolvedModel: resolvedModel, + resolvedWorkingDirectory: resolvedWorkingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + errorMessage: + params['message']?.toString() ?? + params['error']?.toString() ?? + 'Single-agent app-server turn failed.', + ), + ); + } + }, + onError: (Object error, StackTrace stackTrace) { + if (!completion.isCompleted) { + completion.complete( + DirectSingleAgentRunResult( + success: false, + output: output.toString(), + errorMessage: error.toString(), + aborted: _abortedSessions.contains(normalizedSessionId), + resolvedModel: resolvedModel, + resolvedWorkingDirectory: resolvedWorkingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ), + ); + } + }, + onDone: () { + if (!completion.isCompleted) { + completion.complete( + DirectSingleAgentRunResult( + success: false, + output: output.toString(), + errorMessage: _abortedSessions.contains(normalizedSessionId) + ? 'Single-agent app-server run aborted.' + : 'Single-agent app-server connection closed before completion.', + aborted: _abortedSessions.contains(normalizedSessionId), + resolvedModel: resolvedModel, + resolvedWorkingDirectory: resolvedWorkingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ), + ); + } + }, + ); + + try { + final input = >[ + {'type': 'text', 'text': request.prompt}, + for (final skill in request.selectedSkills) + if (skill.label.trim().isNotEmpty && + skill.sourcePath.trim().isNotEmpty) + { + 'type': 'skill', + 'name': skill.label.trim(), + 'path': skill.sourcePath.trim(), + }, + ]; + final started = await connection.request( + 'turn/start', + params: {'threadId': threadId, 'input': input}, + ); + resolvedModel = _extractModel(started) ?? resolvedModel; + return await completion.future.timeout( + const Duration(minutes: 10), + onTimeout: () => DirectSingleAgentRunResult( + success: false, + output: output.toString(), + errorMessage: 'Single-agent app-server request timed out.', + aborted: _abortedSessions.contains(normalizedSessionId), + resolvedModel: resolvedModel, + resolvedWorkingDirectory: resolvedWorkingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ), + ); + } finally { + await subscription.cancel(); + } + } catch (error) { + return DirectSingleAgentRunResult( + success: false, + output: '', + errorMessage: error.toString(), + aborted: _abortedSessions.contains(normalizedSessionId), + resolvedModel: '', + resolvedWorkingDirectory: request.workingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ); + } finally { + _activeConnections.remove(normalizedSessionId); + await connection.close(); + _abortedSessions.remove(normalizedSessionId); + } + } + + Future abort(String sessionId) async { + final normalizedSessionId = sessionId.trim(); + if (normalizedSessionId.isEmpty) { + return; + } + _abortedSessions.add(normalizedSessionId); + final connection = _activeConnections[normalizedSessionId]; + final threadId = _threadIds[normalizedSessionId]; + if (connection == null || threadId == null || threadId.isEmpty) { + return; + } + try { + await connection.request( + 'turn/interrupt', + params: {'threadId': threadId}, + ); + } catch (_) { + // Best effort only. + } + await connection.close(); + } + + Future dispose() async { + final connections = _activeConnections.values.toList(growable: false); + _activeConnections.clear(); + for (final connection in connections) { + await connection.close(); + } + } + + Future<_ResolvedDirectThread> _ensureThread( + _DirectAppServerConnection connection, { + required String sessionId, + required String workingDirectory, + required String model, + }) async { + final normalizedWorkingDirectory = workingDirectory.trim(); + final existingThreadId = _threadIds[sessionId]?.trim() ?? ''; + final existingWorkingDirectory = + _threadWorkingDirectories[sessionId]?.trim() ?? ''; + final canReuseExistingThread = + existingThreadId.isNotEmpty && + (normalizedWorkingDirectory.isEmpty || + (existingWorkingDirectory.isNotEmpty && + existingWorkingDirectory == normalizedWorkingDirectory)); + if (existingThreadId.isNotEmpty) { + if (!canReuseExistingThread) { + _threadIds.remove(sessionId); + _threadWorkingDirectories.remove(sessionId); + } + } + if (canReuseExistingThread) { + try { + final resumed = await connection.request( + 'thread/resume', + params: { + 'threadId': existingThreadId, + if (normalizedWorkingDirectory.isNotEmpty) + 'cwd': normalizedWorkingDirectory, + }, + ); + final resumedId = _extractThreadId(resumed) ?? existingThreadId; + final resumedWorkingDirectory = + _extractThreadPath(resumed)?.trim() ?? normalizedWorkingDirectory; + _threadIds[sessionId] = resumedId; + if (resumedWorkingDirectory.isNotEmpty) { + _threadWorkingDirectories[sessionId] = resumedWorkingDirectory; + } + return _ResolvedDirectThread( + threadId: resumedId, + workingDirectory: resumedWorkingDirectory, + ); + } catch (_) { + _threadIds.remove(sessionId); + _threadWorkingDirectories.remove(sessionId); + } + } + + final created = await connection.request( + 'thread/start', + params: { + if (normalizedWorkingDirectory.isNotEmpty) + 'cwd': normalizedWorkingDirectory, + if (model.trim().isNotEmpty) 'model': model.trim(), + }, + ); + final threadId = _extractThreadId(created) ?? ''; + if (threadId.isEmpty) { + throw StateError('Single-agent app-server returned an empty thread id.'); + } + final createdWorkingDirectory = + _extractThreadPath(created)?.trim() ?? normalizedWorkingDirectory; + _threadIds[sessionId] = threadId; + if (createdWorkingDirectory.isNotEmpty) { + _threadWorkingDirectories[sessionId] = createdWorkingDirectory; + } + return _ResolvedDirectThread( + threadId: threadId, + workingDirectory: createdWorkingDirectory, + ); + } +} + +class _DirectSingleAgentRestTransport { + final Map _restSessionIds = {}; + final Map _restSessionWorkingDirectories = {}; + final Set _abortedSessions = {}; + + Future probe(Uri base, {required String gatewayToken}) async { + await _fetchJson( + _buildRestUri(base, '/global/health'), + gatewayToken: gatewayToken, + ); + } + + Future run( + DirectSingleAgentRunRequest request, { + required Uri base, + required WorkspaceRefKind workspaceRefKind, + }) async { + final normalizedSessionId = request.sessionId.trim(); + if (normalizedSessionId.isEmpty) { + return const DirectSingleAgentRunResult( + success: false, + output: '', + errorMessage: 'Single-agent session id is missing.', + ); + } + + _abortedSessions.remove(normalizedSessionId); + final remoteSessionId = await _ensureRestSession( + base, + sessionId: normalizedSessionId, + workingDirectory: request.workingDirectory, + gatewayToken: request.gatewayToken, + ); + + final output = StringBuffer(); + final completion = Completer(); + String? activeAssistantMessageId; + String? lastAssistantText; + var busySeen = false; + + void completeFailure(String message) { + if (completion.isCompleted) { + return; + } + completion.complete( + DirectSingleAgentRunResult( + success: false, + output: output.toString(), + errorMessage: message, + aborted: _abortedSessions.contains(normalizedSessionId), + resolvedModel: request.model, + resolvedWorkingDirectory: request.workingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ), + ); + } + + final eventClient = HttpClient() + ..connectionTimeout = const Duration(seconds: 8); + late final HttpClientRequest eventRequest; + late final HttpClientResponse eventResponse; + StreamSubscription? lineSubscription; + + void completeSuccess() { + if (completion.isCompleted) { + return; + } + final resolvedOutput = output.toString().trim().isNotEmpty + ? output.toString() + : (lastAssistantText ?? ''); + if (resolvedOutput.trim().isEmpty) { + completeFailure( + 'OpenCode REST session completed without assistant content.', + ); + return; + } + completion.complete( + DirectSingleAgentRunResult( + success: true, + output: resolvedOutput, + errorMessage: '', + resolvedModel: request.model, + resolvedWorkingDirectory: request.workingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ), + ); + } + + try { + final eventUri = _buildRestUri(base, '/global/event'); + eventRequest = await eventClient.getUrl(eventUri); + eventRequest.headers.set(HttpHeaders.acceptHeader, 'text/event-stream'); + final normalizedToken = request.gatewayToken.trim(); + if (normalizedToken.isNotEmpty) { + eventRequest.headers.set( + HttpHeaders.authorizationHeader, + 'Bearer $normalizedToken', + ); + } + eventResponse = await eventRequest.close(); + lineSubscription = eventResponse + .transform(utf8.decoder) + .transform(const LineSplitter()) + .listen( + (line) { + if (!line.startsWith('data: ')) { + return; + } + final event = _decodeMap(line.substring(6)); + final payload = _asMap(event['payload']); + final type = payload['type']?.toString().trim() ?? ''; + final properties = _asMap(payload['properties']); + if (properties['sessionID']?.toString().trim() != + remoteSessionId) { + return; + } + if (type == 'session.status') { + final status = _asMap(properties['status']); + final statusType = status['type']?.toString().trim() ?? ''; + if (statusType == 'busy') { + busySeen = true; + } + if (statusType == 'idle' && busySeen) { + completeSuccess(); + } + return; + } + if (type == 'session.idle' && busySeen) { + completeSuccess(); + return; + } + if (type == 'session.error' && !completion.isCompleted) { + final error = _asMap(properties['error']); + completeFailure( + error['message']?.toString() ?? + error['name']?.toString() ?? + 'OpenCode session failed.', + ); + return; + } + if (type == 'message.updated') { + final info = _asMap(properties['info']); + if (info['role']?.toString().trim() == 'assistant') { + activeAssistantMessageId = info['id']?.toString().trim(); + } + return; + } + if (type == 'message.part.delta') { + final part = _asMap(properties['part']); + if (activeAssistantMessageId != null && + part['messageID']?.toString().trim() == + activeAssistantMessageId) { + final delta = + properties['text']?.toString() ?? + properties['delta']?.toString() ?? + ''; + if (delta.isNotEmpty) { + output.write(delta); + request.onOutput?.call(delta); + } + } + return; + } + if (type == 'message.part.updated') { + final part = _asMap(properties['part']); + if (activeAssistantMessageId != null && + part['messageID']?.toString().trim() == + activeAssistantMessageId && + part['type']?.toString().trim() == 'text') { + lastAssistantText = part['text']?.toString(); + if ((lastAssistantText?.trim().isNotEmpty ?? false)) { + completeSuccess(); + } + } + } + }, + onError: (Object error, StackTrace stackTrace) {}, + onDone: () {}, + cancelOnError: true, + ); + + await _postJson( + _buildRestUri( + base, + '/session/$remoteSessionId/message', + queryParameters: { + 'directory': request.workingDirectory, + }, + ), + body: { + 'agent': 'build', + 'parts': >[ + {'type': 'text', 'text': request.prompt}, + ], + }, + gatewayToken: request.gatewayToken, + ); + unawaited( + _pollRestAssistantMessage( + base, + remoteSessionId: remoteSessionId, + workingDirectory: request.workingDirectory, + gatewayToken: request.gatewayToken, + onResolved: (text) { + if (text.trim().isNotEmpty) { + lastAssistantText = text; + if (output.toString().trim().isEmpty) { + output.write(text); + request.onOutput?.call(text); + } + completeSuccess(); + } + }, + onError: completeFailure, + ), + ); + + return await completion.future.timeout( + const Duration(minutes: 10), + onTimeout: () => DirectSingleAgentRunResult( + success: false, + output: output.toString(), + errorMessage: 'OpenCode REST request timed out.', + aborted: _abortedSessions.contains(normalizedSessionId), + resolvedModel: request.model, + resolvedWorkingDirectory: request.workingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ), + ); + } catch (error) { + return DirectSingleAgentRunResult( + success: false, + output: output.toString(), + errorMessage: error.toString(), + aborted: _abortedSessions.contains(normalizedSessionId), + resolvedModel: request.model, + resolvedWorkingDirectory: request.workingDirectory, + resolvedWorkspaceRefKind: workspaceRefKind, + ); + } finally { + unawaited(lineSubscription?.cancel()); + eventClient.close(force: true); + _abortedSessions.remove(normalizedSessionId); + } + } + + Future abort( + String sessionId, { + required List candidateBases, + }) async { + final normalizedSessionId = sessionId.trim(); + if (normalizedSessionId.isEmpty) { + return; + } + _abortedSessions.add(normalizedSessionId); + final restSessionId = _restSessionIds[normalizedSessionId]?.trim() ?? ''; + if (restSessionId.isEmpty) { + return; + } + for (final base in candidateBases) { + try { + await _postJson( + _buildRestUri(base, '/session/$restSessionId/abort'), + body: null, + gatewayToken: '', + ); + } catch (_) { + // Best effort only. + } + break; + } + } + + Future _ensureRestSession( + Uri base, { + required String sessionId, + required String workingDirectory, + required String gatewayToken, + }) async { + final normalizedWorkingDirectory = workingDirectory.trim(); + final existing = _restSessionIds[sessionId]?.trim() ?? ''; + if (existing.isNotEmpty) { + final existingWorkingDirectory = + _restSessionWorkingDirectories[sessionId]?.trim() ?? ''; + final canReuseExistingSession = + normalizedWorkingDirectory.isEmpty || + (existingWorkingDirectory.isNotEmpty && + existingWorkingDirectory == normalizedWorkingDirectory); + if (canReuseExistingSession) { + return existing; + } + _restSessionIds.remove(sessionId); + _restSessionWorkingDirectories.remove(sessionId); + } + final created = await _postJson( + _buildRestUri( + base, + '/session', + queryParameters: { + 'directory': normalizedWorkingDirectory, + }, + ), + body: {'title': sessionId}, + gatewayToken: gatewayToken, + ); + final createdId = created['id']?.toString().trim() ?? ''; + if (createdId.isEmpty) { + throw StateError('OpenCode REST endpoint returned an empty session id.'); + } + _restSessionIds[sessionId] = createdId; + if (normalizedWorkingDirectory.isNotEmpty) { + _restSessionWorkingDirectories[sessionId] = normalizedWorkingDirectory; + } + return createdId; + } + + Future _pollRestAssistantMessage( + Uri base, { + required String remoteSessionId, + required String workingDirectory, + required String gatewayToken, + required void Function(String text) onResolved, + required void Function(String message) onError, + }) async { + String? previousText; + var stableCount = 0; + for (var attempt = 0; attempt < 100; attempt++) { + try { + final items = await _fetchJsonList( + _buildRestUri( + base, + '/session/$remoteSessionId/message', + queryParameters: { + 'directory': workingDirectory, + 'limit': '20', + }, + ), + gatewayToken: gatewayToken, + ); + final text = _latestAssistantTextFromRestMessages(items); + if (text.trim().isNotEmpty) { + if (text == previousText) { + stableCount += 1; + } else { + previousText = text; + stableCount = 1; + } + if (stableCount >= 2) { + onResolved(text); + return; + } + } + } catch (error) { + onError(error.toString()); + return; + } + await Future.delayed(const Duration(milliseconds: 200)); + } + onError('OpenCode REST session completed without assistant content.'); + } + + String _latestAssistantTextFromRestMessages(List items) { + for (final raw in items.reversed) { + final item = _asMap(raw); + final info = _asMap(item['info']); + if (info['role']?.toString().trim() != 'assistant') { + continue; + } + final parts = item['parts']; + if (parts is! List) { + continue; + } + for (final rawPart in parts) { + final part = _asMap(rawPart); + if (part['type']?.toString().trim() == 'text') { + final text = part['text']?.toString() ?? ''; + if (text.trim().isNotEmpty) { + return text; + } + } + } + } + return ''; + } +} + +class _DirectAppServerConnection { + _DirectAppServerConnection(this._socket); + + final WebSocket _socket; + final StreamController> _notifications = + StreamController>.broadcast(); + final Map>> _pendingRequests = + >>{}; + int _requestCounter = 0; + bool _initialized = false; + StreamSubscription? _subscription; + + Stream> get notifications => _notifications.stream; + + static Future<_DirectAppServerConnection> connect( + Uri endpoint, { + String gatewayToken = '', + }) async { + final headers = {}; + final normalizedToken = gatewayToken.trim(); + if (normalizedToken.isNotEmpty) { + headers[HttpHeaders.authorizationHeader] = 'Bearer $normalizedToken'; + } + final socket = + await WebSocket.connect( + endpoint.toString(), + headers: headers.isEmpty ? null : headers, + ).timeout( + const Duration(seconds: 8), + onTimeout: () => throw TimeoutException( + 'Single-agent app-server websocket connect timed out.', + ), + ); + final connection = _DirectAppServerConnection(socket); + connection._attach(); + return connection; + } + + Future initialize() async { + if (_initialized) { + return; + } + await request( + 'initialize', + params: const { + 'clientInfo': {'name': 'xworkmate', 'version': '0'}, + 'capabilities': { + 'optOutNotificationMethods': [], + }, + }, + ); + await notify('initialized', params: const {}); + _initialized = true; + } + + Future> request( + String method, { + Map params = const {}, + Duration timeout = const Duration(seconds: 60), + }) async { + final id = '${DateTime.now().microsecondsSinceEpoch}-${_requestCounter++}'; + final completer = Completer>(); + _pendingRequests[id] = completer; + _socket.add( + jsonEncode({ + 'jsonrpc': '2.0', + 'id': id, + 'method': method, + 'params': params, + }), + ); + return completer.future.timeout( + timeout, + onTimeout: () { + _pendingRequests.remove(id); + throw TimeoutException( + 'Single-agent app-server request $method timed out.', + ); + }, + ); + } + + Future notify( + String method, { + required Map params, + }) async { + _socket.add( + jsonEncode({ + 'jsonrpc': '2.0', + 'method': method, + 'params': params, + }), + ); + } + + void _attach() { + _subscription = _socket.listen( + (dynamic raw) { + final message = _decodeMap(raw); + final id = message['id']?.toString(); + if (id != null && message.containsKey('result')) { + final completer = _pendingRequests.remove(id); + if (completer != null && !completer.isCompleted) { + completer.complete(_asMap(message['result'])); + } + return; + } + if (id != null && message.containsKey('error')) { + final completer = _pendingRequests.remove(id); + if (completer != null && !completer.isCompleted) { + final error = _asMap(message['error']); + completer.completeError( + StateError( + error['message']?.toString() ?? + 'Single-agent app-server request failed.', + ), + ); + } + return; + } + if (message.containsKey('method')) { + _notifications.add(message); + } + }, + onError: (Object error, StackTrace stackTrace) { + for (final completer in _pendingRequests.values) { + if (!completer.isCompleted) { + completer.completeError(error); + } + } + _pendingRequests.clear(); + _notifications.addError(error, stackTrace); + }, + onDone: () { + final error = StateError( + 'Single-agent app-server websocket closed unexpectedly.', + ); + for (final completer in _pendingRequests.values) { + if (!completer.isCompleted) { + completer.completeError(error); + } + } + _pendingRequests.clear(); + if (!_notifications.isClosed) { + unawaited(_notifications.close()); + } + }, + cancelOnError: true, + ); + } + + Future close() async { + await _subscription?.cancel(); + _subscription = null; + for (final completer in _pendingRequests.values) { + if (!completer.isCompleted) { + completer.completeError( + StateError('Single-agent app-server connection closed.'), + ); + } + } + _pendingRequests.clear(); + if (!_notifications.isClosed) { + await _notifications.close(); + } + try { + await _socket.close(); + } catch (_) { + // Best effort only. + } + } +} + diff --git a/lib/runtime/multi_agent_orchestrator.dart b/lib/runtime/multi_agent_orchestrator.dart index fe02aaf4..aeb13e2b 100644 --- a/lib/runtime/multi_agent_orchestrator.dart +++ b/lib/runtime/multi_agent_orchestrator.dart @@ -11,4 +11,7 @@ import 'aris_llm_chat_client.dart'; import 'multi_agent_frameworks.dart'; import 'runtime_models.dart'; +part 'multi_agent_orchestrator_protocol.part.dart'; +part 'multi_agent_orchestrator_workflow.part.dart'; +part 'multi_agent_orchestrator_support.part.dart'; part 'multi_agent_orchestrator_core.part.dart'; diff --git a/lib/runtime/multi_agent_orchestrator_core.part.dart b/lib/runtime/multi_agent_orchestrator_core.part.dart index 9ad3e99d..86be7a71 100644 --- a/lib/runtime/multi_agent_orchestrator_core.part.dart +++ b/lib/runtime/multi_agent_orchestrator_core.part.dart @@ -1,13 +1,5 @@ part of 'multi_agent_orchestrator.dart'; -typedef CliProcessStarter = - Future Function( - String executable, - List arguments, { - Map? environment, - String? workingDirectory, - }); - /// 多 Agent 协作编排器 /// /// 管理 Architect（调度/文档）→ Lead Engineer（主程）→ Worker/Review（并行 worker + 复审） @@ -428,1015 +420,6 @@ class MultiAgentOrchestrator extends ChangeNotifier { } } - /// 运行 Architect（调度/文档分析） - Future _runArchitect( - String task, { - required FrameworkPreset preset, - required List selectedSkills, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - final stopwatch = Stopwatch()..start(); - - try { - // 根据配置选择 Architect 工具 - if (_config.architectEnabled) { - final tool = await _resolveToolForRole( - MultiAgentRole.architect, - _config.architectTool, - ); - final instructionBlock = await preset.roleInstructionBlock( - role: MultiAgentRole.architect, - tool: tool, - selectedSkills: selectedSkills, - ); - final result = await _runCliPrompt( - role: MultiAgentRole.architect, - tool: tool, - model: _resolvedModelForRole( - MultiAgentRole.architect, - configuredModel: _config.architectModel, - ), - prompt: _buildArchitectPrompt(task, selectedSkills, instructionBlock), - cwd: '', - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - stopwatch.stop(); - - // 解析分解后的任务 - final tasks = _parseDecomposedTasks(result.output); - return ArchitectResult( - output: result.output, - decomposedTasks: tasks, - duration: stopwatch.elapsed, - ); - } else { - // Architect 被禁用，直接返回原任务作为单一子任务 - stopwatch.stop(); - return ArchitectResult( - output: task, - decomposedTasks: [ - SubTask( - id: '1', - description: task, - order: 1, - type: SubTaskType.implementation, - ), - ], - duration: stopwatch.elapsed, - ); - } - } catch (e) { - stopwatch.stop(); - rethrow; - } - } - - /// 运行 Lead Engineer（主实现） - Future _runEngineer( - List tasks, - String workingDirectory, - List attachments, { - required FrameworkPreset preset, - required List selectedSkills, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - final stopwatch = Stopwatch()..start(); - final tool = await _resolveToolForRole( - MultiAgentRole.engineer, - _config.engineerTool, - ); - final instructionBlock = await preset.roleInstructionBlock( - role: MultiAgentRole.engineer, - tool: tool, - selectedSkills: selectedSkills, - ); - - final taskList = tasks - .map((t) => '## ${t.order}. ${t.description}') - .join('\n\n'); - - final prompt = - ''' -$instructionBlock - -你是一个资深工程师，负责完成以下编码任务： - -### 任务列表 -$taskList - -### 工作目录 -$workingDirectory - -### 附件信息 -${attachments.map((a) => '- ${a.name}: ${a.description}').join('\n')} - -### 优先技能 -${selectedSkills.isEmpty ? '- 无' : selectedSkills.map((item) => '- $item').join('\n')} - -请完成这些任务，输出完整的代码实现。 -'''; - - final result = await _runCliPrompt( - role: MultiAgentRole.engineer, - tool: tool, - model: _resolvedModelForRole( - MultiAgentRole.engineer, - configuredModel: _config.engineerModel, - ), - prompt: prompt, - cwd: workingDirectory, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - stopwatch.stop(); - - return EngineerResult( - output: result.output, - codeOutput: result.output, - completedTasks: tasks, - duration: stopwatch.elapsed, - ); - } - - /// 运行 Worker/Review（代码审阅） - Future _runTester( - String codeOutput, { - required FrameworkPreset preset, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - final stopwatch = Stopwatch()..start(); - final tool = await _resolveToolForRole( - MultiAgentRole.testerDoc, - _config.testerTool, - ); - final instructionBlock = await preset.roleInstructionBlock( - role: MultiAgentRole.testerDoc, - tool: tool, - selectedSkills: const [], - ); - - final prompt = - ''' -$instructionBlock - -请审阅以下代码，并按以下格式输出： - -## 评分 (1-10) -[1-10 的分数，10 最高] - -## 问题列表 -[发现的问题，格式：- 问题描述 (严重程度: 高/中/低)] - -## 改进建议 -[具体的改进建议] - -## 测试用例 -```[语言] -[生成的测试用例代码] -``` - -## 文档建议 -[如有需要补充的文档说明] - -### 待审阅代码 -${codeOutput.length > 4000 ? '${codeOutput.substring(0, 4000)}\n...[代码已截断]' : codeOutput} -'''; - - final testerModel = _resolvedModelForRole( - MultiAgentRole.testerDoc, - configuredModel: _config.testerModel, - ); - final result = _config.usesAris && tool == 'claude' - ? await _runArisTesterViaClaudeReview( - model: testerModel, - prompt: prompt, - ) - : await _runCliPrompt( - role: MultiAgentRole.testerDoc, - tool: tool, - model: testerModel, - prompt: prompt, - cwd: '', - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - stopwatch.stop(); - - final score = _parseReviewScore(result.output); - final feedback = _extractFeedback(result.output); - - return TesterResult( - output: result.output, - score: score, - feedback: feedback, - duration: stopwatch.elapsed, - ); - } - - /// 运行修复（迭代循环中） - Future _runFix( - String originalCode, - String feedback, - String workingDirectory, { - required FrameworkPreset preset, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - final stopwatch = Stopwatch()..start(); - final tool = await _resolveToolForRole( - MultiAgentRole.engineer, - _config.engineerTool, - ); - final instructionBlock = await preset.roleInstructionBlock( - role: MultiAgentRole.engineer, - tool: tool, - selectedSkills: const [], - ); - - final prompt = - ''' -$instructionBlock - -你是一个资深工程师。请根据审阅反馈修复代码。 - -## 审阅反馈 -$feedback - -## 原始代码 -$originalCode - -请完成修复，输出修复后的完整代码。 -'''; - - final result = await _runCliPrompt( - role: MultiAgentRole.engineer, - tool: tool, - model: _resolvedModelForRole( - MultiAgentRole.engineer, - configuredModel: _config.engineerModel, - ), - prompt: prompt, - cwd: workingDirectory, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - stopwatch.stop(); - - return EngineerResult( - output: result.output, - codeOutput: result.output, - completedTasks: [], - duration: stopwatch.elapsed, - ); - } - - /// 通用的 CLI 进程执行方法 - Future _runCliPrompt({ - required MultiAgentRole role, - required String tool, - required String model, - required String prompt, - required String cwd, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - late final List args; - late final String command; - late final Map envVars; - final useOllamaLaunch = _prefersOllamaLaunch(tool: tool, model: model); - - switch (tool) { - case 'claude': - command = useOllamaLaunch ? 'ollama' : _resolveCliPath('claude'); - envVars = _buildCliEnvVars( - tool: tool, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - if (useOllamaLaunch) { - args = _buildOllamaLaunchArgs( - tool: tool, - model: model, - prompt: prompt, - cwd: cwd, - ); - } else if (model.isNotEmpty) { - args = ['--model', model, '-p', prompt]; - } else { - args = ['-p', prompt]; - } - break; - - case 'codex': - command = useOllamaLaunch ? 'ollama' : _resolveCliPath('codex'); - envVars = _buildCliEnvVars( - tool: tool, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - if (useOllamaLaunch) { - args = _buildOllamaLaunchArgs( - tool: tool, - model: model, - prompt: prompt, - cwd: cwd, - ); - } else if (model.isNotEmpty) { - args = [ - 'exec', - '--skip-git-repo-check', - '--color', - 'never', - if (cwd.isNotEmpty) ...['-C', cwd], - '-m', - model, - prompt, - ]; - } else { - args = [ - 'exec', - '--skip-git-repo-check', - '--color', - 'never', - if (cwd.isNotEmpty) ...['-C', cwd], - prompt, - ]; - } - break; - - case 'gemini': - command = _resolveCliPath('gemini'); - envVars = _buildCliEnvVars( - tool: tool, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - if (model.isNotEmpty) { - args = ['--model', model, '-p', prompt]; - } else { - args = ['-p', prompt]; - } - break; - - case 'opencode': - command = useOllamaLaunch ? 'ollama' : _resolveCliPath('opencode'); - envVars = _buildCliEnvVars( - tool: tool, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - args = useOllamaLaunch - ? _buildOllamaLaunchArgs( - tool: tool, - model: model, - prompt: prompt, - cwd: cwd, - ) - : [ - 'run', - '--format', - 'default', - if (cwd.isNotEmpty) ...['--dir', cwd], - if (model.isNotEmpty) ...['-m', model], - prompt, - ]; - break; - - default: - throw ArgumentError('Unknown tool: $tool'); - } - - final cliAvailable = await _binaryExists(command); - if (_config.usesAris && !cliAvailable) { - return _runArisFallback( - role: role, - model: model, - prompt: prompt, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - } - - try { - final process = await _processStarter( - command, - args, - environment: envVars, - workingDirectory: cwd.isNotEmpty ? cwd : null, - ); - _activeCliProcess = process; - - await process.stdin.close(); - - // 超时控制 - final timeout = Duration(seconds: _config.timeoutSeconds); - - final stdoutFuture = process.stdout - .transform(utf8.decoder) - .join() - .timeout( - timeout, - onTimeout: () { - process.kill(); - return '[超时或进程已终止]'; - }, - ); - - final stderrFuture = process.stderr - .transform(utf8.decoder) - .join() - .timeout(timeout, onTimeout: () => ''); - - final results = await Future.wait([stdoutFuture, stderrFuture]); - final exitCode = await process.exitCode.timeout( - timeout, - onTimeout: () => -1, - ); - _activeCliProcess = null; - - final cliResult = CliResult( - output: results[0], - error: results[1], - exitCode: exitCode, - ); - if (_config.usesAris && !cliResult.success) { - return _runArisFallback( - role: role, - model: model, - prompt: prompt, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - } - return cliResult; - } catch (e) { - _activeCliProcess = null; - if (_config.usesAris) { - return _runArisFallback( - role: role, - model: model, - prompt: prompt, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - } - return CliResult(output: '', error: e.toString(), exitCode: -1); - } - } - - /// 构建 Architect 的 Prompt - String _buildArchitectPrompt( - String task, - List selectedSkills, - String instructionBlock, - ) { - return ''' -$instructionBlock - -你是一个多 Agent 协作调度者。请先收敛 requirements -> acceptance evidence，再输出可执行的主程/worker分工。 - -## 用户需求 -$task - -## 优先技能 -${selectedSkills.isEmpty ? '- 无' : selectedSkills.map((item) => '- $item').join('\n')} - -请输出： -1. 任务概述（2-3 句话） -2. 子任务列表（3-5 个），每个子任务包含： - - 任务编号和描述 - - 负责角色（文档/主程/worker） - - 接受标准 - - 关键技术点 -3. 推荐的执行顺序与关键里程碑 - -请严格按以下格式输出： -## 概述 -[你的概述] - -## 子任务 -1. [任务描述] | 角色：[文档/主程/worker] | 接受标准：[可验证结果] | 关键技术：[技术点] -2. [任务描述] | 角色：[文档/主程/worker] | 接受标准：[可验证结果] | 关键技术：[技术点] -... -'''; - } - - Future _resolveToolForRole( - MultiAgentRole role, - String configuredTool, - ) async { - if (!_config.usesAris) { - return configuredTool; - } - final configuredModel = _resolvedModelForRole( - role, - configuredModel: _modelForRole(role).trim(), - ); - final candidates = switch (role) { - MultiAgentRole.architect => [ - configuredTool, - 'claude', - 'codex', - 'opencode', - 'gemini', - ], - MultiAgentRole.engineer => [ - configuredTool, - 'codex', - 'opencode', - 'claude', - 'gemini', - ], - MultiAgentRole.testerDoc => [ - configuredTool, - 'opencode', - 'codex', - 'claude', - 'gemini', - ], - }; - for (final candidate in candidates) { - final trimmed = candidate.trim(); - if (trimmed.isEmpty) { - continue; - } - if (_prefersOllamaLaunch(tool: trimmed, model: configuredModel)) { - if (await _binaryExists('ollama')) { - return trimmed; - } - } else if (await _binaryExists(_resolveCliPath(trimmed))) { - return trimmed; - } - } - return configuredTool; - } - - String _resolvedModelForRole( - MultiAgentRole role, { - required String configuredModel, - }) { - final trimmed = configuredModel.trim(); - if (trimmed.isNotEmpty) { - return trimmed; - } - switch (role) { - case MultiAgentRole.architect: - return 'kimi-k2.5:cloud'; - case MultiAgentRole.engineer: - return 'minimax-m2.7:cloud'; - case MultiAgentRole.testerDoc: - return 'glm-5:cloud'; - } - } - - Future _binaryExists(String command) async { - final resolver = _binaryExistsResolver; - if (resolver != null) { - return resolver(command); - } - final check = await Process.run( - Platform.isWindows ? 'where' : 'which', - [command], - runInShell: true, - ); - return check.exitCode == 0 && '${check.stdout}'.trim().isNotEmpty; - } - - Future _runArisFallback({ - required MultiAgentRole role, - required String model, - required String prompt, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - if (role == MultiAgentRole.testerDoc) { - final viaLlmChat = await _runArisTesterViaLlmChat( - model: model, - prompt: prompt, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - if (viaLlmChat.success) { - return viaLlmChat; - } - } - return _runOpenAiCompatiblePrompt( - role: role, - model: model, - prompt: prompt, - aiGatewayBaseUrl: aiGatewayBaseUrl, - aiGatewayApiKey: aiGatewayApiKey, - ); - } - - Future _runArisTesterViaLlmChat({ - required String model, - required String prompt, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - try { - if (!await _goCoreLocator.isAvailable()) { - return const CliResult( - output: '', - error: 'Go core is unavailable for llm-chat', - exitCode: -1, - ); - } - final endpoint = _openAiCompatibleBaseUrl( - aiGatewayBaseUrl: aiGatewayBaseUrl, - ); - final apiKey = _openAiCompatibleApiKey(aiGatewayApiKey: aiGatewayApiKey); - final output = await _arisLlmChatClient.chat( - endpoint: endpoint, - apiKey: apiKey, - model: model, - prompt: prompt, - systemPrompt: - 'You are the ARIS reviewer. Review the provided implementation and return actionable feedback.', - ); - return CliResult(output: output, error: '', exitCode: 0); - } catch (error) { - return CliResult(output: '', error: error.toString(), exitCode: -1); - } - } - - Future _runArisTesterViaClaudeReview({ - required String model, - required String prompt, - }) async { - try { - if (!await _goCoreLocator.isAvailable()) { - return const CliResult( - output: '', - error: 'Go core is unavailable for claude-review', - exitCode: -1, - ); - } - if (!await _binaryExists(_resolveCliPath('claude'))) { - return const CliResult( - output: '', - error: 'Claude CLI is unavailable for claude-review', - exitCode: -1, - ); - } - final output = await _arisLlmChatClient.claudeReview( - prompt: prompt, - model: model, - systemPrompt: - 'You are the ARIS reviewer. Review the provided implementation and return actionable feedback.', - ); - return CliResult(output: output, error: '', exitCode: 0); - } catch (error) { - return CliResult(output: '', error: error.toString(), exitCode: -1); - } - } - - Future _runOpenAiCompatiblePrompt({ - required MultiAgentRole role, - required String model, - required String prompt, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) async { - final client = _httpClientFactory(); - _activeHttpClient = client; - try { - final request = await client.postUrl( - Uri.parse( - '${_openAiCompatibleBaseUrl(aiGatewayBaseUrl: aiGatewayBaseUrl).replaceAll(RegExp(r'/$'), '')}/chat/completions', - ), - ); - request.headers.set(HttpHeaders.contentTypeHeader, 'application/json'); - request.headers.set( - HttpHeaders.authorizationHeader, - 'Bearer ${_openAiCompatibleApiKey(aiGatewayApiKey: aiGatewayApiKey)}', - ); - request.add( - utf8.encode( - jsonEncode({ - 'model': model, - 'stream': false, - 'messages': >[ - { - 'role': 'system', - 'content': _systemPromptForRole(role), - }, - {'role': 'user', 'content': prompt}, - ], - }), - ), - ); - final response = await request.close().timeout( - Duration(seconds: _config.timeoutSeconds), - ); - final body = await utf8.decodeStream(response); - if (response.statusCode < 200 || response.statusCode >= 300) { - return CliResult( - output: '', - error: body, - exitCode: response.statusCode, - ); - } - final decoded = jsonDecode(body) as Map; - final choices = decoded['choices'] as List? ?? const []; - final firstChoice = choices.isNotEmpty ? choices.first : null; - final output = - ((firstChoice as Map?)?['message'] as Map?)?['content']?.toString() ?? - ''; - return CliResult(output: output, error: '', exitCode: 0); - } catch (error) { - return CliResult(output: '', error: error.toString(), exitCode: -1); - } finally { - _activeHttpClient = null; - try { - client.close(force: true); - } catch (_) { - // Best effort only. - } - } - } - - String _openAiCompatibleBaseUrl({required String aiGatewayBaseUrl}) { - if (_config.aiGatewayInjectionPolicy != AiGatewayInjectionPolicy.disabled && - aiGatewayBaseUrl.trim().isNotEmpty) { - final normalized = aiGatewayBaseUrl.trim(); - return normalized.endsWith('/v1') ? normalized : '$normalized/v1'; - } - final normalized = _config.ollamaEndpoint.trim(); - return normalized.endsWith('/v1') ? normalized : '$normalized/v1'; - } - - String _openAiCompatibleApiKey({required String aiGatewayApiKey}) { - if (_config.aiGatewayInjectionPolicy != AiGatewayInjectionPolicy.disabled && - aiGatewayApiKey.trim().isNotEmpty) { - return aiGatewayApiKey.trim(); - } - return 'ollama'; - } - - String _systemPromptForRole(MultiAgentRole role) { - return switch (role) { - MultiAgentRole.architect => - 'You are the architecture and documentation lane in a multi-agent coding workflow. Focus on requirements, acceptance evidence, task slicing, and milestones.', - MultiAgentRole.engineer => - 'You are the lead engineer in a multi-agent coding workflow. Produce implementation-oriented output for the critical path.', - MultiAgentRole.testerDoc => - 'You are the worker-review lane in a multi-agent coding workflow. Review, score, and suggest follow-up fixes and worker follow-ups.', - }; - } - - String _roleLabel(MultiAgentRole role) { - return switch (role) { - MultiAgentRole.architect => 'Architect', - MultiAgentRole.engineer => 'Lead Engineer', - MultiAgentRole.testerDoc => 'Worker/Review', - }; - } - - String _modelForRole(MultiAgentRole role) { - return switch (role) { - MultiAgentRole.architect => _config.architect.model, - MultiAgentRole.engineer => _config.engineer.model, - MultiAgentRole.testerDoc => _config.tester.model, - }; - } - - bool _prefersOllamaLaunch({required String tool, required String model}) { - final normalizedTool = tool.trim().toLowerCase(); - final normalizedModel = model.trim(); - if (normalizedModel.isEmpty) { - return false; - } - if (normalizedTool != 'claude' && - normalizedTool != 'codex' && - normalizedTool != 'opencode') { - return false; - } - return true; - } - - List _buildOllamaLaunchArgs({ - required String tool, - required String model, - required String prompt, - required String cwd, - }) { - final args = ['launch', tool, '--model', model]; - if (tool == 'claude') { - args.add('--yes'); - args.addAll(['--', '-p', prompt]); - return args; - } - if (tool == 'codex') { - args.addAll([ - '--', - 'exec', - '--skip-git-repo-check', - '--color', - 'never', - if (cwd.isNotEmpty) ...['-C', cwd], - prompt, - ]); - return args; - } - if (tool == 'opencode') { - args.addAll([ - '--', - 'run', - '--format', - 'default', - if (cwd.isNotEmpty) ...['--dir', cwd], - prompt, - ]); - return args; - } - args.addAll(['--', '-p', prompt]); - return args; - } - - void _throwIfAborted() { - if (_abortRequested) { - throw StateError('Multi-agent collaboration aborted.'); - } - } - - /// 解析 Architect 分解的任务 - List _parseDecomposedTasks(String architectOutput) { - final tasks = []; - final lines = architectOutput.split('\n'); - - var order = 1; - for (final line in lines) { - final trimmed = line.trim(); - if (trimmed.isEmpty) continue; - - // 匹配 "- 描述" 或 "1. 描述" 格式 - final dashMatch = RegExp(r'^[-*]\s+(.+)').firstMatch(trimmed); - final numMatch = RegExp(r'^\d+[.、)]\s*(.+)').firstMatch(trimmed); - - String? description; - if (dashMatch != null) { - description = dashMatch.group(1); - } else if (numMatch != null) { - description = numMatch.group(1); - } - - if (description != null && description.isNotEmpty) { - // 去除复杂度等技术注释 - description = description.replaceAll(RegExp(r'\s*\|.*'), '').trim(); - - // 判断任务类型 - SubTaskType type = SubTaskType.implementation; - final lower = description.toLowerCase(); - if (lower.contains('测试') || lower.contains('test')) { - type = SubTaskType.testing; - } else if (lower.contains('文档') || lower.contains('doc')) { - type = SubTaskType.documentation; - } else if (lower.contains('设计') || lower.contains('design')) { - type = SubTaskType.design; - } else if (lower.contains('部署') || lower.contains('deploy')) { - type = SubTaskType.deployment; - } - - tasks.add( - SubTask( - id: order.toString(), - description: description, - order: order, - type: type, - ), - ); - order++; - } - } - - // 如果解析失败，至少返回一个包含完整需求的子任务 - if (tasks.isEmpty) { - tasks.add( - SubTask( - id: '1', - description: architectOutput.length > 200 - ? '${architectOutput.substring(0, 200)}...' - : architectOutput, - order: 1, - type: SubTaskType.implementation, - ), - ); - } - - return tasks; - } - - /// 解析审阅评分 - int _parseReviewScore(String output) { - // 尝试匹配 "评分 (1-10)" 模式 - final patterns = [ - RegExp(r'评分\s*\(?[1１00]\)?\s*[:：]?\s*(\d+)'), - RegExp(r'score\s*[:：]?\s*(\d+)', caseSensitive: false), - RegExp(r'评分[：:\s]*(\d+)'), - RegExp(r'\*\*(\d+)\s*/\s*10\*\*'), - RegExp(r'(\d+)\s*/\s*10'), - ]; - - for (final pattern in patterns) { - final match = pattern.firstMatch(output); - if (match != null) { - final scoreStr = match.group(1)!; - final score = int.tryParse( - scoreStr.replaceAll('１', '1').replaceAll('０', '0'), - ); - if (score != null && score >= 1 && score <= 10) { - return score; - } - } - } - - // 默认中等评分 - return 5; - } - - /// 提取审阅反馈 - String _extractFeedback(String output) { - final feedbackIndex = output.indexOf(RegExp(r'##?\s*问题|##?\s*改进|##?\s*建议')); - if (feedbackIndex >= 0) { - final endIndex = output.indexOf( - RegExp(r'##?\s*测试|##?\s*文档'), - feedbackIndex + 1, - ); - if (endIndex > feedbackIndex) { - return output.substring(feedbackIndex, endIndex).trim(); - } - return output.substring(feedbackIndex).trim(); - } - return output; - } - - /// 构建 Ollama 环境变量 - Map _buildCliEnvVars({ - required String tool, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) { - 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 (tool == '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 (tool == 'claude' || tool == 'codex') { - baseEnv['ANTHROPIC_AUTH_TOKEN'] = 'ollama'; - baseEnv['ANTHROPIC_API_KEY'] = ''; - baseEnv['ANTHROPIC_BASE_URL'] = ollamaEndpoint; - } - return baseEnv; - } - - /// 解析 CLI 工具路径 - String _resolveCliPath(String tool) { - switch (tool) { - case 'claude': - return 'claude'; - case 'codex': - return 'codex'; - case 'gemini': - return 'gemini'; - case 'opencode': - return 'opencode'; - default: - return tool; - } - } - - void _emitEvent( - void Function(MultiAgentRunEvent event)? onEvent, - MultiAgentRunEvent event, - ) { - onEvent?.call(event); - } - /// 记录日志 void _log(CollaborationLogLevel level, String emoji, String message) { _logEntries.add( @@ -1456,184 +439,3 @@ ${selectedSkills.isEmpty ? '- 无' : selectedSkills.map((item) => '- $item').joi notifyListeners(); } } - -// ============================================================ -// 数据模型 -// ============================================================ - -/// 协作日志条目 -class CollaborationLogEntry { - const CollaborationLogEntry({ - required this.timestamp, - required this.level, - required this.emoji, - required this.message, - }); - - final DateTime timestamp; - final CollaborationLogLevel level; - final String emoji; - final String message; - - String get formattedTime { - final h = timestamp.hour.toString().padLeft(2, '0'); - final m = timestamp.minute.toString().padLeft(2, '0'); - final s = timestamp.second.toString().padLeft(2, '0'); - return '$h:$m:$s'; - } -} - -enum CollaborationLogLevel { debug, info, warning, error, success } - -/// CLI 执行结果 -class CliResult { - const CliResult({ - required this.output, - required this.error, - required this.exitCode, - }); - - final String output; - final String error; - final int exitCode; - - bool get success => exitCode == 0 && error.isEmpty; -} - -/// Architect 执行结果 -class ArchitectResult { - ArchitectResult({ - required this.output, - required this.decomposedTasks, - required this.duration, - }); - - final String output; - final List decomposedTasks; - final Duration duration; -} - -/// Engineer 执行结果 -class EngineerResult { - EngineerResult({ - required this.output, - required this.codeOutput, - required this.completedTasks, - required this.duration, - }); - - final String output; - String codeOutput; - final List completedTasks; - final Duration duration; -} - -/// Tester 执行结果 -class TesterResult { - TesterResult({ - required this.output, - required this.score, - required this.feedback, - required this.duration, - }); - - final String output; - final int score; - final String feedback; - final Duration duration; -} - -/// 协作步骤 -class CollaborationStep { - const CollaborationStep({ - required this.role, - required this.status, - required this.output, - required this.duration, - this.iteration, - this.score, - }); - - final String role; - final StepStatus status; - final String output; - final Duration duration; - final int? iteration; - final int? score; - - Map toJson() { - return { - 'role': role, - 'status': status.name, - 'output': output, - 'durationMs': duration.inMilliseconds, - if (iteration != null) 'iteration': iteration, - if (score != null) 'score': score, - }; - } -} - -enum StepStatus { pending, running, completed, failed } - -/// 子任务 -class SubTask { - const SubTask({ - required this.id, - required this.description, - required this.order, - required this.type, - }); - - final String id; - final String description; - final int order; - final SubTaskType type; -} - -enum SubTaskType { design, implementation, testing, documentation, deployment } - -/// 附件 -class CollaborationAttachment { - const CollaborationAttachment({ - required this.name, - required this.description, - required this.path, - }); - - final String name; - final String description; - final String path; -} - -/// 协作最终结果 -class CollaborationResult { - const CollaborationResult({ - required this.success, - required this.steps, - required this.finalCode, - required this.finalScore, - required this.duration, - required this.iterations, - this.error, - }); - - final bool success; - final List steps; - final String finalCode; - final int finalScore; - final Duration duration; - final int iterations; - final String? error; - - Map toJson() { - return { - 'success': success, - 'steps': steps.map((item) => item.toJson()).toList(growable: false), - 'finalCode': finalCode, - 'finalScore': finalScore, - 'durationMs': duration.inMilliseconds, - 'iterations': iterations, - if (error != null) 'error': error, - }; - } -} diff --git a/lib/runtime/multi_agent_orchestrator_protocol.part.dart b/lib/runtime/multi_agent_orchestrator_protocol.part.dart new file mode 100644 index 00000000..481faf19 --- /dev/null +++ b/lib/runtime/multi_agent_orchestrator_protocol.part.dart @@ -0,0 +1,186 @@ +part of 'multi_agent_orchestrator.dart'; + +typedef CliProcessStarter = + Future Function( + String executable, + List arguments, { + Map? environment, + String? workingDirectory, + }); + +/// 协作日志条目 +class CollaborationLogEntry { + const CollaborationLogEntry({ + required this.timestamp, + required this.level, + required this.emoji, + required this.message, + }); + + final DateTime timestamp; + final CollaborationLogLevel level; + final String emoji; + final String message; + + String get formattedTime { + final h = timestamp.hour.toString().padLeft(2, '0'); + final m = timestamp.minute.toString().padLeft(2, '0'); + final s = timestamp.second.toString().padLeft(2, '0'); + return '$h:$m:$s'; + } +} + +enum CollaborationLogLevel { debug, info, warning, error, success } + +/// CLI 执行结果 +class CliResult { + const CliResult({ + required this.output, + required this.error, + required this.exitCode, + }); + + final String output; + final String error; + final int exitCode; + + bool get success => exitCode == 0 && error.isEmpty; +} + +/// Architect 执行结果 +class ArchitectResult { + ArchitectResult({ + required this.output, + required this.decomposedTasks, + required this.duration, + }); + + final String output; + final List decomposedTasks; + final Duration duration; +} + +/// Engineer 执行结果 +class EngineerResult { + EngineerResult({ + required this.output, + required this.codeOutput, + required this.completedTasks, + required this.duration, + }); + + final String output; + String codeOutput; + final List completedTasks; + final Duration duration; +} + +/// Tester 执行结果 +class TesterResult { + TesterResult({ + required this.output, + required this.score, + required this.feedback, + required this.duration, + }); + + final String output; + final int score; + final String feedback; + final Duration duration; +} + +/// 协作步骤 +class CollaborationStep { + const CollaborationStep({ + required this.role, + required this.status, + required this.output, + required this.duration, + this.iteration, + this.score, + }); + + final String role; + final StepStatus status; + final String output; + final Duration duration; + final int? iteration; + final int? score; + + Map toJson() { + return { + 'role': role, + 'status': status.name, + 'output': output, + 'durationMs': duration.inMilliseconds, + if (iteration != null) 'iteration': iteration, + if (score != null) 'score': score, + }; + } +} + +enum StepStatus { pending, running, completed, failed } + +/// 子任务 +class SubTask { + const SubTask({ + required this.id, + required this.description, + required this.order, + required this.type, + }); + + final String id; + final String description; + final int order; + final SubTaskType type; +} + +enum SubTaskType { design, implementation, testing, documentation, deployment } + +/// 附件 +class CollaborationAttachment { + const CollaborationAttachment({ + required this.name, + required this.description, + required this.path, + }); + + final String name; + final String description; + final String path; +} + +/// 协作最终结果 +class CollaborationResult { + const CollaborationResult({ + required this.success, + required this.steps, + required this.finalCode, + required this.finalScore, + required this.duration, + required this.iterations, + this.error, + }); + + final bool success; + final List steps; + final String finalCode; + final int finalScore; + final Duration duration; + final int iterations; + final String? error; + + Map toJson() { + return { + 'success': success, + 'steps': steps.map((item) => item.toJson()).toList(growable: false), + 'finalCode': finalCode, + 'finalScore': finalScore, + 'durationMs': duration.inMilliseconds, + 'iterations': iterations, + if (error != null) 'error': error, + }; + } +} diff --git a/lib/runtime/multi_agent_orchestrator_support.part.dart b/lib/runtime/multi_agent_orchestrator_support.part.dart new file mode 100644 index 00000000..626bf68f --- /dev/null +++ b/lib/runtime/multi_agent_orchestrator_support.part.dart @@ -0,0 +1,280 @@ +part of 'multi_agent_orchestrator.dart'; + +extension _MultiAgentOrchestratorSupport on MultiAgentOrchestrator { + String _openAiCompatibleBaseUrl({required String aiGatewayBaseUrl}) { + if (_config.aiGatewayInjectionPolicy != AiGatewayInjectionPolicy.disabled && + aiGatewayBaseUrl.trim().isNotEmpty) { + final normalized = aiGatewayBaseUrl.trim(); + return normalized.endsWith('/v1') ? normalized : '$normalized/v1'; + } + final normalized = _config.ollamaEndpoint.trim(); + return normalized.endsWith('/v1') ? normalized : '$normalized/v1'; + } + + String _openAiCompatibleApiKey({required String aiGatewayApiKey}) { + if (_config.aiGatewayInjectionPolicy != AiGatewayInjectionPolicy.disabled && + aiGatewayApiKey.trim().isNotEmpty) { + return aiGatewayApiKey.trim(); + } + return 'ollama'; + } + + String _systemPromptForRole(MultiAgentRole role) { + return switch (role) { + MultiAgentRole.architect => + 'You are the architecture and documentation lane in a multi-agent coding workflow. Focus on requirements, acceptance evidence, task slicing, and milestones.', + MultiAgentRole.engineer => + 'You are the lead engineer in a multi-agent coding workflow. Produce implementation-oriented output for the critical path.', + MultiAgentRole.testerDoc => + 'You are the worker-review lane in a multi-agent coding workflow. Review, score, and suggest follow-up fixes and worker follow-ups.', + }; + } + + String _roleLabel(MultiAgentRole role) { + return switch (role) { + MultiAgentRole.architect => 'Architect', + MultiAgentRole.engineer => 'Lead Engineer', + MultiAgentRole.testerDoc => 'Worker/Review', + }; + } + + String _modelForRole(MultiAgentRole role) { + return switch (role) { + MultiAgentRole.architect => _config.architect.model, + MultiAgentRole.engineer => _config.engineer.model, + MultiAgentRole.testerDoc => _config.tester.model, + }; + } + + bool _prefersOllamaLaunch({required String tool, required String model}) { + final normalizedTool = tool.trim().toLowerCase(); + final normalizedModel = model.trim(); + if (normalizedModel.isEmpty) { + return false; + } + if (normalizedTool != 'claude' && + normalizedTool != 'codex' && + normalizedTool != 'opencode') { + return false; + } + return true; + } + + List _buildOllamaLaunchArgs({ + required String tool, + required String model, + required String prompt, + required String cwd, + }) { + final args = ['launch', tool, '--model', model]; + if (tool == 'claude') { + args.add('--yes'); + args.addAll(['--', '-p', prompt]); + return args; + } + if (tool == 'codex') { + args.addAll([ + '--', + 'exec', + '--skip-git-repo-check', + '--color', + 'never', + if (cwd.isNotEmpty) ...['-C', cwd], + prompt, + ]); + return args; + } + if (tool == 'opencode') { + args.addAll([ + '--', + 'run', + '--format', + 'default', + if (cwd.isNotEmpty) ...['--dir', cwd], + prompt, + ]); + return args; + } + args.addAll(['--', '-p', prompt]); + return args; + } + + void _throwIfAborted() { + if (_abortRequested) { + throw StateError('Multi-agent collaboration aborted.'); + } + } + + /// 解析 Architect 分解的任务 + List _parseDecomposedTasks(String architectOutput) { + final tasks = []; + final lines = architectOutput.split('\n'); + + var order = 1; + for (final line in lines) { + final trimmed = line.trim(); + if (trimmed.isEmpty) continue; + + // 匹配 "- 描述" 或 "1. 描述" 格式 + final dashMatch = RegExp(r'^[-*]\s+(.+)').firstMatch(trimmed); + final numMatch = RegExp(r'^\d+[.、)]\s*(.+)').firstMatch(trimmed); + + String? description; + if (dashMatch != null) { + description = dashMatch.group(1); + } else if (numMatch != null) { + description = numMatch.group(1); + } + + if (description != null && description.isNotEmpty) { + // 去除复杂度等技术注释 + description = description.replaceAll(RegExp(r'\s*\|.*'), '').trim(); + + // 判断任务类型 + SubTaskType type = SubTaskType.implementation; + final lower = description.toLowerCase(); + if (lower.contains('测试') || lower.contains('test')) { + type = SubTaskType.testing; + } else if (lower.contains('文档') || lower.contains('doc')) { + type = SubTaskType.documentation; + } else if (lower.contains('设计') || lower.contains('design')) { + type = SubTaskType.design; + } else if (lower.contains('部署') || lower.contains('deploy')) { + type = SubTaskType.deployment; + } + + tasks.add( + SubTask( + id: order.toString(), + description: description, + order: order, + type: type, + ), + ); + order++; + } + } + + // 如果解析失败，至少返回一个包含完整需求的子任务 + if (tasks.isEmpty) { + tasks.add( + SubTask( + id: '1', + description: architectOutput.length > 200 + ? '${architectOutput.substring(0, 200)}...' + : architectOutput, + order: 1, + type: SubTaskType.implementation, + ), + ); + } + + return tasks; + } + + /// 解析审阅评分 + int _parseReviewScore(String output) { + // 尝试匹配 "评分 (1-10)" 模式 + final patterns = [ + RegExp(r'评分\s*\(?[1１00]\)?\s*[:：]?\s*(\d+)'), + RegExp(r'score\s*[:：]?\s*(\d+)', caseSensitive: false), + RegExp(r'评分[：:\s]*(\d+)'), + RegExp(r'\*\*(\d+)\s*/\s*10\*\*'), + RegExp(r'(\d+)\s*/\s*10'), + ]; + + for (final pattern in patterns) { + final match = pattern.firstMatch(output); + if (match != null) { + final scoreStr = match.group(1)!; + final score = int.tryParse( + scoreStr.replaceAll('１', '1').replaceAll('０', '0'), + ); + if (score != null && score >= 1 && score <= 10) { + return score; + } + } + } + + // 默认中等评分 + return 5; + } + + /// 提取审阅反馈 + String _extractFeedback(String output) { + final feedbackIndex = output.indexOf(RegExp(r'##?\s*问题|##?\s*改进|##?\s*建议')); + if (feedbackIndex >= 0) { + final endIndex = output.indexOf( + RegExp(r'##?\s*测试|##?\s*文档'), + feedbackIndex + 1, + ); + if (endIndex > feedbackIndex) { + return output.substring(feedbackIndex, endIndex).trim(); + } + return output.substring(feedbackIndex).trim(); + } + return output; + } + + /// 构建 Ollama 环境变量 + Map _buildCliEnvVars({ + required String tool, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) { + 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 (tool == '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 (tool == 'claude' || tool == 'codex') { + baseEnv['ANTHROPIC_AUTH_TOKEN'] = 'ollama'; + baseEnv['ANTHROPIC_API_KEY'] = ''; + baseEnv['ANTHROPIC_BASE_URL'] = ollamaEndpoint; + } + return baseEnv; + } + + /// 解析 CLI 工具路径 + String _resolveCliPath(String tool) { + switch (tool) { + case 'claude': + return 'claude'; + case 'codex': + return 'codex'; + case 'gemini': + return 'gemini'; + case 'opencode': + return 'opencode'; + default: + return tool; + } + } + + void _emitEvent( + void Function(MultiAgentRunEvent event)? onEvent, + MultiAgentRunEvent event, + ) { + onEvent?.call(event); + } + +} diff --git a/lib/runtime/multi_agent_orchestrator_workflow.part.dart b/lib/runtime/multi_agent_orchestrator_workflow.part.dart new file mode 100644 index 00000000..82be7b43 --- /dev/null +++ b/lib/runtime/multi_agent_orchestrator_workflow.part.dart @@ -0,0 +1,738 @@ +part of 'multi_agent_orchestrator.dart'; + +extension _MultiAgentOrchestratorWorkflow on MultiAgentOrchestrator { + /// 运行 Architect（调度/文档分析） + Future _runArchitect( + String task, { + required FrameworkPreset preset, + required List selectedSkills, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + final stopwatch = Stopwatch()..start(); + + try { + // 根据配置选择 Architect 工具 + if (_config.architectEnabled) { + final tool = await _resolveToolForRole( + MultiAgentRole.architect, + _config.architectTool, + ); + final instructionBlock = await preset.roleInstructionBlock( + role: MultiAgentRole.architect, + tool: tool, + selectedSkills: selectedSkills, + ); + final result = await _runCliPrompt( + role: MultiAgentRole.architect, + tool: tool, + model: _resolvedModelForRole( + MultiAgentRole.architect, + configuredModel: _config.architectModel, + ), + prompt: _buildArchitectPrompt(task, selectedSkills, instructionBlock), + cwd: '', + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + stopwatch.stop(); + + // 解析分解后的任务 + final tasks = _parseDecomposedTasks(result.output); + return ArchitectResult( + output: result.output, + decomposedTasks: tasks, + duration: stopwatch.elapsed, + ); + } else { + // Architect 被禁用，直接返回原任务作为单一子任务 + stopwatch.stop(); + return ArchitectResult( + output: task, + decomposedTasks: [ + SubTask( + id: '1', + description: task, + order: 1, + type: SubTaskType.implementation, + ), + ], + duration: stopwatch.elapsed, + ); + } + } catch (e) { + stopwatch.stop(); + rethrow; + } + } + + /// 运行 Lead Engineer（主实现） + Future _runEngineer( + List tasks, + String workingDirectory, + List attachments, { + required FrameworkPreset preset, + required List selectedSkills, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + final stopwatch = Stopwatch()..start(); + final tool = await _resolveToolForRole( + MultiAgentRole.engineer, + _config.engineerTool, + ); + final instructionBlock = await preset.roleInstructionBlock( + role: MultiAgentRole.engineer, + tool: tool, + selectedSkills: selectedSkills, + ); + + final taskList = tasks + .map((t) => '## ${t.order}. ${t.description}') + .join('\n\n'); + + final prompt = + ''' +$instructionBlock + +你是一个资深工程师，负责完成以下编码任务： + +### 任务列表 +$taskList + +### 工作目录 +$workingDirectory + +### 附件信息 +${attachments.map((a) => '- ${a.name}: ${a.description}').join('\n')} + +### 优先技能 +${selectedSkills.isEmpty ? '- 无' : selectedSkills.map((item) => '- $item').join('\n')} + +请完成这些任务，输出完整的代码实现。 +'''; + + final result = await _runCliPrompt( + role: MultiAgentRole.engineer, + tool: tool, + model: _resolvedModelForRole( + MultiAgentRole.engineer, + configuredModel: _config.engineerModel, + ), + prompt: prompt, + cwd: workingDirectory, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + stopwatch.stop(); + + return EngineerResult( + output: result.output, + codeOutput: result.output, + completedTasks: tasks, + duration: stopwatch.elapsed, + ); + } + + /// 运行 Worker/Review（代码审阅） + Future _runTester( + String codeOutput, { + required FrameworkPreset preset, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + final stopwatch = Stopwatch()..start(); + final tool = await _resolveToolForRole( + MultiAgentRole.testerDoc, + _config.testerTool, + ); + final instructionBlock = await preset.roleInstructionBlock( + role: MultiAgentRole.testerDoc, + tool: tool, + selectedSkills: const [], + ); + + final prompt = + ''' +$instructionBlock + +请审阅以下代码，并按以下格式输出： + +## 评分 (1-10) +[1-10 的分数，10 最高] + +## 问题列表 +[发现的问题，格式：- 问题描述 (严重程度: 高/中/低)] + +## 改进建议 +[具体的改进建议] + +## 测试用例 +```[语言] +[生成的测试用例代码] +``` + +## 文档建议 +[如有需要补充的文档说明] + +### 待审阅代码 +${codeOutput.length > 4000 ? '${codeOutput.substring(0, 4000)}\n...[代码已截断]' : codeOutput} +'''; + + final testerModel = _resolvedModelForRole( + MultiAgentRole.testerDoc, + configuredModel: _config.testerModel, + ); + final result = _config.usesAris && tool == 'claude' + ? await _runArisTesterViaClaudeReview( + model: testerModel, + prompt: prompt, + ) + : await _runCliPrompt( + role: MultiAgentRole.testerDoc, + tool: tool, + model: testerModel, + prompt: prompt, + cwd: '', + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + stopwatch.stop(); + + final score = _parseReviewScore(result.output); + final feedback = _extractFeedback(result.output); + + return TesterResult( + output: result.output, + score: score, + feedback: feedback, + duration: stopwatch.elapsed, + ); + } + + /// 运行修复（迭代循环中） + Future _runFix( + String originalCode, + String feedback, + String workingDirectory, { + required FrameworkPreset preset, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + final stopwatch = Stopwatch()..start(); + final tool = await _resolveToolForRole( + MultiAgentRole.engineer, + _config.engineerTool, + ); + final instructionBlock = await preset.roleInstructionBlock( + role: MultiAgentRole.engineer, + tool: tool, + selectedSkills: const [], + ); + + final prompt = + ''' +$instructionBlock + +你是一个资深工程师。请根据审阅反馈修复代码。 + +## 审阅反馈 +$feedback + +## 原始代码 +$originalCode + +请完成修复，输出修复后的完整代码。 +'''; + + final result = await _runCliPrompt( + role: MultiAgentRole.engineer, + tool: tool, + model: _resolvedModelForRole( + MultiAgentRole.engineer, + configuredModel: _config.engineerModel, + ), + prompt: prompt, + cwd: workingDirectory, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + stopwatch.stop(); + + return EngineerResult( + output: result.output, + codeOutput: result.output, + completedTasks: [], + duration: stopwatch.elapsed, + ); + } + + /// 通用的 CLI 进程执行方法 + Future _runCliPrompt({ + required MultiAgentRole role, + required String tool, + required String model, + required String prompt, + required String cwd, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + late final List args; + late final String command; + late final Map envVars; + final useOllamaLaunch = _prefersOllamaLaunch(tool: tool, model: model); + + switch (tool) { + case 'claude': + command = useOllamaLaunch ? 'ollama' : _resolveCliPath('claude'); + envVars = _buildCliEnvVars( + tool: tool, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + if (useOllamaLaunch) { + args = _buildOllamaLaunchArgs( + tool: tool, + model: model, + prompt: prompt, + cwd: cwd, + ); + } else if (model.isNotEmpty) { + args = ['--model', model, '-p', prompt]; + } else { + args = ['-p', prompt]; + } + break; + + case 'codex': + command = useOllamaLaunch ? 'ollama' : _resolveCliPath('codex'); + envVars = _buildCliEnvVars( + tool: tool, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + if (useOllamaLaunch) { + args = _buildOllamaLaunchArgs( + tool: tool, + model: model, + prompt: prompt, + cwd: cwd, + ); + } else if (model.isNotEmpty) { + args = [ + 'exec', + '--skip-git-repo-check', + '--color', + 'never', + if (cwd.isNotEmpty) ...['-C', cwd], + '-m', + model, + prompt, + ]; + } else { + args = [ + 'exec', + '--skip-git-repo-check', + '--color', + 'never', + if (cwd.isNotEmpty) ...['-C', cwd], + prompt, + ]; + } + break; + + case 'gemini': + command = _resolveCliPath('gemini'); + envVars = _buildCliEnvVars( + tool: tool, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + if (model.isNotEmpty) { + args = ['--model', model, '-p', prompt]; + } else { + args = ['-p', prompt]; + } + break; + + case 'opencode': + command = useOllamaLaunch ? 'ollama' : _resolveCliPath('opencode'); + envVars = _buildCliEnvVars( + tool: tool, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + args = useOllamaLaunch + ? _buildOllamaLaunchArgs( + tool: tool, + model: model, + prompt: prompt, + cwd: cwd, + ) + : [ + 'run', + '--format', + 'default', + if (cwd.isNotEmpty) ...['--dir', cwd], + if (model.isNotEmpty) ...['-m', model], + prompt, + ]; + break; + + default: + throw ArgumentError('Unknown tool: $tool'); + } + + final cliAvailable = await _binaryExists(command); + if (_config.usesAris && !cliAvailable) { + return _runArisFallback( + role: role, + model: model, + prompt: prompt, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + } + + try { + final process = await _processStarter( + command, + args, + environment: envVars, + workingDirectory: cwd.isNotEmpty ? cwd : null, + ); + _activeCliProcess = process; + + await process.stdin.close(); + + // 超时控制 + final timeout = Duration(seconds: _config.timeoutSeconds); + + final stdoutFuture = process.stdout + .transform(utf8.decoder) + .join() + .timeout( + timeout, + onTimeout: () { + process.kill(); + return '[超时或进程已终止]'; + }, + ); + + final stderrFuture = process.stderr + .transform(utf8.decoder) + .join() + .timeout(timeout, onTimeout: () => ''); + + final results = await Future.wait([stdoutFuture, stderrFuture]); + final exitCode = await process.exitCode.timeout( + timeout, + onTimeout: () => -1, + ); + _activeCliProcess = null; + + final cliResult = CliResult( + output: results[0], + error: results[1], + exitCode: exitCode, + ); + if (_config.usesAris && !cliResult.success) { + return _runArisFallback( + role: role, + model: model, + prompt: prompt, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + } + return cliResult; + } catch (e) { + _activeCliProcess = null; + if (_config.usesAris) { + return _runArisFallback( + role: role, + model: model, + prompt: prompt, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + } + return CliResult(output: '', error: e.toString(), exitCode: -1); + } + } + + /// 构建 Architect 的 Prompt + String _buildArchitectPrompt( + String task, + List selectedSkills, + String instructionBlock, + ) { + return ''' +$instructionBlock + +你是一个多 Agent 协作调度者。请先收敛 requirements -> acceptance evidence，再输出可执行的主程/worker分工。 + +## 用户需求 +$task + +## 优先技能 +${selectedSkills.isEmpty ? '- 无' : selectedSkills.map((item) => '- $item').join('\n')} + +请输出： +1. 任务概述（2-3 句话） +2. 子任务列表（3-5 个），每个子任务包含： + - 任务编号和描述 + - 负责角色（文档/主程/worker） + - 接受标准 + - 关键技术点 +3. 推荐的执行顺序与关键里程碑 + +请严格按以下格式输出： +## 概述 +[你的概述] + +## 子任务 +1. [任务描述] | 角色：[文档/主程/worker] | 接受标准：[可验证结果] | 关键技术：[技术点] +2. [任务描述] | 角色：[文档/主程/worker] | 接受标准：[可验证结果] | 关键技术：[技术点] +... +'''; + } + + Future _resolveToolForRole( + MultiAgentRole role, + String configuredTool, + ) async { + if (!_config.usesAris) { + return configuredTool; + } + final configuredModel = _resolvedModelForRole( + role, + configuredModel: _modelForRole(role).trim(), + ); + final candidates = switch (role) { + MultiAgentRole.architect => [ + configuredTool, + 'claude', + 'codex', + 'opencode', + 'gemini', + ], + MultiAgentRole.engineer => [ + configuredTool, + 'codex', + 'opencode', + 'claude', + 'gemini', + ], + MultiAgentRole.testerDoc => [ + configuredTool, + 'opencode', + 'codex', + 'claude', + 'gemini', + ], + }; + for (final candidate in candidates) { + final trimmed = candidate.trim(); + if (trimmed.isEmpty) { + continue; + } + if (_prefersOllamaLaunch(tool: trimmed, model: configuredModel)) { + if (await _binaryExists('ollama')) { + return trimmed; + } + } else if (await _binaryExists(_resolveCliPath(trimmed))) { + return trimmed; + } + } + return configuredTool; + } + + String _resolvedModelForRole( + MultiAgentRole role, { + required String configuredModel, + }) { + final trimmed = configuredModel.trim(); + if (trimmed.isNotEmpty) { + return trimmed; + } + switch (role) { + case MultiAgentRole.architect: + return 'kimi-k2.5:cloud'; + case MultiAgentRole.engineer: + return 'minimax-m2.7:cloud'; + case MultiAgentRole.testerDoc: + return 'glm-5:cloud'; + } + } + + Future _binaryExists(String command) async { + final resolver = _binaryExistsResolver; + if (resolver != null) { + return resolver(command); + } + final check = await Process.run( + Platform.isWindows ? 'where' : 'which', + [command], + runInShell: true, + ); + return check.exitCode == 0 && '${check.stdout}'.trim().isNotEmpty; + } + + Future _runArisFallback({ + required MultiAgentRole role, + required String model, + required String prompt, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + if (role == MultiAgentRole.testerDoc) { + final viaLlmChat = await _runArisTesterViaLlmChat( + model: model, + prompt: prompt, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + if (viaLlmChat.success) { + return viaLlmChat; + } + } + return _runOpenAiCompatiblePrompt( + role: role, + model: model, + prompt: prompt, + aiGatewayBaseUrl: aiGatewayBaseUrl, + aiGatewayApiKey: aiGatewayApiKey, + ); + } + + Future _runArisTesterViaLlmChat({ + required String model, + required String prompt, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + try { + if (!await _goCoreLocator.isAvailable()) { + return const CliResult( + output: '', + error: 'Go core is unavailable for llm-chat', + exitCode: -1, + ); + } + final endpoint = _openAiCompatibleBaseUrl( + aiGatewayBaseUrl: aiGatewayBaseUrl, + ); + final apiKey = _openAiCompatibleApiKey(aiGatewayApiKey: aiGatewayApiKey); + final output = await _arisLlmChatClient.chat( + endpoint: endpoint, + apiKey: apiKey, + model: model, + prompt: prompt, + systemPrompt: + 'You are the ARIS reviewer. Review the provided implementation and return actionable feedback.', + ); + return CliResult(output: output, error: '', exitCode: 0); + } catch (error) { + return CliResult(output: '', error: error.toString(), exitCode: -1); + } + } + + Future _runArisTesterViaClaudeReview({ + required String model, + required String prompt, + }) async { + try { + if (!await _goCoreLocator.isAvailable()) { + return const CliResult( + output: '', + error: 'Go core is unavailable for claude-review', + exitCode: -1, + ); + } + if (!await _binaryExists(_resolveCliPath('claude'))) { + return const CliResult( + output: '', + error: 'Claude CLI is unavailable for claude-review', + exitCode: -1, + ); + } + final output = await _arisLlmChatClient.claudeReview( + prompt: prompt, + model: model, + systemPrompt: + 'You are the ARIS reviewer. Review the provided implementation and return actionable feedback.', + ); + return CliResult(output: output, error: '', exitCode: 0); + } catch (error) { + return CliResult(output: '', error: error.toString(), exitCode: -1); + } + } + + Future _runOpenAiCompatiblePrompt({ + required MultiAgentRole role, + required String model, + required String prompt, + required String aiGatewayBaseUrl, + required String aiGatewayApiKey, + }) async { + final client = _httpClientFactory(); + _activeHttpClient = client; + try { + final request = await client.postUrl( + Uri.parse( + '${_openAiCompatibleBaseUrl(aiGatewayBaseUrl: aiGatewayBaseUrl).replaceAll(RegExp(r'/$'), '')}/chat/completions', + ), + ); + request.headers.set(HttpHeaders.contentTypeHeader, 'application/json'); + request.headers.set( + HttpHeaders.authorizationHeader, + 'Bearer ${_openAiCompatibleApiKey(aiGatewayApiKey: aiGatewayApiKey)}', + ); + request.add( + utf8.encode( + jsonEncode({ + 'model': model, + 'stream': false, + 'messages': >[ + { + 'role': 'system', + 'content': _systemPromptForRole(role), + }, + {'role': 'user', 'content': prompt}, + ], + }), + ), + ); + final response = await request.close().timeout( + Duration(seconds: _config.timeoutSeconds), + ); + final body = await utf8.decodeStream(response); + if (response.statusCode < 200 || response.statusCode >= 300) { + return CliResult( + output: '', + error: body, + exitCode: response.statusCode, + ); + } + final decoded = jsonDecode(body) as Map; + final choices = decoded['choices'] as List? ?? const []; + final firstChoice = choices.isNotEmpty ? choices.first : null; + final output = + ((firstChoice as Map?)?['message'] as Map?)?['content']?.toString() ?? + ''; + return CliResult(output: output, error: '', exitCode: 0); + } catch (error) { + return CliResult(output: '', error: error.toString(), exitCode: -1); + } finally { + _activeHttpClient = null; + try { + client.close(force: true); + } catch (_) { + // Best effort only. + } + } + } + + +} diff --git a/test/runtime/direct_single_agent_app_server_suite.dart b/test/runtime/direct_single_agent_app_server_suite.dart index a51231d3..ff4b011e 100644 --- a/test/runtime/direct_single_agent_app_server_suite.dart +++ b/test/runtime/direct_single_agent_app_server_suite.dart @@ -11,6 +11,18 @@ import 'package:xworkmate/runtime/runtime_models.dart'; void main() { group('DirectSingleAgentAppServerClient', () { + test('direct single-agent app-server core file stays split into focused parts', () { + final lines = File( + 'lib/runtime/direct_single_agent_app_server_client_core.part.dart', + ).readAsLinesSync(); + + expect( + lines.length, + lessThanOrEqualTo(1000), + reason: 'The core file should stay under the target line budget.', + ); + }); + test('classifies the four endpoint modes', () { expect( DirectSingleAgentEndpointDescriptor.describe( diff --git a/test/runtime/multi_agent_orchestrator_aris_suite.dart b/test/runtime/multi_agent_orchestrator_aris_suite.dart index 73c7e70f..59c10b9d 100644 --- a/test/runtime/multi_agent_orchestrator_aris_suite.dart +++ b/test/runtime/multi_agent_orchestrator_aris_suite.dart @@ -12,6 +12,18 @@ import 'package:xworkmate/runtime/multi_agent_orchestrator.dart'; import 'package:xworkmate/runtime/runtime_models.dart'; void main() { + test('multi-agent orchestrator core file stays split into focused parts', () { + final lines = File( + 'lib/runtime/multi_agent_orchestrator_core.part.dart', + ).readAsLinesSync(); + + expect( + lines.length, + lessThanOrEqualTo(1000), + reason: 'The core file should stay under the target line budget.', + ); + }); + test( 'MultiAgentOrchestrator falls back to local Ollama + ARIS Go core chat runtime', () async {