diff --git a/lib/app/app_controller_desktop_core.dart b/lib/app/app_controller_desktop_core.dart index 2171f112..e49ff029 100644 --- a/lib/app/app_controller_desktop_core.dart +++ b/lib/app/app_controller_desktop_core.dart @@ -42,6 +42,7 @@ import '../runtime/multi_agent_orchestrator.dart'; import '../runtime/platform_environment.dart'; import '../runtime/skill_directory_access.dart'; import 'task_thread_repositories.dart'; +import 'app_controller_openclaw_task_queue.dart'; import 'app_controller_desktop_navigation.dart'; import 'app_controller_desktop_gateway.dart'; import 'app_controller_desktop_settings.dart'; @@ -254,6 +255,14 @@ class AppController extends ChangeNotifier { return; } disposedInternal = true; + for (final turn in openClawGatewayQueuedTurnsInternal) { + turn.cancelled = true; + if (!turn.completer.isCompleted) { + turn.completer.complete(); + } + } + openClawGatewayQueuedTurnsInternal.clear(); + openClawGatewayQueuedTurnsBySessionInternal.clear(); unawaited(persistSharedSingleAgentLocalSkillsCacheInternal()); runtimeEventsSubscriptionInternal?.cancel(); detachChildListenersInternal(); @@ -335,6 +344,12 @@ class AppController extends ChangeNotifier { final Set aiGatewayAbortedSessionKeysInternal = {}; final Map> assistantThreadTurnQueuesInternal = >{}; + final List + openClawGatewayQueuedTurnsInternal = []; + final Map> + openClawGatewayQueuedTurnsBySessionInternal = + >{}; + int openClawGatewayActiveTasksInternal = 0; bool multiAgentRunPendingInternal = false; int localMessageCounterInternal = 0; diff --git a/lib/app/app_controller_desktop_settings.dart b/lib/app/app_controller_desktop_settings.dart index 4e7ef582..88386688 100644 --- a/lib/app/app_controller_desktop_settings.dart +++ b/lib/app/app_controller_desktop_settings.dart @@ -298,6 +298,15 @@ extension AppControllerDesktopSettings on AppController { aiGatewayPendingSessionKeysInternal.clear(); aiGatewayAbortedSessionKeysInternal.clear(); assistantThreadTurnQueuesInternal.clear(); + for (final turn in openClawGatewayQueuedTurnsInternal) { + turn.cancelled = true; + if (!turn.completer.isCompleted) { + turn.completer.complete(); + } + } + openClawGatewayQueuedTurnsInternal.clear(); + openClawGatewayQueuedTurnsBySessionInternal.clear(); + openClawGatewayActiveTasksInternal = 0; multiAgentRunPendingInternal = false; initializeAssistantThreadContext( 'main', diff --git a/lib/app/app_controller_desktop_thread_actions.dart b/lib/app/app_controller_desktop_thread_actions.dart index 44936394..48fd6e88 100644 --- a/lib/app/app_controller_desktop_thread_actions.dart +++ b/lib/app/app_controller_desktop_thread_actions.dart @@ -33,6 +33,7 @@ import '../runtime/agent_registry.dart'; import '../runtime/multi_agent_orchestrator.dart'; import '../runtime/platform_environment.dart'; import '../runtime/skill_directory_access.dart'; +import 'app_controller_openclaw_task_queue.dart'; import 'app_controller_desktop_core.dart'; import 'app_controller_desktop_navigation.dart'; import 'app_controller_desktop_gateway.dart'; @@ -320,18 +321,62 @@ extension AppControllerDesktopThreadActions on AppController { throw error; } } + final provider = assistantProviderForSession(sessionKey); + final model = assistantModelForSession(sessionKey); + final routing = buildExternalAcpRoutingForSessionInternal(sessionKey); + final dispatch = await codeAgentNodeOrchestratorInternal + .buildGatewayDispatch( + buildCodeAgentNodeStateInternal(executionTarget: currentTarget), + ); + final capturedSelectedSkillLabels = List.unmodifiable( + selectedSkillLabels, + ); + final capturedAttachments = List.unmodifiable( + attachments, + ); + final capturedLocalAttachments = List.unmodifiable( + localAttachments, + ); + if (usesOpenClawGatewayQueueInternal(currentTarget, provider)) { + await enqueueOpenClawGatewayTurnInternal( + OpenClawGatewayQueuedTurnInternal( + queueId: + 'openclaw-${DateTime.now().microsecondsSinceEpoch}-$localMessageCounterInternal', + sessionKey: sessionKey, + target: currentTarget, + provider: provider, + message: message, + thinking: thinking, + selectedSkillLabels: capturedSelectedSkillLabels, + attachments: capturedAttachments, + localAttachments: capturedLocalAttachments, + workingDirectory: workingDirectory, + remoteWorkingDirectoryHint: remoteWorkingDirectoryHint, + model: model, + routing: routing, + agentId: dispatch.agentId ?? '', + metadata: Map.unmodifiable(dispatch.metadata), + ), + ); + return; + } await enqueueThreadTurnInternal( sessionKey, () => runGatewayChatTurnInternal( sessionKey: sessionKey, target: currentTarget, + provider: provider, message: message, thinking: thinking, - selectedSkillLabels: selectedSkillLabels, - attachments: attachments, - localAttachments: localAttachments, + selectedSkillLabels: capturedSelectedSkillLabels, + attachments: capturedAttachments, + localAttachments: capturedLocalAttachments, workingDirectory: workingDirectory, remoteWorkingDirectoryHint: remoteWorkingDirectoryHint, + model: model, + routing: routing, + agentId: dispatch.agentId ?? '', + metadata: Map.unmodifiable(dispatch.metadata), ), ); recomputeTasksInternal(); @@ -340,6 +385,7 @@ extension AppControllerDesktopThreadActions on AppController { Future runGatewayChatTurnInternal({ required String sessionKey, required AssistantExecutionTarget target, + required SingleAgentProvider provider, required String message, required String thinking, required List selectedSkillLabels, @@ -347,6 +393,10 @@ extension AppControllerDesktopThreadActions on AppController { required List localAttachments, required String workingDirectory, required String remoteWorkingDirectoryHint, + required String model, + required ExternalCodeAgentAcpRoutingConfig routing, + required String agentId, + required Map metadata, }) async { final resumeSession = shouldResumeGatewaySessionForNextSendInternal( sessionKey, @@ -354,28 +404,23 @@ extension AppControllerDesktopThreadActions on AppController { appendGatewayUserTurnInternal(sessionKey, message); markGatewayChatRunInternal(sessionKey); try { - final dispatch = await codeAgentNodeOrchestratorInternal - .buildGatewayDispatch( - buildCodeAgentNodeStateInternal(executionTarget: target), - ); - markGatewayChatRunInternal(sessionKey); final result = await goTaskServiceClientInternal.executeTask( GoTaskServiceRequest( sessionId: sessionKey, threadId: sessionKey, target: target, - provider: assistantProviderForSession(sessionKey), + provider: provider, prompt: message, workingDirectory: workingDirectory, remoteWorkingDirectoryHint: remoteWorkingDirectoryHint, - model: assistantModelForSession(sessionKey), + model: model, thinking: thinking, selectedSkills: selectedSkillLabels, inlineAttachments: attachments, localAttachments: localAttachments, - agentId: dispatch.agentId ?? '', - metadata: dispatch.metadata, - routing: buildExternalAcpRoutingForSessionInternal(sessionKey), + agentId: agentId, + metadata: metadata, + routing: routing, routingHint: 'gateway', resumeSession: resumeSession, ), @@ -417,6 +462,169 @@ extension AppControllerDesktopThreadActions on AppController { } } + bool usesOpenClawGatewayQueueInternal( + AssistantExecutionTarget target, + SingleAgentProvider provider, + ) { + return target.isGateway && + provider.providerId == kCanonicalGatewayProviderId; + } + + Future enqueueOpenClawGatewayTurnInternal( + OpenClawGatewayQueuedTurnInternal turn, + ) async { + if (openClawGatewayActiveTasksInternal >= + openClawGatewayMaxActiveTasksInternal && + openClawGatewayQueuedTurnsInternal.length >= + openClawGatewayMaxQueuedTasksInternal) { + final error = StateError( + appText( + 'OpenClaw 任务队列已满，请等待当前任务完成后重试。', + 'OpenClaw task queue is full. Wait for the current tasks to finish and try again.', + ), + ); + await failOpenClawGatewayQueuedTurnInternal(turn.sessionKey, error); + throw error; + } + + openClawGatewayQueuedTurnsInternal.add(turn); + openClawGatewayQueuedTurnsBySessionInternal + .putIfAbsent( + turn.sessionKey, + () => [], + ) + .add(turn); + markOpenClawGatewayQueuedTurnInternal(turn.sessionKey); + drainOpenClawGatewayQueueInternal(); + await turn.completer.future; + } + + void markOpenClawGatewayQueuedTurnInternal(String sessionKey) { + upsertTaskThreadInternal( + sessionKey, + lifecycleStatus: 'queued', + lastResultCode: 'queued', + updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + ); + recomputeTasksInternal(); + notifyIfActiveInternal(); + } + + Future failOpenClawGatewayQueuedTurnInternal( + String sessionKey, + StateError error, + ) async { + upsertTaskThreadInternal( + sessionKey, + lifecycleStatus: 'ready', + lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + lastResultCode: 'OPENCLAW_GATEWAY_QUEUE_FULL', + lastArtifactSyncAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + lastArtifactSyncStatus: 'failed', + lastTaskArtifactRelativePaths: const [], + updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + ); + appendLocalSessionMessageInternal( + sessionKey, + assistantErrorMessageInternal(error.message), + persistInThreadContext: true, + ); + await flushAssistantThreadPersistenceInternal(); + recomputeTasksInternal(); + notifyIfActiveInternal(); + } + + bool abortQueuedOpenClawGatewayTurnInternal(String sessionKey) { + final normalizedSessionKey = normalizedAssistantSessionKeyInternal( + sessionKey, + ); + final queuedForSession = + openClawGatewayQueuedTurnsBySessionInternal[normalizedSessionKey]; + if (queuedForSession == null || queuedForSession.isEmpty) { + return false; + } + final turn = queuedForSession.removeAt(0); + if (queuedForSession.isEmpty) { + openClawGatewayQueuedTurnsBySessionInternal.remove(normalizedSessionKey); + } + openClawGatewayQueuedTurnsInternal.remove(turn); + turn.cancelled = true; + clearAiGatewayStreamingTextInternal(normalizedSessionKey); + upsertTaskThreadInternal( + normalizedSessionKey, + lifecycleStatus: 'ready', + lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + lastResultCode: 'aborted', + updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(), + ); + if (!turn.completer.isCompleted) { + turn.completer.complete(); + } + recomputeTasksInternal(); + notifyIfActiveInternal(); + drainOpenClawGatewayQueueInternal(); + return true; + } + + void drainOpenClawGatewayQueueInternal() { + while (openClawGatewayActiveTasksInternal < + openClawGatewayMaxActiveTasksInternal && + openClawGatewayQueuedTurnsInternal.isNotEmpty) { + final turn = openClawGatewayQueuedTurnsInternal.removeAt(0); + final queuedForSession = + openClawGatewayQueuedTurnsBySessionInternal[turn.sessionKey]; + queuedForSession?.remove(turn); + if (queuedForSession != null && queuedForSession.isEmpty) { + openClawGatewayQueuedTurnsBySessionInternal.remove(turn.sessionKey); + } + if (turn.cancelled) { + if (!turn.completer.isCompleted) { + turn.completer.complete(); + } + continue; + } + turn.started = true; + openClawGatewayActiveTasksInternal += 1; + unawaited(runOpenClawGatewayQueuedTurnInternal(turn)); + } + } + + Future runOpenClawGatewayQueuedTurnInternal( + OpenClawGatewayQueuedTurnInternal turn, + ) async { + try { + await enqueueThreadTurnInternal( + turn.sessionKey, + () => runGatewayChatTurnInternal( + sessionKey: turn.sessionKey, + target: turn.target, + provider: turn.provider, + message: turn.message, + thinking: turn.thinking, + selectedSkillLabels: turn.selectedSkillLabels, + attachments: turn.attachments, + localAttachments: turn.localAttachments, + workingDirectory: turn.workingDirectory, + remoteWorkingDirectoryHint: turn.remoteWorkingDirectoryHint, + model: turn.model, + routing: turn.routing, + agentId: turn.agentId, + metadata: turn.metadata, + ), + ); + if (!turn.completer.isCompleted) { + turn.completer.complete(); + } + } catch (error, stackTrace) { + if (!turn.completer.isCompleted) { + turn.completer.completeError(error, stackTrace); + } + } finally { + openClawGatewayActiveTasksInternal -= 1; + drainOpenClawGatewayQueueInternal(); + } + } + void appendGatewayUserTurnInternal(String sessionKey, String message) { final userText = message.trim().isEmpty ? 'See attached.' : message.trim(); appendLocalSessionMessageInternal( @@ -632,6 +840,9 @@ extension AppControllerDesktopThreadActions on AppController { final sessionKey = normalizedAssistantSessionKeyInternal( sessionsControllerInternal.currentSessionKey, ); + if (abortQueuedOpenClawGatewayTurnInternal(sessionKey)) { + return; + } if (aiGatewayPendingSessionKeysInternal.contains(sessionKey)) { await goTaskServiceClientInternal.cancelTask( route: GoTaskServiceRoute.externalAcpSingle, diff --git a/lib/app/app_controller_openclaw_task_queue.dart b/lib/app/app_controller_openclaw_task_queue.dart new file mode 100644 index 00000000..7c76fa3c --- /dev/null +++ b/lib/app/app_controller_openclaw_task_queue.dart @@ -0,0 +1,47 @@ +import 'dart:async'; + +import '../runtime/go_task_service_client.dart'; +import '../runtime/runtime_models.dart'; + +const int openClawGatewayMaxActiveTasksInternal = 1; +const int openClawGatewayMaxQueuedTasksInternal = 20; + +class OpenClawGatewayQueuedTurnInternal { + OpenClawGatewayQueuedTurnInternal({ + required this.queueId, + required this.sessionKey, + required this.target, + required this.provider, + required this.message, + required this.thinking, + required this.selectedSkillLabels, + required this.attachments, + required this.localAttachments, + required this.workingDirectory, + required this.remoteWorkingDirectoryHint, + required this.model, + required this.routing, + required this.agentId, + required this.metadata, + }); + + final String queueId; + final String sessionKey; + final AssistantExecutionTarget target; + final SingleAgentProvider provider; + final String message; + final String thinking; + final List selectedSkillLabels; + final List attachments; + final List localAttachments; + final String workingDirectory; + final String remoteWorkingDirectoryHint; + final String model; + final ExternalCodeAgentAcpRoutingConfig routing; + final String agentId; + final Map metadata; + final Completer completer = Completer(); + + bool cancelled = false; + bool started = false; +} diff --git a/test/runtime/assistant_execution_target_test.dart b/test/runtime/assistant_execution_target_test.dart index fd738baf..8ee1d4f5 100644 --- a/test/runtime/assistant_execution_target_test.dart +++ b/test/runtime/assistant_execution_target_test.dart @@ -1490,6 +1490,229 @@ void main() { ); }); + test( + 'OpenClaw gateway tasks queue globally and keep captured session context', + () async { + final fakeGoTaskService = _BlockingGoTaskServiceClient(); + final controller = _connectedGatewayController(fakeGoTaskService); + addTearDown(() { + fakeGoTaskService.completeAll(); + controller.dispose(); + }); + + await _selectGatewaySession(controller, 'queue-task-a'); + final taskAFuture = controller.sendChatMessage('same prompt'); + await fakeGoTaskService.waitForRequestCount(1); + + await _selectGatewaySession(controller, 'queue-task-b'); + final taskBFuture = controller.sendChatMessage('same prompt'); + await _selectGatewaySession(controller, 'queue-task-c'); + final taskCFuture = controller.sendChatMessage('different prompt'); + await _waitForThreadLifecycleStatus( + controller, + 'queue-task-b', + 'queued', + ); + await _waitForThreadLifecycleStatus( + controller, + 'queue-task-c', + 'queued', + ); + + expect(fakeGoTaskService.requests, hasLength(1)); + expect( + controller + .requireTaskThreadForSessionInternal('queue-task-b') + .lifecycleState + .status, + 'queued', + ); + expect( + controller + .requireTaskThreadForSessionInternal('queue-task-c') + .lifecycleState + .status, + 'queued', + ); + + fakeGoTaskService.complete( + 'queue-task-a', + const GoTaskServiceResult( + success: true, + message: 'result A', + turnId: 'turn-a', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + await taskAFuture; + await fakeGoTaskService.waitForRequestCount(2); + + final taskBRequest = fakeGoTaskService.requests[1]; + expect(taskBRequest.sessionId, 'queue-task-b'); + expect(taskBRequest.prompt, 'same prompt'); + expect(taskBRequest.resumeSession, isFalse); + expect(taskBRequest.workingDirectory, endsWith('/queue-task-b')); + expect( + taskBRequest.remoteWorkingDirectoryHint, + endsWith('/threads/queue-task-b'), + ); + + fakeGoTaskService.complete( + 'queue-task-b', + const GoTaskServiceResult( + success: true, + message: 'result B', + turnId: 'turn-b', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + await taskBFuture; + await fakeGoTaskService.waitForRequestCount(3); + + final taskCRequest = fakeGoTaskService.requests[2]; + expect(taskCRequest.sessionId, 'queue-task-c'); + expect(taskCRequest.prompt, 'different prompt'); + expect(taskCRequest.workingDirectory, endsWith('/queue-task-c')); + fakeGoTaskService.complete( + 'queue-task-c', + const GoTaskServiceResult( + success: true, + message: 'result C', + turnId: 'turn-c', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + await taskCFuture; + }, + ); + + test( + 'abortRun removes a queued OpenClaw task without bridge cancel', + () async { + final fakeGoTaskService = _BlockingGoTaskServiceClient(); + final controller = _connectedGatewayController(fakeGoTaskService); + addTearDown(() { + fakeGoTaskService.completeAll(); + controller.dispose(); + }); + + await _selectGatewaySession(controller, 'running-openclaw-task'); + final runningFuture = controller.sendChatMessage('running'); + await fakeGoTaskService.waitForRequestCount(1); + + await _selectGatewaySession(controller, 'queued-openclaw-task'); + final queuedFuture = controller.sendChatMessage('queued'); + await _waitForThreadLifecycleStatus( + controller, + 'queued-openclaw-task', + 'queued', + ); + expect(fakeGoTaskService.requests, hasLength(1)); + + await controller.abortRun(); + + expect(fakeGoTaskService.cancelledSessionIds, isEmpty); + expect( + controller + .requireTaskThreadForSessionInternal('queued-openclaw-task') + .lifecycleState + .lastResultCode, + 'aborted', + ); + await queuedFuture; + + fakeGoTaskService.complete( + 'running-openclaw-task', + const GoTaskServiceResult( + success: true, + message: 'running done', + turnId: 'turn-running', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + await runningFuture; + await Future.delayed(const Duration(milliseconds: 50)); + expect(fakeGoTaskService.requests, hasLength(1)); + }, + ); + + test('OpenClaw queue overflow fails without artifact sync', () async { + final fakeGoTaskService = _BlockingGoTaskServiceClient(); + final controller = _connectedGatewayController(fakeGoTaskService); + addTearDown(() { + fakeGoTaskService.completeAll(); + controller.dispose(); + }); + + await _selectGatewaySession(controller, 'queue-full-running'); + final runningFuture = controller.sendChatMessage('running'); + await fakeGoTaskService.waitForRequestCount(1); + + final queuedFutures = >[]; + for (var index = 0; index < 20; index += 1) { + final sessionKey = 'queue-full-waiting-$index'; + await _selectGatewaySession(controller, sessionKey); + queuedFutures.add(controller.sendChatMessage('queued $index')); + } + await _waitForOpenClawQueuedTaskCount(controller, 20); + + await _selectGatewaySession(controller, 'queue-full-overflow'); + await expectLater( + controller.sendChatMessage('overflow'), + throwsA(isA()), + ); + + expect(fakeGoTaskService.requests, hasLength(1)); + final overflowThread = controller.requireTaskThreadForSessionInternal( + 'queue-full-overflow', + ); + expect(overflowThread.lastArtifactSyncStatus, 'failed'); + expect(overflowThread.lastTaskArtifactRelativePaths, isEmpty); + + fakeGoTaskService.complete( + 'queue-full-running', + const GoTaskServiceResult( + success: true, + message: 'running done', + turnId: 'turn-running', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + await runningFuture; + for (var index = 0; index < 20; index += 1) { + final sessionKey = 'queue-full-waiting-$index'; + await fakeGoTaskService.waitForRequestCount(index + 2); + fakeGoTaskService.complete( + sessionKey, + GoTaskServiceResult( + success: true, + message: 'queued $index done', + turnId: 'turn-queued-$index', + raw: const {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + } + await Future.wait(queuedFutures); + }); + test( 'sendChatMessage restarts stale interrupted and error states', () async { @@ -1794,6 +2017,76 @@ AppController _connectedController(GoTaskServiceClient client) { ); } +AppController _connectedGatewayController(GoTaskServiceClient client) { + return AppController( + goTaskServiceClient: client, + environmentOverride: const { + 'BRIDGE_AUTH_TOKEN': 'bridge-token', + }, + initialBridgeProviderCatalog: const [ + SingleAgentProvider.codex, + ], + initialGatewayProviderCatalog: const [ + SingleAgentProvider.openclaw, + ], + initialAvailableExecutionTargets: const [ + AssistantExecutionTarget.agent, + AssistantExecutionTarget.gateway, + ], + ); +} + +Future _selectGatewaySession( + AppController controller, + String sessionKey, +) async { + await controller.switchSession(sessionKey); + await controller.setAssistantExecutionTarget( + AssistantExecutionTarget.gateway, + ); +} + +Future _waitForThreadLifecycleStatus( + AppController controller, + String sessionKey, + String status, +) async { + final deadline = DateTime.now().add(const Duration(seconds: 15)); + while (DateTime.now().isBefore(deadline)) { + final currentStatus = controller + .taskThreadForSessionInternal(sessionKey) + ?.lifecycleState + .status; + if (currentStatus == status) { + return; + } + await Future.delayed(const Duration(milliseconds: 10)); + } + final currentStatus = controller + .taskThreadForSessionInternal(sessionKey) + ?.lifecycleState + .status; + throw StateError( + 'Timed out waiting for $sessionKey status $status. Current status: $currentStatus.', + ); +} + +Future _waitForOpenClawQueuedTaskCount( + AppController controller, + int count, +) async { + final deadline = DateTime.now().add(const Duration(seconds: 15)); + while (DateTime.now().isBefore(deadline)) { + if (controller.openClawGatewayQueuedTurnsInternal.length >= count) { + return; + } + await Future.delayed(const Duration(milliseconds: 10)); + } + throw StateError( + 'Timed out waiting for $count queued OpenClaw tasks. Current count: ${controller.openClawGatewayQueuedTurnsInternal.length}.', + ); +} + class _RecordingGoTaskServiceClient implements GoTaskServiceClient { int executeCount = 0; final List requests = []; @@ -1923,6 +2216,23 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient { completer.complete(result); } + void completeAll([ + GoTaskServiceResult result = const GoTaskServiceResult( + success: true, + message: 'cleanup', + turnId: 'turn-cleanup', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ]) { + final pendingSessionIds = List.from(_pending.keys); + for (final sessionId in pendingSessionIds) { + complete(sessionId, result); + } + } + void emitDelta(String sessionId, String text) { final onUpdate = _updates[sessionId]; if (onUpdate == null) {