diff --git a/lib/app/app_controller_desktop_core.dart b/lib/app/app_controller_desktop_core.dart index 32a46445..4b77df43 100644 --- a/lib/app/app_controller_desktop_core.dart +++ b/lib/app/app_controller_desktop_core.dart @@ -203,8 +203,12 @@ class AppController extends ChangeNotifier { for (final turn in openClawGatewayQueuedTurnsInternal) { turn.cancelled = true; } + for (final turn in openClawGatewayActiveTurnsInternal.values) { + turn.cancelled = true; + } openClawGatewayQueuedTurnsInternal.clear(); openClawGatewayQueuedTurnsBySessionInternal.clear(); + openClawGatewayActiveTurnsInternal.clear(); runtimeEventsSubscriptionInternal?.cancel(); detachChildListenersInternal(); runtimeCoordinatorInternal.dispose(); @@ -282,7 +286,11 @@ class AppController extends ChangeNotifier { final Map> openClawGatewayQueuedTurnsBySessionInternal = >{}; - int openClawGatewayActiveTasksInternal = 0; + final Map + openClawGatewayActiveTurnsInternal = + {}; + int get openClawGatewayActiveTasksInternal => + openClawGatewayActiveTurnsInternal.length; bool multiAgentRunPendingInternal = false; int localMessageCounterInternal = 0; int assistantDraftSessionCounterInternal = 0; diff --git a/lib/app/app_controller_desktop_settings.dart b/lib/app/app_controller_desktop_settings.dart index 4fca2500..4432c535 100644 --- a/lib/app/app_controller_desktop_settings.dart +++ b/lib/app/app_controller_desktop_settings.dart @@ -298,9 +298,12 @@ extension AppControllerDesktopSettings on AppController { for (final turn in openClawGatewayQueuedTurnsInternal) { turn.cancelled = true; } + for (final turn in openClawGatewayActiveTurnsInternal.values) { + turn.cancelled = true; + } openClawGatewayQueuedTurnsInternal.clear(); openClawGatewayQueuedTurnsBySessionInternal.clear(); - openClawGatewayActiveTasksInternal = 0; + openClawGatewayActiveTurnsInternal.clear(); multiAgentRunPendingInternal = false; final sessionKey = createAssistantDraftSessionKeyInternal(); initializeAssistantThreadContext( diff --git a/lib/app/app_controller_desktop_thread_actions.dart b/lib/app/app_controller_desktop_thread_actions.dart index f9314a88..031bd64c 100644 --- a/lib/app/app_controller_desktop_thread_actions.dart +++ b/lib/app/app_controller_desktop_thread_actions.dart @@ -3,7 +3,6 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; -import 'dart:math' as math; import 'package:flutter/material.dart'; import 'app_metadata.dart'; import 'app_capabilities.dart'; @@ -704,7 +703,7 @@ extension AppControllerDesktopThreadActions on AppController { if (turn.appendUserTurn) { appendGatewayUserTurnInternal(turn.sessionKey, turn.message); } - if (openClawGatewayActiveTasksInternal >= + if (openClawGatewayActiveTurnsInternal.length >= openClawGatewayMaxActiveTasksInternal && openClawGatewayQueuedTurnsInternal.length >= openClawGatewayMaxQueuedTasksInternal) { @@ -800,6 +799,24 @@ extension AppControllerDesktopThreadActions on AppController { return true; } + bool removeActiveOpenClawGatewayTurnsForSessionInternal(String sessionKey) { + final normalizedSessionKey = normalizedAssistantSessionKeyInternal( + sessionKey, + ); + var removed = false; + openClawGatewayActiveTurnsInternal.removeWhere((_, turn) { + final matches = + normalizedAssistantSessionKeyInternal(turn.sessionKey) == + normalizedSessionKey; + if (matches) { + turn.cancelled = true; + removed = true; + } + return matches; + }); + return removed; + } + void markOpenClawGatewayTurnAbortedInternal(String sessionKey) { final normalizedSessionKey = normalizedAssistantSessionKeyInternal( sessionKey, @@ -834,7 +851,7 @@ extension AppControllerDesktopThreadActions on AppController { } void drainOpenClawGatewayQueueInternal() { - while (openClawGatewayActiveTasksInternal < + while (openClawGatewayActiveTurnsInternal.length < openClawGatewayMaxActiveTasksInternal && openClawGatewayQueuedTurnsInternal.isNotEmpty) { final turn = openClawGatewayQueuedTurnsInternal.removeAt(0); @@ -842,11 +859,29 @@ extension AppControllerDesktopThreadActions on AppController { if (turn.cancelled) { continue; } - openClawGatewayActiveTasksInternal += 1; + openClawGatewayActiveTurnsInternal[turn.queueId] = turn; + markOpenClawGatewayRunningTurnInternal(turn.sessionKey); unawaited(runOpenClawGatewayQueuedTurnInternal(turn)); } } + void markOpenClawGatewayRunningTurnInternal(String sessionKey) { + final startedAtMs = DateTime.now().millisecondsSinceEpoch.toDouble(); + aiGatewayPendingSessionKeysInternal.add(sessionKey); + upsertTaskThreadInternal( + sessionKey, + lifecycleStatus: 'running', + lastRunAtMs: startedAtMs, + lastResultCode: 'running', + lastArtifactSyncAtMs: startedAtMs, + lastArtifactSyncStatus: 'running', + lastTaskArtifactRelativePaths: const [], + updatedAtMs: startedAtMs, + ); + recomputeTasksInternal(); + notifyIfActiveInternal(); + } + Future runOpenClawGatewayQueuedTurnInternal( OpenClawGatewayQueuedTurnInternal turn, ) async { @@ -883,10 +918,7 @@ extension AppControllerDesktopThreadActions on AppController { ); } } finally { - openClawGatewayActiveTasksInternal = math.max( - 0, - openClawGatewayActiveTasksInternal - 1, - ); + openClawGatewayActiveTurnsInternal.remove(turn.queueId); if (!disposedInternal) { drainOpenClawGatewayQueueInternal(); recomputeTasksInternal(); @@ -1227,7 +1259,9 @@ extension AppControllerDesktopThreadActions on AppController { // Best effort cancellation only. Local state must still leave pending. } removeQueuedOpenClawGatewayTurnsForSessionInternal(sessionKey); + removeActiveOpenClawGatewayTurnsForSessionInternal(sessionKey); markOpenClawGatewayTurnAbortedInternal(sessionKey); + drainOpenClawGatewayQueueInternal(); return; } } diff --git a/lib/app/app_controller_desktop_workspace_execution.dart b/lib/app/app_controller_desktop_workspace_execution.dart index 44ee304a..b983ede0 100644 --- a/lib/app/app_controller_desktop_workspace_execution.dart +++ b/lib/app/app_controller_desktop_workspace_execution.dart @@ -502,6 +502,15 @@ extension AppControllerDesktopWorkspaceExecution on AppController { normalizedAssistantSessionKeyInternal(turn.sessionKey) == normalizedSessionKey, ); + openClawGatewayActiveTurnsInternal.removeWhere((_, turn) { + final matches = + normalizedAssistantSessionKeyInternal(turn.sessionKey) == + normalizedSessionKey; + if (matches) { + turn.cancelled = true; + } + return matches; + }); recomputeTasksInternal(); if (wasCurrent) { await ensureActiveAssistantThreadInternal(); diff --git a/lib/app/app_controller_openclaw_task_queue.dart b/lib/app/app_controller_openclaw_task_queue.dart index 32e475de..7e3f8e3e 100644 --- a/lib/app/app_controller_openclaw_task_queue.dart +++ b/lib/app/app_controller_openclaw_task_queue.dart @@ -1,7 +1,7 @@ import '../runtime/go_task_service_client.dart'; import '../runtime/runtime_models.dart'; -const int openClawGatewayMaxActiveTasksInternal = 1; +const int openClawGatewayMaxActiveTasksInternal = 5; const int openClawGatewayMaxQueuedTasksInternal = 20; class OpenClawGatewayQueuedTurnInternal { diff --git a/test/runtime/assistant_execution_target_test.dart b/test/runtime/assistant_execution_target_test.dart index 5982f9c2..c5b079a5 100644 --- a/test/runtime/assistant_execution_target_test.dart +++ b/test/runtime/assistant_execution_target_test.dart @@ -2500,88 +2500,78 @@ void main() { controller.dispose(); }); - await _selectGatewaySession(controller, 'queue-task-a'); - final taskAFuture = controller.sendChatMessage('same prompt'); - await fakeGoTaskService.waitForRequestCount(1); - await expectLater( - taskAFuture.timeout(const Duration(milliseconds: 250)), - completes, - ); - - await _selectGatewaySession(controller, 'queue-task-b'); final queuedAttachment = GatewayChatAttachmentPayload( type: 'file', mimeType: 'text/plain', fileName: 'queued.txt', content: base64Encode(utf8.encode('queued content')), ); - final taskBFuture = controller.sendChatMessage( - 'same prompt', - attachments: [queuedAttachment], - ); - await _waitForThreadLifecycleStatus( - controller, - 'queue-task-b', - 'queued', - ); - await _selectGatewaySession(controller, 'queue-task-c'); - final taskCFuture = controller.sendChatMessage('different prompt'); - await _waitForThreadLifecycleStatus( - controller, - 'queue-task-c', - 'queued', - ); - await expectLater( - taskBFuture.timeout(const Duration(milliseconds: 250)), - completes, - ); - await expectLater( - taskCFuture.timeout(const Duration(milliseconds: 250)), - completes, - ); + for ( + var index = 0; + index < openClawGatewayMaxActiveTasksInternal; + index += 1 + ) { + final sessionKey = 'queue-task-$index'; + await _selectGatewaySession(controller, sessionKey); + await expectLater( + controller + .sendChatMessage('active prompt $index') + .timeout(const Duration(milliseconds: 250)), + completes, + ); + await fakeGoTaskService.waitForRequestCount(index + 1); + expect( + controller + .requireTaskThreadForSessionInternal(sessionKey) + .lifecycleState + .status, + 'running', + ); + } - expect(fakeGoTaskService.requests, hasLength(1)); + await _selectGatewaySession(controller, 'queue-task-waiting'); + await expectLater( + controller + .sendChatMessage( + 'queued prompt', + attachments: [queuedAttachment], + ) + .timeout(const Duration(milliseconds: 250)), + completes, + ); + await _waitForThreadLifecycleStatus( + controller, + 'queue-task-waiting', + 'queued', + ); + expect( + fakeGoTaskService.requests, + hasLength(openClawGatewayMaxActiveTasksInternal), + ); expect( controller - .requireTaskThreadForSessionInternal('queue-task-b') + .requireTaskThreadForSessionInternal('queue-task-waiting') .lifecycleState .status, 'queued', ); expect( - controller - .requireTaskThreadForSessionInternal('queue-task-c') - .lifecycleState - .status, - 'queued', - ); - expect( - controller.assistantSessionHasPendingRun('queue-task-b'), + controller.assistantSessionHasPendingRun('queue-task-waiting'), isTrue, ); expect( - controller.assistantSessionHasPendingRun('queue-task-c'), - isTrue, - ); - expect( - controller.localSessionMessagesInternal['queue-task-b']!.map( + controller.localSessionMessagesInternal['queue-task-waiting']!.map( (message) => message.text, ), - contains('same prompt'), - ); - expect( - controller.localSessionMessagesInternal['queue-task-c']!.map( - (message) => message.text, - ), - contains('different prompt'), + contains('queued prompt'), ); fakeGoTaskService.complete( - 'queue-task-a', + 'queue-task-0', const GoTaskServiceResult( success: true, - message: 'result A', - turnId: 'turn-a', + message: 'result 0', + turnId: 'turn-0', raw: {}, errorMessage: '', resolvedModel: '', @@ -2590,36 +2580,41 @@ void main() { ); await _waitForThreadLifecycleStatus( controller, - 'queue-task-a', + 'queue-task-0', 'ready', ); - await fakeGoTaskService.waitForRequestCount(2); + await fakeGoTaskService.waitForRequestCount( + openClawGatewayMaxActiveTasksInternal + 1, + ); - final taskBRequest = fakeGoTaskService.requests[1]; - expect(taskBRequest.sessionId, 'queue-task-b'); - expect(taskBRequest.prompt, contains('TaskThread workspace context:')); - expect(taskBRequest.prompt, contains('- sessionKey: queue-task-b')); - expect(taskBRequest.prompt, contains('User request:\nsame prompt')); - expect(taskBRequest.resumeSession, isFalse); - expect(taskBRequest.inlineAttachments, hasLength(1)); - expect(taskBRequest.inlineAttachments.single.fileName, 'queued.txt'); + final queuedRequest = fakeGoTaskService.requests.last; + expect(queuedRequest.sessionId, 'queue-task-waiting'); + expect(queuedRequest.prompt, contains('TaskThread workspace context:')); expect( - taskBRequest.inlineAttachments.single.content, + queuedRequest.prompt, + contains('- sessionKey: queue-task-waiting'), + ); + expect(queuedRequest.prompt, contains('User request:\nqueued prompt')); + expect(queuedRequest.resumeSession, isFalse); + expect(queuedRequest.inlineAttachments, hasLength(1)); + expect(queuedRequest.inlineAttachments.single.fileName, 'queued.txt'); + expect( + queuedRequest.inlineAttachments.single.content, queuedAttachment.content, ); - expect(taskBRequest.workingDirectory, endsWith('/queue-task-b')); - expect(taskBRequest.prompt, contains(taskBRequest.workingDirectory)); + expect(queuedRequest.workingDirectory, endsWith('/queue-task-waiting')); + expect(queuedRequest.prompt, contains(queuedRequest.workingDirectory)); expect( - taskBRequest.remoteWorkingDirectoryHint, - endsWith('/threads/queue-task-b'), + queuedRequest.remoteWorkingDirectoryHint, + endsWith('/threads/queue-task-waiting'), ); fakeGoTaskService.complete( - 'queue-task-b', + 'queue-task-waiting', const GoTaskServiceResult( success: true, - message: 'result B', - turnId: 'turn-b', + message: 'queued done', + turnId: 'turn-waiting', raw: {}, errorMessage: '', resolvedModel: '', @@ -2628,45 +2623,18 @@ void main() { ); await _waitForThreadLifecycleStatus( controller, - 'queue-task-b', + 'queue-task-waiting', 'ready', ); expect( - controller.localSessionMessagesInternal['queue-task-b']! + controller.localSessionMessagesInternal['queue-task-waiting']! .where( (message) => - message.role == 'user' && message.text == 'same prompt', + message.role == 'user' && message.text == 'queued prompt', ) .length, 1, ); - await fakeGoTaskService.waitForRequestCount(3); - - final taskCRequest = fakeGoTaskService.requests[2]; - expect(taskCRequest.sessionId, 'queue-task-c'); - expect( - taskCRequest.prompt, - contains('User request:\ndifferent prompt'), - ); - expect(taskCRequest.workingDirectory, endsWith('/queue-task-c')); - expect(taskCRequest.prompt, contains(taskCRequest.workingDirectory)); - fakeGoTaskService.complete( - 'queue-task-c', - const GoTaskServiceResult( - success: true, - message: 'result C', - turnId: 'turn-c', - raw: {}, - errorMessage: '', - resolvedModel: '', - route: GoTaskServiceRoute.externalAcpSingle, - ), - ); - await _waitForThreadLifecycleStatus( - controller, - 'queue-task-c', - 'ready', - ); }, ); @@ -2732,9 +2700,11 @@ void main() { controller.dispose(); }); - await _selectGatewaySession(controller, 'running-openclaw-task'); - final runningFuture = controller.sendChatMessage('running'); - await fakeGoTaskService.waitForRequestCount(1); + final activeSessionKeys = await _startOpenClawActiveTasks( + controller, + fakeGoTaskService, + prefix: 'running-openclaw-task', + ); await _selectGatewaySession(controller, 'queued-openclaw-task'); final queuedFuture = controller.sendChatMessage('queued'); @@ -2743,7 +2713,10 @@ void main() { 'queued-openclaw-task', 'queued', ); - expect(fakeGoTaskService.requests, hasLength(1)); + expect( + fakeGoTaskService.requests, + hasLength(openClawGatewayMaxActiveTasksInternal), + ); await controller.abortRun(); @@ -2758,7 +2731,7 @@ void main() { await queuedFuture; fakeGoTaskService.complete( - 'running-openclaw-task', + activeSessionKeys.first, const GoTaskServiceResult( success: true, message: 'running done', @@ -2769,14 +2742,16 @@ void main() { route: GoTaskServiceRoute.externalAcpSingle, ), ); - await runningFuture; await _waitForThreadLifecycleStatus( controller, - 'running-openclaw-task', + activeSessionKeys.first, 'ready', ); await Future.delayed(const Duration(milliseconds: 50)); - expect(fakeGoTaskService.requests, hasLength(1)); + expect( + fakeGoTaskService.requests, + hasLength(openClawGatewayMaxActiveTasksInternal), + ); }, ); @@ -2790,9 +2765,12 @@ void main() { controller.dispose(); }); - await _selectGatewaySession(controller, 'running-openclaw-stop-task'); - final runningFuture = controller.sendChatMessage('running'); - await fakeGoTaskService.waitForRequestCount(1); + final activeSessionKeys = await _startOpenClawActiveTasks( + controller, + fakeGoTaskService, + prefix: 'running-openclaw-stop-task', + ); + final runningSessionKey = activeSessionKeys.first; await _selectGatewaySession(controller, 'queued-openclaw-after-stop'); final queuedFuture = controller.sendChatMessage('queued'); @@ -2802,25 +2780,30 @@ void main() { 'queued', ); - await _selectGatewaySession(controller, 'running-openclaw-stop-task'); + await _selectGatewaySession(controller, runningSessionKey); await controller.abortRun(); expect(fakeGoTaskService.cancelledSessionIds, [ - 'running-openclaw-stop-task', + runningSessionKey, ]); expect( - controller.assistantSessionHasPendingRun( - 'running-openclaw-stop-task', - ), + controller.assistantSessionHasPendingRun(runningSessionKey), isFalse, ); expect( controller - .requireTaskThreadForSessionInternal('running-openclaw-stop-task') + .requireTaskThreadForSessionInternal(runningSessionKey) .lifecycleState .lastResultCode, 'aborted', ); + await fakeGoTaskService.waitForRequestCount( + openClawGatewayMaxActiveTasksInternal + 1, + ); + expect( + fakeGoTaskService.requests.last.sessionId, + 'queued-openclaw-after-stop', + ); expect( controller.assistantSessionHasPendingRun( 'queued-openclaw-after-stop', @@ -2829,7 +2812,7 @@ void main() { ); fakeGoTaskService.complete( - 'running-openclaw-stop-task', + runningSessionKey, const GoTaskServiceResult( success: true, message: 'late stopped result', @@ -2840,12 +2823,6 @@ void main() { route: GoTaskServiceRoute.externalAcpSingle, ), ); - await runningFuture; - await fakeGoTaskService.waitForRequestCount(2); - expect( - fakeGoTaskService.requests.last.sessionId, - 'queued-openclaw-after-stop', - ); fakeGoTaskService.complete( 'queued-openclaw-after-stop', @@ -2865,7 +2842,10 @@ void main() { 'queued-openclaw-after-stop', 'ready', ); - expect(fakeGoTaskService.requests, hasLength(2)); + expect( + fakeGoTaskService.requests, + hasLength(openClawGatewayMaxActiveTasksInternal + 1), + ); }, ); @@ -2879,9 +2859,11 @@ void main() { controller.dispose(); }); - await _selectGatewaySession(controller, 'continue-active-openclaw'); - await controller.sendChatMessage('active task'); - await fakeGoTaskService.waitForRequestCount(1); + final activeSessionKeys = await _startOpenClawActiveTasks( + controller, + fakeGoTaskService, + prefix: 'continue-active-openclaw', + ); await _selectGatewaySession(controller, 'continue-queued-openclaw'); await controller.sendChatMessage('queued before continue'); @@ -2925,7 +2907,10 @@ void main() { 'continue-stopped-openclaw', 'queued', ); - expect(fakeGoTaskService.requests, hasLength(1)); + expect( + fakeGoTaskService.requests, + hasLength(openClawGatewayMaxActiveTasksInternal), + ); expect( controller.assistantSessionHasPendingRun('continue-queued-openclaw'), isTrue, @@ -2946,7 +2931,7 @@ void main() { ); fakeGoTaskService.complete( - 'continue-active-openclaw', + activeSessionKeys.first, const GoTaskServiceResult( success: true, message: 'active done', @@ -2957,7 +2942,9 @@ void main() { route: GoTaskServiceRoute.externalAcpSingle, ), ); - await fakeGoTaskService.waitForRequestCount(2); + await fakeGoTaskService.waitForRequestCount( + openClawGatewayMaxActiveTasksInternal + 1, + ); expect( fakeGoTaskService.requests.last.sessionId, 'continue-queued-openclaw', @@ -2975,7 +2962,21 @@ void main() { route: GoTaskServiceRoute.externalAcpSingle, ), ); - await fakeGoTaskService.waitForRequestCount(3); + fakeGoTaskService.complete( + activeSessionKeys[1], + const GoTaskServiceResult( + success: true, + message: 'second active done', + turnId: 'turn-second-active', + raw: {}, + errorMessage: '', + resolvedModel: '', + route: GoTaskServiceRoute.externalAcpSingle, + ), + ); + await fakeGoTaskService.waitForRequestCount( + openClawGatewayMaxActiveTasksInternal + 2, + ); final continuedRequest = fakeGoTaskService.requests.last; expect(continuedRequest.sessionId, 'continue-stopped-openclaw'); @@ -3027,13 +3028,93 @@ void main() { }, ); + test( + 'OpenClaw drain starts queued work when no active turns remain', + () async { + final fakeGoTaskService = _BlockingGoTaskServiceClient(); + final controller = _connectedGatewayController(fakeGoTaskService); + addTearDown(() { + fakeGoTaskService.completeAll(); + controller.dispose(); + }); + + const sessionKey = 'stale-slot-queued-task'; + await _selectGatewaySession(controller, sessionKey); + final turn = OpenClawGatewayQueuedTurnInternal( + queueId: 'stale-slot-queued-turn', + sessionKey: sessionKey, + target: AssistantExecutionTarget.gateway, + provider: SingleAgentProvider.openclaw, + message: 'recover queued work', + thinking: 'off', + selectedSkillLabels: const [], + attachments: const [], + localAttachments: const [], + workingDirectory: '/tmp/$sessionKey', + localWorkingDirectory: '/tmp/$sessionKey-local', + remoteWorkingDirectoryHint: '/threads/$sessionKey', + model: '', + routing: const ExternalCodeAgentAcpRoutingConfig.auto( + preferredGatewayTarget: kCanonicalGatewayProviderId, + ), + agentId: '', + metadata: const {}, + resumeSessionHint: false, + ); + controller.openClawGatewayQueuedTurnsInternal.add(turn); + controller.openClawGatewayQueuedTurnsBySessionInternal[sessionKey] = + [turn]; + controller.markOpenClawGatewayQueuedTurnInternal(sessionKey); + + controller.drainOpenClawGatewayQueueInternal(); + + await fakeGoTaskService.waitForRequestCount(1); + expect(fakeGoTaskService.requests.single.sessionId, sessionKey); + expect( + controller + .requireTaskThreadForSessionInternal(sessionKey) + .lifecycleState + .status, + 'running', + ); + expect(controller.openClawGatewayActiveTasksInternal, 1); + }, + ); + test('OpenClaw queue overflow fails without artifact sync', () async { final fakeGoTaskService = _BlockingGoTaskServiceClient(); final controller = _connectedGatewayController(fakeGoTaskService); addTearDown(controller.dispose); - controller.openClawGatewayActiveTasksInternal = - openClawGatewayMaxActiveTasksInternal; + for ( + var index = 0; + index < openClawGatewayMaxActiveTasksInternal; + index += 1 + ) { + final sessionKey = 'queue-full-active-$index'; + final turn = OpenClawGatewayQueuedTurnInternal( + queueId: 'queue-full-active-$index', + sessionKey: sessionKey, + target: AssistantExecutionTarget.gateway, + provider: SingleAgentProvider.openclaw, + message: 'active $index', + thinking: 'off', + selectedSkillLabels: const [], + attachments: const [], + localAttachments: const [], + workingDirectory: '/tmp/$sessionKey', + localWorkingDirectory: '/tmp/$sessionKey-local', + remoteWorkingDirectoryHint: '/threads/$sessionKey', + model: '', + routing: const ExternalCodeAgentAcpRoutingConfig.auto( + preferredGatewayTarget: kCanonicalGatewayProviderId, + ), + agentId: '', + metadata: const {}, + resumeSessionHint: false, + ); + controller.openClawGatewayActiveTurnsInternal[turn.queueId] = turn; + } for ( var index = 0; index < openClawGatewayMaxQueuedTasksInternal; @@ -3609,6 +3690,38 @@ Future _selectGatewaySession( ); } +Future> _startOpenClawActiveTasks( + AppController controller, + _BlockingGoTaskServiceClient fakeGoTaskService, { + required String prefix, +}) async { + final sessionKeys = []; + for ( + var index = 0; + index < openClawGatewayMaxActiveTasksInternal; + index += 1 + ) { + final sessionKey = '$prefix-$index'; + sessionKeys.add(sessionKey); + await _selectGatewaySession(controller, sessionKey); + await expectLater( + controller + .sendChatMessage('active task $index') + .timeout(const Duration(milliseconds: 250)), + completes, + ); + await fakeGoTaskService.waitForRequestCount(index + 1); + expect( + controller + .requireTaskThreadForSessionInternal(sessionKey) + .lifecycleState + .status, + 'running', + ); + } + return sessionKeys; +} + Future _waitForThreadLifecycleStatus( AppController controller, String sessionKey,