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

merge: fix openclaw stop queue handling

Browse Source
This commit is contained in:
Haitao Pan 2026-05-25 08:49:23 +08:00
commit d3fae7bf2c
7 changed files with 300 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
lib/app/app_controller_desktop_thread_actions.dart
View File

@ -66,8 +66,10 @@ extension AppControllerDesktopThreadActions on AppController {
bool assistantSessionHasPendingRun(String sessionKey) {
final normalized = normalizedAssistantSessionKeyInternal(sessionKey);
return aiGatewayPendingSessionKeysInternal.contains(normalized) ||
(openClawGatewayQueuedTurnsBySessionInternal[normalized]?.isNotEmpty ??
false) ||
openClawGatewayQueuedTurnsBySessionInternal[normalized]?.any(
(turn) => !turn.cancelled,
) ==
true ||
(multiAgentRunPendingInternal &&
matchesSessionKey(
normalized,
@ -638,33 +640,64 @@ extension AppControllerDesktopThreadActions on AppController {
final normalizedSessionKey = normalizedAssistantSessionKeyInternal(
sessionKey,
);
final queuedForSession =
openClawGatewayQueuedTurnsBySessionInternal[normalizedSessionKey];
if (!removeQueuedOpenClawGatewayTurnsForSessionInternal(
normalizedSessionKey,
)) {
return false;
}
markOpenClawGatewayTurnAbortedInternal(normalizedSessionKey);
drainOpenClawGatewayQueueInternal();
return true;
}
bool removeQueuedOpenClawGatewayTurnsForSessionInternal(String sessionKey) {
final normalizedSessionKey = normalizedAssistantSessionKeyInternal(
sessionKey,
);
final queuedForSession = openClawGatewayQueuedTurnsBySessionInternal.remove(
normalizedSessionKey,
);
if (queuedForSession == null || queuedForSession.isEmpty) {
return false;
}
final turn = queuedForSession.removeAt(0);
if (queuedForSession.isEmpty) {
openClawGatewayQueuedTurnsBySessionInternal.remove(normalizedSessionKey);
for (final turn in queuedForSession) {
turn.cancelled = true;
openClawGatewayQueuedTurnsInternal.remove(turn);
}
openClawGatewayQueuedTurnsInternal.remove(turn);
turn.cancelled = true;
return true;
}
void markOpenClawGatewayTurnAbortedInternal(String sessionKey) {
final normalizedSessionKey = normalizedAssistantSessionKeyInternal(
sessionKey,
);
clearAiGatewayStreamingTextInternal(normalizedSessionKey);
aiGatewayPendingSessionKeysInternal.remove(normalizedSessionKey);
final nowMs = DateTime.now().millisecondsSinceEpoch.toDouble();
upsertTaskThreadInternal(
normalizedSessionKey,
lifecycleStatus: 'ready',
lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
lastRunAtMs: nowMs,
lastResultCode: 'aborted',
lastRemoteWorkingDirectory: '',
lastArtifactSyncAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
lastArtifactSyncAtMs: nowMs,
lastArtifactSyncStatus: 'failed',
lastTaskArtifactRelativePaths: const <String>[],
updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
updatedAtMs: nowMs,
);
recomputeTasksInternal();
notifyIfActiveInternal();
drainOpenClawGatewayQueueInternal();
return true;
}
void removeOpenClawGatewayQueuedTurnIndexInternal(
OpenClawGatewayQueuedTurnInternal turn,
) {
final queuedForSession =
openClawGatewayQueuedTurnsBySessionInternal[turn.sessionKey];
queuedForSession?.remove(turn);
if (queuedForSession != null && queuedForSession.isEmpty) {
openClawGatewayQueuedTurnsBySessionInternal.remove(turn.sessionKey);
}
}
void drainOpenClawGatewayQueueInternal() {
@ -672,12 +705,7 @@ extension AppControllerDesktopThreadActions on AppController {
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);
}
removeOpenClawGatewayQueuedTurnIndexInternal(turn);
if (turn.cancelled) {
continue;
}
@ -1017,27 +1045,18 @@ extension AppControllerDesktopThreadActions on AppController {
return;
}
if (aiGatewayPendingSessionKeysInternal.contains(sessionKey)) {
await goTaskServiceClientInternal.cancelTask(
route: GoTaskServiceRoute.externalAcpSingle,
target: assistantExecutionTargetForSession(sessionKey),
sessionId: sessionKey,
threadId: sessionKey,
);
aiGatewayPendingSessionKeysInternal.remove(sessionKey);
clearAiGatewayStreamingTextInternal(sessionKey);
upsertTaskThreadInternal(
sessionKey,
lifecycleStatus: 'ready',
lastRunAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
lastResultCode: 'aborted',
lastRemoteWorkingDirectory: '',
lastArtifactSyncAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
lastArtifactSyncStatus: 'failed',
lastTaskArtifactRelativePaths: const <String>[],
updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
);
recomputeTasksInternal();
notifyIfActiveInternal();
try {
await goTaskServiceClientInternal.cancelTask(
route: GoTaskServiceRoute.externalAcpSingle,
target: assistantExecutionTargetForSession(sessionKey),
sessionId: sessionKey,
threadId: sessionKey,
);
} catch (_) {
// Best effort cancellation only. Local state must still leave pending.
}
removeQueuedOpenClawGatewayTurnsForSessionInternal(sessionKey);
markOpenClawGatewayTurnAbortedInternal(sessionKey);
return;
}
}

5
lib/features/assistant/assistant_page_state_closure.dart
View File

@ -177,6 +177,11 @@ extension AssistantPageStateClosureInternal on AssistantPageStateInternal {
unawaited(controller.abortRun());
}
: null,
onContinue: progressState.recoverable
? AssistantPageStateActionsInternal(
this,
).focusComposerInternal
: null,
),
ColoredBox(
color: palette.canvas,

37
lib/widgets/assistant_task_progress_bar.dart
View File

@ -8,6 +8,7 @@ enum AssistantTaskProgressPhase {
running,
syncingArtifacts,
interrupted,
stopped,
}
class AssistantTaskProgressState {
@ -31,6 +32,10 @@ class AssistantTaskProgressState {
bool get visible => phase != AssistantTaskProgressPhase.idle;
bool get interrupted => phase == AssistantTaskProgressPhase.interrupted;
bool get stopped => phase == AssistantTaskProgressPhase.stopped;
bool get recoverable =>
phase == AssistantTaskProgressPhase.interrupted ||
phase == AssistantTaskProgressPhase.stopped;
bool get running =>
phase == AssistantTaskProgressPhase.queued ||
phase == AssistantTaskProgressPhase.running ||
@ -38,10 +43,16 @@ class AssistantTaskProgressState {
}
class AssistantTaskProgressBar extends StatelessWidget {
const AssistantTaskProgressBar({super.key, required this.state, this.onStop});
const AssistantTaskProgressBar({
super.key,
required this.state,
this.onStop,
this.onContinue,
});
final AssistantTaskProgressState state;
final VoidCallback? onStop;
final VoidCallback? onContinue;
@override
Widget build(BuildContext context) {
@ -51,6 +62,8 @@ class AssistantTaskProgressBar extends StatelessWidget {
final theme = Theme.of(context);
final color = state.interrupted
? theme.colorScheme.error
: state.stopped
? theme.colorScheme.tertiary
: theme.colorScheme.primary;
return Container(
key: const Key('assistant-task-progress-bar'),
@ -59,6 +72,8 @@ class AssistantTaskProgressBar extends StatelessWidget {
decoration: BoxDecoration(
color: state.interrupted
? theme.colorScheme.errorContainer.withValues(alpha: 0.18)
: state.stopped
? theme.colorScheme.tertiaryContainer.withValues(alpha: 0.22)
: theme.colorScheme.primaryContainer.withValues(alpha: 0.18),
border: Border(
top: BorderSide(color: theme.dividerColor.withValues(alpha: 0.42)),
@ -104,6 +119,16 @@ class AssistantTaskProgressBar extends StatelessWidget {
onPressed: onStop,
),
],
if (state.recoverable && onContinue != null) ...[
const SizedBox(width: 8),
_AssistantTaskProgressActionButton(
key: const Key('assistant-task-progress-continue-button'),
icon: Icons.play_arrow_rounded,
label: appText('继续', 'Continue'),
color: color,
onPressed: onContinue,
),
],
],
),
);
@ -184,6 +209,16 @@ AssistantTaskProgressState assistantTaskProgressState({
value: 0.48,
);
}
if (result == 'ABORTED') {
return AssistantTaskProgressState(
phase: AssistantTaskProgressPhase.stopped,
label: appText(
'任务已停止，可继续补充需求恢复执行。',
'Task stopped. Continue by adding the next request.',
),
value: 0,
);
}
return const AssistantTaskProgressState.idle();
}

3
lib/widgets/sidebar_navigation_task_section.dart
View File

@ -676,7 +676,8 @@ StatusInfo? _sidebarTaskStatusInfo(SidebarTaskItem item) {
final lifecycleStatus = item.lifecycleStatus.trim().toLowerCase();
final lastResultCode = item.lastResultCode.trim();
final normalizedResultCode = lastResultCode.toLowerCase();
if (lifecycleStatus == 'queued' || normalizedResultCode == 'queued') {
if (item.pending &&
(lifecycleStatus == 'queued' || normalizedResultCode == 'queued')) {
return StatusInfo(appText('Pending', 'Pending'), StatusTone.warning);
}
if (item.pending ||

29
test/features/app/sidebar_navigation_task_status_test.dart
View File

@ -35,7 +35,7 @@ void main() {
updatedAtMs: 1,
executionTarget: AssistantExecutionTarget.gateway,
isCurrent: false,
pending: false,
pending: true,
lifecycleStatus: 'queued',
lastResultCode: 'queued',
),
@ -90,6 +90,33 @@ void main() {
findsNothing,
);
});
testWidgets('sidebar does not show pending for stale queued lifecycle', (
tester,
) async {
await _pumpSidebar(
tester,
items: const <SidebarTaskItem>[
SidebarTaskItem(
sessionKey: 'stale-queued-task',
title: '已停止任务',
preview: '不应继续 Pending',
updatedAtMs: 1,
executionTarget: AssistantExecutionTarget.gateway,
isCurrent: false,
pending: false,
lifecycleStatus: 'queued',
lastResultCode: 'queued',
),
],
);
expect(find.text('Pending'), findsNothing);
expect(
find.byKey(const Key('workspace-sidebar-task-status-chip')),
findsNothing,
);
});
}
Future<void> _pumpSidebar(

60
test/features/assistant/assistant_task_progress_bar_test.dart
View File

@ -129,6 +129,54 @@ void main() {
expect(indicator.value, 0.48);
});
testWidgets('shows continue action for a stopped task', (tester) async {
var continued = false;
await tester.pumpWidget(
_buildTestApp(
assistantTaskProgressState(
pending: false,
lifecycleStatus: 'ready',
lastResultCode: 'aborted',
artifactSyncStatus: 'failed',
),
onContinue: () {
continued = true;
},
),
);
expect(find.text('任务已停止，可继续补充需求恢复执行。'), findsOneWidget);
expect(
find.byKey(const Key('assistant-task-progress-stop-button')),
findsNothing,
);
await tester.tap(
find.byKey(const Key('assistant-task-progress-continue-button')),
);
expect(continued, isTrue);
});
testWidgets('hides continue action for a stopped task without handler', (
tester,
) async {
await tester.pumpWidget(
_buildTestApp(
assistantTaskProgressState(
pending: false,
lifecycleStatus: 'ready',
lastResultCode: 'aborted',
artifactSyncStatus: 'failed',
),
),
);
expect(find.text('任务已停止，可继续补充需求恢复执行。'), findsOneWidget);
expect(
find.byKey(const Key('assistant-task-progress-continue-button')),
findsNothing,
);
});
testWidgets('hides idle progress state', (tester) async {
await tester.pumpWidget(
_buildTestApp(const AssistantTaskProgressState.idle()),
@ -182,13 +230,21 @@ void main() {
});
}
Widget _buildTestApp(AssistantTaskProgressState state, {VoidCallback? onStop}) {
Widget _buildTestApp(
AssistantTaskProgressState state, {
VoidCallback? onStop,
VoidCallback? onContinue,
}) {
return MaterialApp(
theme: AppTheme.light(),
home: Material(
child: SizedBox(
width: 420,
child: AssistantTaskProgressBar(state: state, onStop: onStop),
child: AssistantTaskProgressBar(
state: state,
onStop: onStop,
onContinue: onContinue,
),
),
),
);

111
test/runtime/assistant_execution_target_test.dart
View File

@ -2646,6 +2646,117 @@ void main() {
},
);
test(
'abortRun stops the current running OpenClaw task without clearing other queued tasks',
() async {
final fakeGoTaskService = _BlockingGoTaskServiceClient();
final controller = _connectedGatewayController(fakeGoTaskService);
addTearDown(() {
fakeGoTaskService.completeAll();
controller.dispose();
});
await _selectGatewaySession(controller, 'running-openclaw-stop-task');
final runningFuture = controller.sendChatMessage('running');
await fakeGoTaskService.waitForRequestCount(1);
await _selectGatewaySession(controller, 'queued-openclaw-after-stop');
final queuedFuture = controller.sendChatMessage('queued');
await _waitForThreadLifecycleStatus(
controller,
'queued-openclaw-after-stop',
'queued',
);
await _selectGatewaySession(controller, 'running-openclaw-stop-task');
await controller.abortRun();
expect(fakeGoTaskService.cancelledSessionIds, <String>[
'running-openclaw-stop-task',
]);
expect(
controller.assistantSessionHasPendingRun(
'running-openclaw-stop-task',
),
isFalse,
);
expect(
controller
.requireTaskThreadForSessionInternal('running-openclaw-stop-task')
.lifecycleState
.lastResultCode,
'aborted',
);
expect(
controller.assistantSessionHasPendingRun(
'queued-openclaw-after-stop',
),
isTrue,
);
fakeGoTaskService.complete(
'running-openclaw-stop-task',
const GoTaskServiceResult(
success: true,
message: 'late stopped result',
turnId: 'turn-stopped',
raw: <String, dynamic>{},
errorMessage: '',
resolvedModel: '',
route: GoTaskServiceRoute.externalAcpSingle,
),
);
await runningFuture;
await fakeGoTaskService.waitForRequestCount(2);
expect(
fakeGoTaskService.requests.last.sessionId,
'queued-openclaw-after-stop',
);
fakeGoTaskService.complete(
'queued-openclaw-after-stop',
const GoTaskServiceResult(
success: true,
message: 'queued done',
turnId: 'turn-queued',
raw: <String, dynamic>{},
errorMessage: '',
resolvedModel: '',
route: GoTaskServiceRoute.externalAcpSingle,
),
);
await queuedFuture;
await _waitForThreadLifecycleStatus(
controller,
'queued-openclaw-after-stop',
'ready',
);
expect(fakeGoTaskService.requests, hasLength(2));
},
);
test(
'stale queued lifecycle without a real queue entry is not pending',
() {
final controller = _connectedGatewayController(
_BlockingGoTaskServiceClient(),
);
addTearDown(controller.dispose);
controller.upsertTaskThreadInternal(
'stale-queued-task',
lifecycleStatus: 'queued',
lastResultCode: 'queued',
updatedAtMs: DateTime.now().millisecondsSinceEpoch.toDouble(),
);
expect(
controller.assistantSessionHasPendingRun('stale-queued-task'),
isFalse,
);
},
);
test('OpenClaw queue overflow fails without artifact sync', () async {
final fakeGoTaskService = _BlockingGoTaskServiceClient();
final controller = _connectedGatewayController(fakeGoTaskService);