ai-workspace-lab/xworkmate-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Queue OpenClaw gateway tasks locally

Browse Source
This commit is contained in:
Haitao Pan 2026-05-11 11:45:32 +08:00
parent 0e9ccb7ed4
commit 23adf1f176
5 changed files with 605 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
lib/app/app_controller_desktop_core.dart
View File

@ -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<String> aiGatewayAbortedSessionKeysInternal = <String>{};
final Map<String, Future<void>> assistantThreadTurnQueuesInternal =
<String, Future<void>>{};
final List<OpenClawGatewayQueuedTurnInternal>
openClawGatewayQueuedTurnsInternal = <OpenClawGatewayQueuedTurnInternal>[];
final Map<String, List<OpenClawGatewayQueuedTurnInternal>>
openClawGatewayQueuedTurnsBySessionInternal =
<String, List<OpenClawGatewayQueuedTurnInternal>>{};
int openClawGatewayActiveTasksInternal = 0;
bool multiAgentRunPendingInternal = false;
int localMessageCounterInternal = 0;

9
lib/app/app_controller_desktop_settings.dart
View File

@ -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',

237
lib/app/app_controller_desktop_thread_actions.dart
View File

@ -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<String>.unmodifiable(
selectedSkillLabels,
);
final capturedAttachments = List<GatewayChatAttachmentPayload>.unmodifiable(
attachments,
);
final capturedLocalAttachments = List<CollaborationAttachment>.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<String, dynamic>.unmodifiable(dispatch.metadata),
),
);
return;
}
await enqueueThreadTurnInternal<void>(
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<String, dynamic>.unmodifiable(dispatch.metadata),
),
);
recomputeTasksInternal();
@ -340,6 +385,7 @@ extension AppControllerDesktopThreadActions on AppController {
Future<void> runGatewayChatTurnInternal({
required String sessionKey,
required AssistantExecutionTarget target,
required SingleAgentProvider provider,
required String message,
required String thinking,
required List<String> selectedSkillLabels,
@ -347,6 +393,10 @@ extension AppControllerDesktopThreadActions on AppController {
required List<CollaborationAttachment> localAttachments,
required String workingDirectory,
required String remoteWorkingDirectoryHint,
required String model,
required ExternalCodeAgentAcpRoutingConfig routing,
required String agentId,
required Map<String, dynamic> 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<void> 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,
() => <OpenClawGatewayQueuedTurnInternal>[],
)
.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<void> 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 <String>[],
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<void> runOpenClawGatewayQueuedTurnInternal(
OpenClawGatewayQueuedTurnInternal turn,
) async {
try {
await enqueueThreadTurnInternal<void>(
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,

47
lib/app/app_controller_openclaw_task_queue.dart Normal file
View File

@ -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<String> selectedSkillLabels;
final List<GatewayChatAttachmentPayload> attachments;
final List<CollaborationAttachment> localAttachments;
final String workingDirectory;
final String remoteWorkingDirectoryHint;
final String model;
final ExternalCodeAgentAcpRoutingConfig routing;
final String agentId;
final Map<String, dynamic> metadata;
final Completer<void> completer = Completer<void>();
bool cancelled = false;
bool started = false;
}

310
test/runtime/assistant_execution_target_test.dart
View File

@ -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: <String, dynamic>{},
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: <String, dynamic>{},
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: <String, dynamic>{},
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: <String, dynamic>{},
errorMessage: '',
resolvedModel: '',
route: GoTaskServiceRoute.externalAcpSingle,
),
);
await runningFuture;
await Future<void>.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 = <Future<void>>[];
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<StateError>()),
);
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: <String, dynamic>{},
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 <String, dynamic>{},
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 <String, String>{
'BRIDGE_AUTH_TOKEN': 'bridge-token',
},
initialBridgeProviderCatalog: const <SingleAgentProvider>[
SingleAgentProvider.codex,
],
initialGatewayProviderCatalog: const <SingleAgentProvider>[
SingleAgentProvider.openclaw,
],
initialAvailableExecutionTargets: const <AssistantExecutionTarget>[
AssistantExecutionTarget.agent,
AssistantExecutionTarget.gateway,
],
);
}
Future<void> _selectGatewaySession(
AppController controller,
String sessionKey,
) async {
await controller.switchSession(sessionKey);
await controller.setAssistantExecutionTarget(
AssistantExecutionTarget.gateway,
);
}
Future<void> _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<void>.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<void> _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<void>.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<GoTaskServiceRequest> requests = <GoTaskServiceRequest>[];
@ -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: <String, dynamic>{},
errorMessage: '',
resolvedModel: '',
route: GoTaskServiceRoute.externalAcpSingle,
),
]) {
final pendingSessionIds = List<String>.from(_pending.keys);
for (final sessionId in pendingSessionIds) {
complete(sessionId, result);
}
}
void emitDelta(String sessionId, String text) {
final onUpdate = _updates[sessionId];
if (onUpdate == null) {