Reassociate OpenClaw tasks through Bridge control plane
This commit is contained in:
Haitao Pan
parent
177fdcb46a
commit
55a1ce2af4
@ -76,6 +76,8 @@ extension AppControllerDesktopSkillPermissions on AppController {
|
||||
double? lastArtifactSyncAtMs,
|
||||
String? lastArtifactSyncStatus,
|
||||
List<String>? lastTaskArtifactRelativePaths,
|
||||
OpenClawTaskAssociation? openClawTaskAssociation,
|
||||
bool clearOpenClawTaskAssociation = false,
|
||||
}) {
|
||||
final normalizedSessionKey = normalizedAssistantSessionKeyInternal(
|
||||
sessionKey,
|
||||
@ -225,6 +227,8 @@ extension AppControllerDesktopSkillPermissions on AppController {
|
||||
lastArtifactSyncAtMs: lastArtifactSyncAtMs,
|
||||
lastArtifactSyncStatus: lastArtifactSyncStatus,
|
||||
lastTaskArtifactRelativePaths: lastTaskArtifactRelativePaths,
|
||||
openClawTaskAssociation: openClawTaskAssociation,
|
||||
clearOpenClawTaskAssociation: clearOpenClawTaskAssociation,
|
||||
);
|
||||
final nextStatus =
|
||||
lifecycleStatus ??
|
||||
@ -263,6 +267,7 @@ extension AppControllerDesktopSkillPermissions on AppController {
|
||||
executionBinding: nextExecutionBinding,
|
||||
contextState: nextContextState,
|
||||
lifecycleState: nextLifecycleState,
|
||||
openClawTaskAssociation: nextContextState.openClawTaskAssociation,
|
||||
updatedAtMs:
|
||||
updatedAtMs ??
|
||||
existing?.updatedAtMs ??
|
||||
|
||||
@ -3,6 +3,7 @@
|
||||
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';
|
||||
@ -64,7 +65,11 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
|
||||
bool assistantSessionHasPendingRun(String sessionKey) {
|
||||
final normalized = normalizedAssistantSessionKeyInternal(sessionKey);
|
||||
final association = taskThreadForSessionInternal(
|
||||
normalized,
|
||||
)?.openClawTaskAssociation;
|
||||
return aiGatewayPendingSessionKeysInternal.contains(normalized) ||
|
||||
(association != null && !association.isTerminal) ||
|
||||
openClawGatewayQueuedTurnsBySessionInternal[normalized]?.any(
|
||||
(turn) => !turn.cancelled,
|
||||
) ==
|
||||
@ -558,6 +563,7 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
appendGatewayUserTurnInternal(sessionKey, message);
|
||||
}
|
||||
markGatewayChatRunInternal(sessionKey);
|
||||
var handedOffToBridgeTask = false;
|
||||
try {
|
||||
final result = await goTaskServiceClientInternal.executeTask(
|
||||
GoTaskServiceRequest(
|
||||
@ -590,6 +596,22 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
clearAiGatewayStreamingTextInternal(sessionKey);
|
||||
return;
|
||||
}
|
||||
final association = result.openClawTaskAssociation;
|
||||
if (association != null) {
|
||||
handedOffToBridgeTask = true;
|
||||
persistOpenClawTaskAssociationInternal(
|
||||
sessionKey: sessionKey,
|
||||
association: association,
|
||||
);
|
||||
unawaited(
|
||||
pollOpenClawTaskAssociationInternal(
|
||||
sessionKey: sessionKey,
|
||||
target: target,
|
||||
association: association,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
await applyGatewayChatResultInternal(
|
||||
sessionKey: sessionKey,
|
||||
target: target,
|
||||
@ -610,13 +632,155 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
error: error,
|
||||
);
|
||||
} finally {
|
||||
aiGatewayPendingSessionKeysInternal.remove(sessionKey);
|
||||
clearAiGatewayStreamingTextInternal(sessionKey);
|
||||
if (!handedOffToBridgeTask) {
|
||||
aiGatewayPendingSessionKeysInternal.remove(sessionKey);
|
||||
clearAiGatewayStreamingTextInternal(sessionKey);
|
||||
}
|
||||
recomputeTasksInternal();
|
||||
notifyIfActiveInternal();
|
||||
}
|
||||
}
|
||||
|
||||
void persistOpenClawTaskAssociationInternal({
|
||||
required String sessionKey,
|
||||
required OpenClawTaskAssociation association,
|
||||
}) {
|
||||
final nowMs = DateTime.now().millisecondsSinceEpoch.toDouble();
|
||||
aiGatewayPendingSessionKeysInternal.add(sessionKey);
|
||||
upsertTaskThreadInternal(
|
||||
sessionKey,
|
||||
lifecycleStatus: 'running',
|
||||
lastResultCode: 'running',
|
||||
lastRunAtMs: association.startedAtMs > 0
|
||||
? association.startedAtMs
|
||||
: nowMs,
|
||||
lastArtifactSyncAtMs: nowMs,
|
||||
lastArtifactSyncStatus: 'running',
|
||||
openClawTaskAssociation: association,
|
||||
updatedAtMs: nowMs,
|
||||
);
|
||||
recomputeTasksInternal();
|
||||
notifyIfActiveInternal();
|
||||
unawaited(flushAssistantThreadPersistenceInternal());
|
||||
}
|
||||
|
||||
Future<void> pollOpenClawTaskAssociationInternal({
|
||||
required String sessionKey,
|
||||
required AssistantExecutionTarget target,
|
||||
required OpenClawTaskAssociation association,
|
||||
}) async {
|
||||
var current = association;
|
||||
final pollDelay = _openClawAssociationPollDelayInternal(current);
|
||||
final maxAttempts = math.max(
|
||||
1,
|
||||
((math.max(1, current.runtimeBudgetMinutes) * 60) /
|
||||
math.max(1, pollDelay.inSeconds))
|
||||
.ceil(),
|
||||
);
|
||||
for (var attempt = 0; attempt < maxAttempts; attempt += 1) {
|
||||
if (disposedInternal) {
|
||||
return;
|
||||
}
|
||||
if (!aiGatewayPendingSessionKeysInternal.contains(sessionKey)) {
|
||||
return;
|
||||
}
|
||||
if (attempt > 0) {
|
||||
await Future<void>.delayed(pollDelay);
|
||||
}
|
||||
try {
|
||||
final result = await goTaskServiceClientInternal.getTask(
|
||||
route: GoTaskServiceRoute.externalAcpSingle,
|
||||
target: target,
|
||||
association: current,
|
||||
);
|
||||
final nextAssociation =
|
||||
result.openClawTaskAssociation ??
|
||||
current.copyWith(
|
||||
status: result.status.trim().isEmpty
|
||||
? current.status
|
||||
: result.status.trim(),
|
||||
);
|
||||
current = nextAssociation;
|
||||
if (result.isOpenClawRunningTaskHandle) {
|
||||
persistOpenClawTaskAssociationInternal(
|
||||
sessionKey: sessionKey,
|
||||
association: nextAssociation,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
if (aiGatewayPendingSessionKeysInternal.contains(sessionKey)) {
|
||||
await applyGatewayChatResultInternal(
|
||||
sessionKey: sessionKey,
|
||||
target: target,
|
||||
result: result,
|
||||
);
|
||||
}
|
||||
aiGatewayPendingSessionKeysInternal.remove(sessionKey);
|
||||
clearAiGatewayStreamingTextInternal(sessionKey);
|
||||
recomputeTasksInternal();
|
||||
notifyIfActiveInternal();
|
||||
return;
|
||||
} catch (_) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
final nowMs = DateTime.now().millisecondsSinceEpoch.toDouble();
|
||||
upsertTaskThreadInternal(
|
||||
sessionKey,
|
||||
lifecycleStatus: 'ready',
|
||||
lastRunAtMs: nowMs,
|
||||
lastResultCode: 'TASK_SLA_EXPIRED',
|
||||
lastArtifactSyncAtMs: nowMs,
|
||||
lastArtifactSyncStatus: 'failed',
|
||||
openClawTaskAssociation: current.copyWith(status: 'failed'),
|
||||
updatedAtMs: nowMs,
|
||||
);
|
||||
aiGatewayPendingSessionKeysInternal.remove(sessionKey);
|
||||
clearAiGatewayStreamingTextInternal(sessionKey);
|
||||
recomputeTasksInternal();
|
||||
notifyIfActiveInternal();
|
||||
}
|
||||
|
||||
Duration _openClawAssociationPollDelayInternal(
|
||||
OpenClawTaskAssociation association,
|
||||
) {
|
||||
final budget = association.runtimeBudgetMinutes;
|
||||
if (budget >= 60) {
|
||||
return const Duration(seconds: 5);
|
||||
}
|
||||
if (budget >= 30) {
|
||||
return const Duration(seconds: 3);
|
||||
}
|
||||
return const Duration(seconds: 2);
|
||||
}
|
||||
|
||||
void resumeOpenClawTaskAssociationsInternal({String? onlySessionKey}) {
|
||||
final normalizedOnly = onlySessionKey == null
|
||||
? ''
|
||||
: normalizedAssistantSessionKeyInternal(onlySessionKey);
|
||||
for (final record in taskThreadRepositoryInternal.snapshot()) {
|
||||
if (normalizedOnly.isNotEmpty && record.threadId != normalizedOnly) {
|
||||
continue;
|
||||
}
|
||||
final association = record.openClawTaskAssociation;
|
||||
if (association == null || association.isTerminal) {
|
||||
continue;
|
||||
}
|
||||
aiGatewayPendingSessionKeysInternal.add(record.threadId);
|
||||
unawaited(
|
||||
pollOpenClawTaskAssociationInternal(
|
||||
sessionKey: record.threadId,
|
||||
target: assistantExecutionTargetFromExecutionMode(
|
||||
record.executionBinding.executionMode,
|
||||
),
|
||||
association: association,
|
||||
),
|
||||
);
|
||||
}
|
||||
recomputeTasksInternal();
|
||||
notifyIfActiveInternal();
|
||||
}
|
||||
|
||||
String messageWithSelectedSkillsContextInternal({
|
||||
required String message,
|
||||
required List<String> selectedSkillLabels,
|
||||
@ -844,6 +1008,7 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
lastArtifactSyncAtMs: nowMs,
|
||||
lastArtifactSyncStatus: 'failed',
|
||||
lastTaskArtifactRelativePaths: const <String>[],
|
||||
clearOpenClawTaskAssociation: true,
|
||||
updatedAtMs: nowMs,
|
||||
);
|
||||
recomputeTasksInternal();
|
||||
@ -1017,6 +1182,7 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
lifecycleStatus: 'ready',
|
||||
lastRunAtMs: completedAtMs,
|
||||
lastResultCode: terminalResultCode,
|
||||
clearOpenClawTaskAssociation: true,
|
||||
updatedAtMs: completedAtMs,
|
||||
);
|
||||
if (isOpenClawNoExportedArtifactsGuardResultInternal(result)) {
|
||||
@ -1102,6 +1268,7 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
lastArtifactSyncAtMs: completedAtMs,
|
||||
lastArtifactSyncStatus: 'failed',
|
||||
lastTaskArtifactRelativePaths: const <String>[],
|
||||
clearOpenClawTaskAssociation: true,
|
||||
updatedAtMs: completedAtMs,
|
||||
);
|
||||
appendLocalSessionMessageInternal(
|
||||
@ -1254,6 +1421,9 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
target: assistantExecutionTargetForSession(sessionKey),
|
||||
sessionId: sessionKey,
|
||||
threadId: sessionKey,
|
||||
association: taskThreadForSessionInternal(
|
||||
sessionKey,
|
||||
)?.openClawTaskAssociation,
|
||||
);
|
||||
} catch (_) {
|
||||
// Best effort cancellation only.
|
||||
@ -1287,6 +1457,9 @@ extension AppControllerDesktopThreadActions on AppController {
|
||||
target: assistantExecutionTargetForSession(sessionKey),
|
||||
sessionId: sessionKey,
|
||||
threadId: sessionKey,
|
||||
association: taskThreadForSessionInternal(
|
||||
sessionKey,
|
||||
)?.openClawTaskAssociation,
|
||||
);
|
||||
} catch (_) {
|
||||
// Best effort cancellation only. Local state must still leave pending.
|
||||
|
||||
@ -145,6 +145,7 @@ extension AppControllerDesktopThreadStorage on AppController {
|
||||
normalized,
|
||||
persistSelection: false,
|
||||
);
|
||||
resumeOpenClawTaskAssociationsInternal(onlySessionKey: normalized);
|
||||
}
|
||||
|
||||
void handleRuntimeEventInternal(GatewayPushEvent event) {
|
||||
|
||||
@ -125,11 +125,7 @@ extension AssistantPageStateClosureInternal on AssistantPageStateInternal {
|
||||
lastResultCode: thread?.lifecycleState.lastResultCode ?? '',
|
||||
artifactSyncStatus: thread?.lastArtifactSyncStatus ?? '',
|
||||
runtimeBudgetMinutes:
|
||||
gatewayAcpTaskRuntimeBudgetMinutesForParams(<String, dynamic>{
|
||||
'taskPrompt': currentTask.preview,
|
||||
'requestedExecutionTarget':
|
||||
currentTask.executionTarget.promptValue,
|
||||
}),
|
||||
thread?.openClawTaskAssociation?.runtimeBudgetMinutes ?? 10,
|
||||
);
|
||||
|
||||
return SurfaceCard(
|
||||
|
||||
@ -1,6 +1,5 @@
|
||||
import 'dart:async';
|
||||
import 'dart:io';
|
||||
import 'dart:math' as math;
|
||||
|
||||
import 'package:flutter/foundation.dart';
|
||||
|
||||
@ -104,6 +103,7 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
var streamedText = '';
|
||||
String? completedMessage;
|
||||
Map<String, dynamic>? completedResultSnapshot;
|
||||
Map<String, dynamic>? runningTaskSnapshot;
|
||||
try {
|
||||
final endpointOverride = _taskEndpointResolver == null
|
||||
? _endpointResolver(request.target)
|
||||
@ -136,6 +136,11 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
update,
|
||||
);
|
||||
}
|
||||
if (update.payload['status']?.toString().trim().toLowerCase() ==
|
||||
'running' &&
|
||||
(update.payload['runId']?.toString().trim().isNotEmpty == true)) {
|
||||
runningTaskSnapshot = <String, dynamic>{...update.payload};
|
||||
}
|
||||
onUpdate(update);
|
||||
},
|
||||
);
|
||||
@ -155,6 +160,7 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
streamedText: streamedText,
|
||||
completedMessage: completedMessage,
|
||||
fallbackAvailable: completedResultSnapshot != null,
|
||||
runningTaskSnapshot: runningTaskSnapshot,
|
||||
);
|
||||
if (recovered != null) {
|
||||
return recovered;
|
||||
@ -203,11 +209,27 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
required String streamedText,
|
||||
required String? completedMessage,
|
||||
bool fallbackAvailable = false,
|
||||
Map<String, dynamic>? runningTaskSnapshot,
|
||||
}) async {
|
||||
final endpoint = _sessionSnapshotEndpoint(taskEndpoint);
|
||||
if (endpoint == null) {
|
||||
return null;
|
||||
}
|
||||
final association = OpenClawTaskAssociation.fromJsonOrNull(
|
||||
runningTaskSnapshot,
|
||||
);
|
||||
if (association != null) {
|
||||
return goTaskServiceResultFromAcpResponse(
|
||||
<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': 'recovered-from-running-task-handle',
|
||||
'result': runningTaskSnapshot,
|
||||
},
|
||||
route: request.route,
|
||||
streamedText: streamedText,
|
||||
completedMessage: completedMessage,
|
||||
);
|
||||
}
|
||||
final attempts = _recoveryAttemptsForRequest(request);
|
||||
for (var attempt = 0; attempt < attempts; attempt += 1) {
|
||||
if (attempt > 0) {
|
||||
@ -216,7 +238,7 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
Map<String, dynamic> response;
|
||||
try {
|
||||
response = await _client.request(
|
||||
method: 'xworkmate.sessions.get',
|
||||
method: 'xworkmate.tasks.get',
|
||||
params: <String, dynamic>{
|
||||
'sessionId': request.sessionId,
|
||||
'threadId': request.threadId,
|
||||
@ -232,11 +254,7 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
continue;
|
||||
}
|
||||
final snapshot = _castMap(response['result']);
|
||||
final task = _castMap(snapshot['task']);
|
||||
final status = (task['state'] ?? snapshot['status'] ?? '')
|
||||
.toString()
|
||||
.trim()
|
||||
.toLowerCase();
|
||||
final status = (snapshot['status'] ?? '').toString().trim().toLowerCase();
|
||||
final terminal =
|
||||
status == 'completed' ||
|
||||
status == 'failed' ||
|
||||
@ -245,7 +263,19 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
if (!terminal) {
|
||||
continue;
|
||||
}
|
||||
final result = _recoveredResultFromSessionSnapshot(snapshot);
|
||||
if (status == 'failed' || status == 'cancelled' || status == 'canceled') {
|
||||
return goTaskServiceResultFromAcpResponse(
|
||||
<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': 'recovered-from-terminal-task-snapshot',
|
||||
'result': _failureResultFromTaskSnapshot(snapshot, status),
|
||||
},
|
||||
route: request.route,
|
||||
streamedText: streamedText,
|
||||
completedMessage: completedMessage,
|
||||
);
|
||||
}
|
||||
final result = _recoveredResultFromTaskSnapshot(snapshot);
|
||||
if (result.isNotEmpty) {
|
||||
return goTaskServiceResultFromAcpResponse(
|
||||
<String, dynamic>{
|
||||
@ -258,34 +288,37 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
completedMessage: completedMessage,
|
||||
);
|
||||
}
|
||||
if (status == 'failed' || status == 'cancelled' || status == 'canceled') {
|
||||
return goTaskServiceResultFromAcpResponse(
|
||||
<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': 'recovered-from-terminal-session-snapshot',
|
||||
'result': _failureResultFromSessionSnapshot(snapshot, status),
|
||||
},
|
||||
route: request.route,
|
||||
streamedText: streamedText,
|
||||
completedMessage: completedMessage,
|
||||
);
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@override
|
||||
Future<GoTaskServiceResult> getTask({
|
||||
required AssistantExecutionTarget target,
|
||||
required OpenClawTaskAssociation association,
|
||||
required GoTaskServiceRoute route,
|
||||
}) async {
|
||||
final endpoint = _sessionSnapshotEndpoint(_endpointResolver(target));
|
||||
if (endpoint == null) {
|
||||
throw const GatewayAcpException(
|
||||
'xworkmate-bridge is not connected',
|
||||
code: 'BRIDGE_NOT_CONNECTED',
|
||||
);
|
||||
}
|
||||
final response = await _client.request(
|
||||
method: 'xworkmate.tasks.get',
|
||||
params: association.toTaskGetParams(),
|
||||
endpointOverride: endpoint,
|
||||
);
|
||||
return goTaskServiceResultFromAcpResponse(response, route: route);
|
||||
}
|
||||
|
||||
int _recoveryAttemptsForRequest(GoTaskServiceRequest request) {
|
||||
final configured = _recoveryMaxAttempts;
|
||||
if (configured != null) {
|
||||
return configured <= 0 ? 1 : configured;
|
||||
}
|
||||
final pollMicros = math.max(1, _recoveryPollDelay.inMicroseconds);
|
||||
final budgetMicros = Duration(
|
||||
minutes: gatewayAcpTaskRuntimeBudgetMinutesForParams(
|
||||
request.toExternalAcpParams(),
|
||||
),
|
||||
).inMicroseconds;
|
||||
return math.max(1, (budgetMicros / pollMicros).ceil());
|
||||
return 1;
|
||||
}
|
||||
|
||||
Uri? _sessionSnapshotEndpoint(Uri? taskEndpoint) {
|
||||
@ -303,7 +336,19 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
required AssistantExecutionTarget target,
|
||||
required String sessionId,
|
||||
required String threadId,
|
||||
OpenClawTaskAssociation? association,
|
||||
}) async {
|
||||
if (association != null) {
|
||||
final endpoint = _sessionSnapshotEndpoint(_endpointResolver(target));
|
||||
if (endpoint != null) {
|
||||
await _client.request(
|
||||
method: 'xworkmate.tasks.cancel',
|
||||
params: association.toTaskGetParams(),
|
||||
endpointOverride: endpoint,
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
await _client.cancelSession(
|
||||
sessionId: sessionId,
|
||||
threadId: threadId,
|
||||
@ -359,13 +404,16 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
Map<String, dynamic> _recoveredResultFromSessionSnapshot(
|
||||
Map<String, dynamic> _recoveredResultFromTaskSnapshot(
|
||||
Map<String, dynamic> snapshot,
|
||||
) {
|
||||
final result = <String, dynamic>{..._castMap(snapshot['result'])};
|
||||
final result = <String, dynamic>{
|
||||
..._castMap(snapshot['result']),
|
||||
...snapshot,
|
||||
};
|
||||
final artifactRecord = _castMap(snapshot['artifacts']);
|
||||
final artifactItems = _listValue(artifactRecord['items']);
|
||||
if (artifactItems.isNotEmpty && !_hasArtifactList(result)) {
|
||||
final artifactItems = artifactRecord['items'];
|
||||
if (artifactItems is List && result['artifacts'] == artifactRecord) {
|
||||
result['artifacts'] = artifactItems;
|
||||
}
|
||||
for (final entry in <String, String>{
|
||||
@ -382,31 +430,23 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
return result;
|
||||
}
|
||||
|
||||
Map<String, dynamic> _failureResultFromSessionSnapshot(
|
||||
Map<String, dynamic> _failureResultFromTaskSnapshot(
|
||||
Map<String, dynamic> snapshot,
|
||||
String status,
|
||||
) {
|
||||
final task = _castMap(snapshot['task']);
|
||||
final error = _castMap(snapshot['error']);
|
||||
final message = _firstNonEmptyDisplayText(
|
||||
<String, dynamic>{...error, ...snapshot, 'taskMessage': task['message']},
|
||||
const <String>[
|
||||
'message',
|
||||
'error',
|
||||
'errorMessage',
|
||||
'reason',
|
||||
'taskMessage',
|
||||
'code',
|
||||
],
|
||||
<String, dynamic>{...error, ...snapshot},
|
||||
const <String>['message', 'error', 'errorMessage', 'reason', 'code'],
|
||||
);
|
||||
final code = _firstNonEmptyDisplayText(
|
||||
<String, dynamic>{...error, ...snapshot, 'taskCode': task['code']},
|
||||
const <String>['code', 'errorCode', 'taskCode'],
|
||||
<String, dynamic>{...error, ...snapshot},
|
||||
const <String>['code', 'errorCode'],
|
||||
);
|
||||
final result = <String, dynamic>{
|
||||
'success': false,
|
||||
'status': status,
|
||||
'turnId': task['turnId']?.toString().trim() ?? '',
|
||||
'turnId': snapshot['turnId']?.toString().trim() ?? '',
|
||||
'error': message.isNotEmpty ? message : 'Bridge session ended: $status',
|
||||
'message': message.isNotEmpty ? message : 'Bridge session ended: $status',
|
||||
};
|
||||
@ -416,29 +456,6 @@ class ExternalCodeAgentAcpDesktopTransport
|
||||
return result;
|
||||
}
|
||||
|
||||
bool _hasArtifactList(Map<String, dynamic> result) {
|
||||
for (final key in const <String>['artifacts', 'files', 'attachments']) {
|
||||
if (_listValue(result[key]).isNotEmpty) {
|
||||
return true;
|
||||
}
|
||||
final recordItems = _listValue(_castMap(result[key])['items']);
|
||||
if (recordItems.isNotEmpty) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
for (final key in const <String>['payload', 'result', 'data']) {
|
||||
final nested = _castMap(result[key]);
|
||||
if (nested.isNotEmpty && _hasArtifactList(nested)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
List<Object?> _listValue(Object? value) {
|
||||
return value is List ? value : const <Object?>[];
|
||||
}
|
||||
|
||||
String _firstNonEmptyDisplayText(
|
||||
Map<String, dynamic> values,
|
||||
List<String> keys,
|
||||
|
||||
@ -1457,130 +1457,12 @@ class GatewayAcpRuntimeSessionClient implements GatewayRuntimeSessionClient {
|
||||
}
|
||||
}
|
||||
|
||||
bool _isOpenClawTaskSubmitMethod(String method) {
|
||||
final normalized = method.trim();
|
||||
return normalized == 'session.start' || normalized == 'session.message';
|
||||
}
|
||||
|
||||
Duration gatewayAcpHttpResponseTimeoutFor(
|
||||
Uri endpoint,
|
||||
String method, [
|
||||
Map<String, dynamic> params = const <String, dynamic>{},
|
||||
]) {
|
||||
if (!_isOpenClawTaskSubmitMethod(method)) {
|
||||
return const Duration(seconds: 120);
|
||||
}
|
||||
return Duration(minutes: gatewayAcpTaskRuntimeBudgetMinutesForParams(params));
|
||||
}
|
||||
|
||||
int gatewayAcpTaskRuntimeBudgetMinutesForParams(Map<String, dynamic> params) {
|
||||
if (_looksLikeLongArtifactTask(params)) {
|
||||
return 30;
|
||||
}
|
||||
if (_looksLikeGatewayTask(params)) {
|
||||
return 10;
|
||||
}
|
||||
return 2;
|
||||
}
|
||||
|
||||
bool _looksLikeGatewayTask(Map<String, dynamic> params) {
|
||||
final target = _paramText(params, const <String>[
|
||||
'requestedExecutionTarget',
|
||||
'executionTarget',
|
||||
]).toLowerCase();
|
||||
if (target == AssistantExecutionTarget.gateway.promptValue) {
|
||||
return true;
|
||||
}
|
||||
final providerText = _paramText(params, const <String>[
|
||||
'provider',
|
||||
'gatewayProvider',
|
||||
'preferredGatewayProviderId',
|
||||
]).toLowerCase();
|
||||
if (providerText.contains('openclaw')) {
|
||||
return true;
|
||||
}
|
||||
final routing = params['routing'];
|
||||
if (routing is Map) {
|
||||
final preferred = routing['preferredGatewayProviderId']
|
||||
?.toString()
|
||||
.trim()
|
||||
.toLowerCase();
|
||||
return preferred == kCanonicalGatewayProviderId ||
|
||||
preferred?.contains('openclaw') == true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
bool _looksLikeLongArtifactTask(Map<String, dynamic> params) {
|
||||
final prompt = _paramText(params, const <String>[
|
||||
'taskPrompt',
|
||||
'prompt',
|
||||
'message',
|
||||
]);
|
||||
final lower = prompt.toLowerCase();
|
||||
final attachments =
|
||||
_paramListLength(params['attachments']) +
|
||||
_paramListLength(params['inlineAttachments']);
|
||||
if (attachments >= 2 || prompt.length >= 1200) {
|
||||
return true;
|
||||
}
|
||||
const markers = <String>[
|
||||
'生成文件',
|
||||
'同步生成文件',
|
||||
'产物',
|
||||
'附件',
|
||||
'图片提示词',
|
||||
'完整调研ppt',
|
||||
'markdown格式',
|
||||
'输出markdown',
|
||||
'输出 完整',
|
||||
'ppt',
|
||||
'pptx',
|
||||
'powerpoint',
|
||||
'xls',
|
||||
'.xls',
|
||||
'word格式',
|
||||
'word 格式',
|
||||
'word',
|
||||
'docx',
|
||||
'.docx',
|
||||
'png',
|
||||
'.png',
|
||||
'mp4',
|
||||
'.mp4',
|
||||
'jpg',
|
||||
'.jpg',
|
||||
'markdown格式',
|
||||
'markdown',
|
||||
'.md',
|
||||
'javascript',
|
||||
'.js',
|
||||
'image prompt',
|
||||
'artifacts',
|
||||
'downloadurl',
|
||||
];
|
||||
return markers.any(lower.contains);
|
||||
}
|
||||
|
||||
String _paramText(Map<String, dynamic> params, List<String> keys) {
|
||||
for (final key in keys) {
|
||||
final value = params[key];
|
||||
if (value == null) {
|
||||
continue;
|
||||
}
|
||||
final text = value.toString().trim();
|
||||
if (text.isNotEmpty) {
|
||||
return text;
|
||||
}
|
||||
}
|
||||
return '';
|
||||
}
|
||||
|
||||
int _paramListLength(Object? value) {
|
||||
if (value is List) {
|
||||
return value.length;
|
||||
}
|
||||
return 0;
|
||||
return const Duration(seconds: 120);
|
||||
}
|
||||
|
||||
class _GatewayAcpRpcRequest {
|
||||
|
||||
@ -494,6 +494,27 @@ class GoTaskServiceResult {
|
||||
|
||||
String get code => raw['code']?.toString().trim() ?? '';
|
||||
|
||||
bool get isOpenClawRunningTaskHandle {
|
||||
final normalizedStatus = status.trim().toLowerCase();
|
||||
final runId = raw['runId']?.toString().trim() ?? '';
|
||||
final artifactScope = raw['artifactScope']?.toString().trim() ?? '';
|
||||
final provider =
|
||||
raw['resolvedGatewayProviderId']?.toString().trim().toLowerCase() ??
|
||||
raw['gatewayProviderId']?.toString().trim().toLowerCase() ??
|
||||
'';
|
||||
return normalizedStatus == 'running' &&
|
||||
runId.isNotEmpty &&
|
||||
artifactScope.isNotEmpty &&
|
||||
provider.contains('openclaw');
|
||||
}
|
||||
|
||||
OpenClawTaskAssociation? get openClawTaskAssociation {
|
||||
if (!isOpenClawRunningTaskHandle) {
|
||||
return null;
|
||||
}
|
||||
return OpenClawTaskAssociation.fromJsonOrNull(raw);
|
||||
}
|
||||
|
||||
String get resolvedExecutionTarget =>
|
||||
raw['resolvedExecutionTarget']?.toString().trim() ?? '';
|
||||
|
||||
@ -651,10 +672,17 @@ abstract class ExternalCodeAgentAcpTransport {
|
||||
required void Function(GoTaskServiceUpdate update) onUpdate,
|
||||
});
|
||||
|
||||
Future<GoTaskServiceResult> getTask({
|
||||
required AssistantExecutionTarget target,
|
||||
required OpenClawTaskAssociation association,
|
||||
required GoTaskServiceRoute route,
|
||||
});
|
||||
|
||||
Future<void> cancelTask({
|
||||
required AssistantExecutionTarget target,
|
||||
required String sessionId,
|
||||
required String threadId,
|
||||
OpenClawTaskAssociation? association,
|
||||
});
|
||||
|
||||
Future<void> closeTask({
|
||||
@ -683,11 +711,18 @@ abstract class GoTaskServiceClient {
|
||||
required void Function(GoTaskServiceUpdate update) onUpdate,
|
||||
});
|
||||
|
||||
Future<GoTaskServiceResult> getTask({
|
||||
required AssistantExecutionTarget target,
|
||||
required OpenClawTaskAssociation association,
|
||||
required GoTaskServiceRoute route,
|
||||
});
|
||||
|
||||
Future<void> cancelTask({
|
||||
required GoTaskServiceRoute route,
|
||||
required AssistantExecutionTarget target,
|
||||
required String sessionId,
|
||||
required String threadId,
|
||||
OpenClawTaskAssociation? association,
|
||||
});
|
||||
|
||||
Future<void> closeTask({
|
||||
|
||||
@ -37,16 +37,29 @@ class DesktopGoTaskService implements GoTaskServiceClient {
|
||||
required void Function(GoTaskServiceUpdate update) onUpdate,
|
||||
}) => _acpTransport.executeTask(request, onUpdate: onUpdate);
|
||||
|
||||
@override
|
||||
Future<GoTaskServiceResult> getTask({
|
||||
required AssistantExecutionTarget target,
|
||||
required OpenClawTaskAssociation association,
|
||||
required GoTaskServiceRoute route,
|
||||
}) => _acpTransport.getTask(
|
||||
target: target,
|
||||
association: association,
|
||||
route: route,
|
||||
);
|
||||
|
||||
@override
|
||||
Future<void> cancelTask({
|
||||
required GoTaskServiceRoute route,
|
||||
required AssistantExecutionTarget target,
|
||||
required String sessionId,
|
||||
required String threadId,
|
||||
OpenClawTaskAssociation? association,
|
||||
}) => _acpTransport.cancelTask(
|
||||
target: target,
|
||||
sessionId: sessionId,
|
||||
threadId: threadId,
|
||||
association: association,
|
||||
);
|
||||
|
||||
@override
|
||||
|
||||
@ -677,6 +677,7 @@ class ThreadContextState {
|
||||
this.lastArtifactSyncAtMs,
|
||||
this.lastArtifactSyncStatus,
|
||||
this.lastTaskArtifactRelativePaths = const <String>[],
|
||||
this.openClawTaskAssociation,
|
||||
});
|
||||
|
||||
final List<GatewayChatMessage> messages;
|
||||
@ -694,6 +695,7 @@ class ThreadContextState {
|
||||
final double? lastArtifactSyncAtMs;
|
||||
final String? lastArtifactSyncStatus;
|
||||
final List<String> lastTaskArtifactRelativePaths;
|
||||
final OpenClawTaskAssociation? openClawTaskAssociation;
|
||||
|
||||
ThreadContextState copyWith({
|
||||
List<GatewayChatMessage>? messages,
|
||||
@ -712,6 +714,8 @@ class ThreadContextState {
|
||||
double? lastArtifactSyncAtMs,
|
||||
String? lastArtifactSyncStatus,
|
||||
List<String>? lastTaskArtifactRelativePaths,
|
||||
OpenClawTaskAssociation? openClawTaskAssociation,
|
||||
bool clearOpenClawTaskAssociation = false,
|
||||
}) {
|
||||
return ThreadContextState(
|
||||
messages: messages ?? this.messages,
|
||||
@ -738,6 +742,9 @@ class ThreadContextState {
|
||||
lastTaskArtifactRelativePaths: lastTaskArtifactRelativePaths == null
|
||||
? this.lastTaskArtifactRelativePaths
|
||||
: _stringListFromJson(lastTaskArtifactRelativePaths),
|
||||
openClawTaskAssociation: clearOpenClawTaskAssociation
|
||||
? null
|
||||
: (openClawTaskAssociation ?? this.openClawTaskAssociation),
|
||||
);
|
||||
}
|
||||
|
||||
@ -758,6 +765,7 @@ class ThreadContextState {
|
||||
'lastArtifactSyncAtMs': lastArtifactSyncAtMs,
|
||||
'lastArtifactSyncStatus': lastArtifactSyncStatus,
|
||||
'lastTaskArtifactRelativePaths': lastTaskArtifactRelativePaths,
|
||||
'openClawTaskAssociation': openClawTaskAssociation?.toJson(),
|
||||
};
|
||||
}
|
||||
|
||||
@ -823,6 +831,163 @@ class ThreadContextState {
|
||||
lastTaskArtifactRelativePaths: _stringListFromJson(
|
||||
json['lastTaskArtifactRelativePaths'],
|
||||
),
|
||||
openClawTaskAssociation: OpenClawTaskAssociation.fromJsonOrNull(
|
||||
json['openClawTaskAssociation'],
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
class OpenClawTaskAssociation {
|
||||
const OpenClawTaskAssociation({
|
||||
required this.sessionId,
|
||||
required this.threadId,
|
||||
required this.turnId,
|
||||
required this.runId,
|
||||
required this.artifactScope,
|
||||
required this.artifactDirectory,
|
||||
required this.gatewayProviderId,
|
||||
required this.runtimeBudgetMinutes,
|
||||
required this.startedAtMs,
|
||||
required this.status,
|
||||
this.taskLoadClass = '',
|
||||
this.sessionKey = '',
|
||||
this.requiredArtifactExtensions = const <String>[],
|
||||
this.expectedArtifactExtensions = const <String>[],
|
||||
});
|
||||
|
||||
final String sessionId;
|
||||
final String threadId;
|
||||
final String turnId;
|
||||
final String runId;
|
||||
final String artifactScope;
|
||||
final String artifactDirectory;
|
||||
final String gatewayProviderId;
|
||||
final int runtimeBudgetMinutes;
|
||||
final double startedAtMs;
|
||||
final String status;
|
||||
final String taskLoadClass;
|
||||
final String sessionKey;
|
||||
final List<String> requiredArtifactExtensions;
|
||||
final List<String> expectedArtifactExtensions;
|
||||
|
||||
bool get isTerminal {
|
||||
final normalized = status.trim().toLowerCase();
|
||||
return normalized == 'completed' ||
|
||||
normalized == 'failed' ||
|
||||
normalized == 'cancelled' ||
|
||||
normalized == 'canceled';
|
||||
}
|
||||
|
||||
OpenClawTaskAssociation copyWith({String? status}) {
|
||||
return OpenClawTaskAssociation(
|
||||
sessionId: sessionId,
|
||||
threadId: threadId,
|
||||
turnId: turnId,
|
||||
runId: runId,
|
||||
artifactScope: artifactScope,
|
||||
artifactDirectory: artifactDirectory,
|
||||
gatewayProviderId: gatewayProviderId,
|
||||
runtimeBudgetMinutes: runtimeBudgetMinutes,
|
||||
startedAtMs: startedAtMs,
|
||||
status: status ?? this.status,
|
||||
taskLoadClass: taskLoadClass,
|
||||
sessionKey: sessionKey,
|
||||
requiredArtifactExtensions: requiredArtifactExtensions,
|
||||
expectedArtifactExtensions: expectedArtifactExtensions,
|
||||
);
|
||||
}
|
||||
|
||||
Map<String, dynamic> toJson() {
|
||||
return <String, dynamic>{
|
||||
'sessionId': sessionId,
|
||||
'threadId': threadId,
|
||||
'turnId': turnId,
|
||||
'runId': runId,
|
||||
'artifactScope': artifactScope,
|
||||
'artifactDirectory': artifactDirectory,
|
||||
'gatewayProviderId': gatewayProviderId,
|
||||
'runtimeBudgetMinutes': runtimeBudgetMinutes,
|
||||
'startedAtMs': startedAtMs,
|
||||
'status': status,
|
||||
'taskLoadClass': taskLoadClass,
|
||||
'sessionKey': sessionKey,
|
||||
'requiredArtifactExtensions': requiredArtifactExtensions,
|
||||
'expectedArtifactExtensions': expectedArtifactExtensions,
|
||||
};
|
||||
}
|
||||
|
||||
Map<String, dynamic> toTaskGetParams() {
|
||||
return <String, dynamic>{
|
||||
'sessionId': sessionId,
|
||||
'threadId': threadId,
|
||||
'turnId': turnId,
|
||||
'runId': runId,
|
||||
'artifactScope': artifactScope,
|
||||
'artifactDirectory': artifactDirectory,
|
||||
'gatewayProviderId': gatewayProviderId,
|
||||
'runtimeBudgetMinutes': runtimeBudgetMinutes,
|
||||
'taskLoadClass': taskLoadClass,
|
||||
'sessionKey': sessionKey,
|
||||
'requiredArtifactExtensions': requiredArtifactExtensions,
|
||||
'expectedArtifactExtensions': expectedArtifactExtensions,
|
||||
};
|
||||
}
|
||||
|
||||
static OpenClawTaskAssociation? fromJsonOrNull(Object? value) {
|
||||
if (value is! Map) {
|
||||
return null;
|
||||
}
|
||||
final json = value.cast<String, dynamic>();
|
||||
final runId = json['runId']?.toString().trim() ?? '';
|
||||
final artifactScope = json['artifactScope']?.toString().trim() ?? '';
|
||||
if (runId.isEmpty || artifactScope.isEmpty) {
|
||||
return null;
|
||||
}
|
||||
int asInt(Object? raw) {
|
||||
if (raw is int) {
|
||||
return raw;
|
||||
}
|
||||
if (raw is num) {
|
||||
return raw.toInt();
|
||||
}
|
||||
return int.tryParse(raw?.toString() ?? '') ?? 60;
|
||||
}
|
||||
|
||||
double asDouble(Object? raw) {
|
||||
if (raw is num) {
|
||||
return raw.toDouble();
|
||||
}
|
||||
return double.tryParse(raw?.toString() ?? '') ?? 0;
|
||||
}
|
||||
|
||||
return OpenClawTaskAssociation(
|
||||
sessionId: json['sessionId']?.toString().trim() ?? '',
|
||||
threadId: json['threadId']?.toString().trim() ?? '',
|
||||
turnId: json['turnId']?.toString().trim() ?? '',
|
||||
runId: runId,
|
||||
artifactScope: artifactScope,
|
||||
artifactDirectory: json['artifactDirectory']?.toString().trim() ?? '',
|
||||
gatewayProviderId:
|
||||
json['gatewayProviderId']?.toString().trim().isNotEmpty == true
|
||||
? json['gatewayProviderId'].toString().trim()
|
||||
: (json['resolvedGatewayProviderId']?.toString().trim().isNotEmpty ==
|
||||
true
|
||||
? json['resolvedGatewayProviderId'].toString().trim()
|
||||
: 'openclaw'),
|
||||
runtimeBudgetMinutes: asInt(json['runtimeBudgetMinutes']),
|
||||
startedAtMs: asDouble(json['startedAtMs']),
|
||||
status: json['status']?.toString().trim().isNotEmpty == true
|
||||
? json['status'].toString().trim()
|
||||
: 'running',
|
||||
taskLoadClass: json['taskLoadClass']?.toString().trim() ?? '',
|
||||
sessionKey: json['sessionKey']?.toString().trim() ?? '',
|
||||
requiredArtifactExtensions: _stringListFromJson(
|
||||
json['requiredArtifactExtensions'],
|
||||
),
|
||||
expectedArtifactExtensions: _stringListFromJson(
|
||||
json['expectedArtifactExtensions'],
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -931,6 +1096,7 @@ class TaskThread {
|
||||
double? lastArtifactSyncAtMs,
|
||||
String? lastArtifactSyncStatus,
|
||||
List<String>? lastTaskArtifactRelativePaths,
|
||||
OpenClawTaskAssociation? openClawTaskAssociation,
|
||||
}) : threadId = _resolveThreadId(threadId),
|
||||
title = title ?? '',
|
||||
ownerScope =
|
||||
@ -977,6 +1143,7 @@ class TaskThread {
|
||||
lastTaskArtifactRelativePaths: _stringListFromJson(
|
||||
lastTaskArtifactRelativePaths,
|
||||
),
|
||||
openClawTaskAssociation: openClawTaskAssociation,
|
||||
),
|
||||
lifecycleState =
|
||||
lifecycleState ??
|
||||
@ -1015,6 +1182,8 @@ class TaskThread {
|
||||
String? get lastArtifactSyncStatus => contextState.lastArtifactSyncStatus;
|
||||
List<String> get lastTaskArtifactRelativePaths =>
|
||||
contextState.lastTaskArtifactRelativePaths;
|
||||
OpenClawTaskAssociation? get openClawTaskAssociation =>
|
||||
contextState.openClawTaskAssociation;
|
||||
String get latestResolvedRuntimeModel =>
|
||||
contextState.latestResolvedRuntimeModel;
|
||||
String get latestResolvedProviderId => contextState.latestResolvedProviderId;
|
||||
@ -1058,6 +1227,8 @@ class TaskThread {
|
||||
double? lastArtifactSyncAtMs,
|
||||
String? lastArtifactSyncStatus,
|
||||
List<String>? lastTaskArtifactRelativePaths,
|
||||
OpenClawTaskAssociation? openClawTaskAssociation,
|
||||
bool clearOpenClawTaskAssociation = false,
|
||||
}) {
|
||||
return TaskThread(
|
||||
threadId: threadId ?? this.threadId,
|
||||
@ -1081,6 +1252,8 @@ class TaskThread {
|
||||
lastArtifactSyncAtMs: lastArtifactSyncAtMs,
|
||||
lastArtifactSyncStatus: lastArtifactSyncStatus,
|
||||
lastTaskArtifactRelativePaths: lastTaskArtifactRelativePaths,
|
||||
openClawTaskAssociation: openClawTaskAssociation,
|
||||
clearOpenClawTaskAssociation: clearOpenClawTaskAssociation,
|
||||
),
|
||||
lifecycleState: (lifecycleState ?? this.lifecycleState).copyWith(
|
||||
archived: archived,
|
||||
@ -1197,6 +1370,7 @@ class TaskThread {
|
||||
'lastArtifactSyncAtMs': json['lastArtifactSyncAtMs'],
|
||||
'lastArtifactSyncStatus': json['lastArtifactSyncStatus'],
|
||||
'lastTaskArtifactRelativePaths': json['lastTaskArtifactRelativePaths'],
|
||||
'openClawTaskAssociation': json['openClawTaskAssociation'],
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@ -3615,7 +3615,7 @@ void main() {
|
||||
'openclaw-second-task',
|
||||
);
|
||||
await secondSubmitFuture;
|
||||
expect(controller.openClawGatewayActiveTasksInternal, 0);
|
||||
await _waitForOpenClawActiveTaskCount(controller, 0);
|
||||
expect(
|
||||
controller.chatMessages.map((message) => message.text),
|
||||
contains('second task completed'),
|
||||
@ -4083,6 +4083,20 @@ Future<void> _selectGatewaySession(
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> _waitForOpenClawActiveTaskCount(
|
||||
AppController controller,
|
||||
int expected,
|
||||
) async {
|
||||
final deadline = DateTime.now().add(const Duration(seconds: 5));
|
||||
while (DateTime.now().isBefore(deadline)) {
|
||||
if (controller.openClawGatewayActiveTasksInternal == expected) {
|
||||
return;
|
||||
}
|
||||
await Future<void>.delayed(const Duration(milliseconds: 10));
|
||||
}
|
||||
expect(controller.openClawGatewayActiveTasksInternal, expected);
|
||||
}
|
||||
|
||||
Future<List<String>> _startOpenClawActiveTasks(
|
||||
AppController controller,
|
||||
_BlockingGoTaskServiceClient fakeGoTaskService, {
|
||||
@ -4217,12 +4231,36 @@ class _RecordingGoTaskServiceClient implements GoTaskServiceClient {
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<GoTaskServiceResult> getTask({
|
||||
required AssistantExecutionTarget target,
|
||||
required OpenClawTaskAssociation association,
|
||||
required GoTaskServiceRoute route,
|
||||
}) async {
|
||||
return GoTaskServiceResult(
|
||||
success: true,
|
||||
message: 'ok',
|
||||
turnId: association.turnId,
|
||||
raw: <String, dynamic>{
|
||||
'success': true,
|
||||
'status': 'completed',
|
||||
'turnId': association.turnId,
|
||||
'runId': association.runId,
|
||||
'output': 'ok',
|
||||
},
|
||||
errorMessage: '',
|
||||
resolvedModel: '',
|
||||
route: route,
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> cancelTask({
|
||||
required GoTaskServiceRoute route,
|
||||
required AssistantExecutionTarget target,
|
||||
required String sessionId,
|
||||
required String threadId,
|
||||
OpenClawTaskAssociation? association,
|
||||
}) async {}
|
||||
|
||||
@override
|
||||
@ -4275,6 +4313,29 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient {
|
||||
return completer.future;
|
||||
}
|
||||
|
||||
@override
|
||||
Future<GoTaskServiceResult> getTask({
|
||||
required AssistantExecutionTarget target,
|
||||
required OpenClawTaskAssociation association,
|
||||
required GoTaskServiceRoute route,
|
||||
}) async {
|
||||
return GoTaskServiceResult(
|
||||
success: true,
|
||||
message: 'cleanup',
|
||||
turnId: association.turnId,
|
||||
raw: <String, dynamic>{
|
||||
'success': true,
|
||||
'status': 'completed',
|
||||
'turnId': association.turnId,
|
||||
'runId': association.runId,
|
||||
'output': 'cleanup',
|
||||
},
|
||||
errorMessage: '',
|
||||
resolvedModel: '',
|
||||
route: route,
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> waitForRequestCount(int count) async {
|
||||
final deadline = DateTime.now().add(const Duration(seconds: 15));
|
||||
while (requests.length < count && DateTime.now().isBefore(deadline)) {
|
||||
@ -4338,6 +4399,7 @@ class _BlockingGoTaskServiceClient implements GoTaskServiceClient {
|
||||
required AssistantExecutionTarget target,
|
||||
required String sessionId,
|
||||
required String threadId,
|
||||
OpenClawTaskAssociation? association,
|
||||
}) async {
|
||||
cancelledSessionIds.add(sessionId);
|
||||
}
|
||||
|
||||
@ -253,38 +253,6 @@ void main() {
|
||||
});
|
||||
});
|
||||
|
||||
group('Gateway ACP task runtime budget', () {
|
||||
test(
|
||||
'treats explicit document and media formats as long artifact tasks',
|
||||
() {
|
||||
expect(
|
||||
gatewayAcpTaskRuntimeBudgetMinutesForParams(const <String, dynamic>{
|
||||
'taskPrompt': '围绕上述话题输出一篇800-1000字左右的文章 markdown格式',
|
||||
}),
|
||||
30,
|
||||
);
|
||||
expect(
|
||||
gatewayAcpTaskRuntimeBudgetMinutesForParams(const <String, dynamic>{
|
||||
'taskPrompt': 'save the final article as article.docx',
|
||||
}),
|
||||
30,
|
||||
);
|
||||
expect(
|
||||
gatewayAcpTaskRuntimeBudgetMinutesForParams(const <String, dynamic>{
|
||||
'taskPrompt': '生成封面图 png 和短视频 mp4',
|
||||
}),
|
||||
30,
|
||||
);
|
||||
expect(
|
||||
gatewayAcpTaskRuntimeBudgetMinutesForParams(const <String, dynamic>{
|
||||
'taskPrompt': '导出 jpg、pptx 和 xls 文件',
|
||||
}),
|
||||
30,
|
||||
);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
group('GatewayAcpClient authorization', () {
|
||||
test('normalizes raw resolver token into bearer header for HTTP', () async {
|
||||
final capture = await _startAcpHttpServer();
|
||||
@ -672,7 +640,7 @@ void main() {
|
||||
);
|
||||
|
||||
test(
|
||||
'recovers OpenClaw task result from bridge session snapshot after SSE connection close',
|
||||
'recovers OpenClaw task result from bridge task snapshot after SSE connection close',
|
||||
() async {
|
||||
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
||||
addTearDown(() => server.close(force: true));
|
||||
@ -701,7 +669,7 @@ void main() {
|
||||
socket.destroy();
|
||||
return;
|
||||
}
|
||||
if (method == 'xworkmate.sessions.get') {
|
||||
if (method == 'xworkmate.tasks.get') {
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<String, dynamic>{
|
||||
@ -717,7 +685,7 @@ void main() {
|
||||
},
|
||||
'result': <String, dynamic>{
|
||||
'success': true,
|
||||
'output': 'recovered from bridge session snapshot',
|
||||
'output': 'recovered from bridge task snapshot',
|
||||
'turnId': 'turn-recovered',
|
||||
},
|
||||
'artifacts': <String, dynamic>{
|
||||
@ -772,7 +740,7 @@ void main() {
|
||||
);
|
||||
|
||||
expect(result.success, isTrue);
|
||||
expect(result.message, 'recovered from bridge session snapshot');
|
||||
expect(result.message, 'recovered from bridge task snapshot');
|
||||
expect(result.artifacts.single.relativePath, 'exports/snapshot.md');
|
||||
expect(result.remoteWorkingDirectory, '/remote/openclaw/workspace');
|
||||
expect(result.remoteWorkspaceRefKind, WorkspaceRefKind.remotePath);
|
||||
@ -818,7 +786,7 @@ void main() {
|
||||
socket.destroy();
|
||||
return;
|
||||
}
|
||||
if (method == 'xworkmate.sessions.get') {
|
||||
if (method == 'xworkmate.tasks.get') {
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<String, dynamic>{
|
||||
@ -890,13 +858,13 @@ void main() {
|
||||
expect(result.artifacts.single.relativePath, 'reports/final.md');
|
||||
expect(requestMethods, <String>[
|
||||
'session.start',
|
||||
'xworkmate.sessions.get',
|
||||
'xworkmate.tasks.get',
|
||||
]);
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
'recovers OpenClaw follow-up from bridge session snapshot after SSE ends without final envelope',
|
||||
'recovers OpenClaw follow-up from bridge task snapshot after SSE ends without final envelope',
|
||||
() async {
|
||||
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
||||
addTearDown(() => server.close(force: true));
|
||||
@ -921,7 +889,7 @@ void main() {
|
||||
await request.response.close();
|
||||
return;
|
||||
}
|
||||
if (method == 'xworkmate.sessions.get') {
|
||||
if (method == 'xworkmate.tasks.get') {
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<String, dynamic>{
|
||||
@ -1014,7 +982,7 @@ void main() {
|
||||
socket.destroy();
|
||||
return;
|
||||
}
|
||||
if (method == 'xworkmate.sessions.get') {
|
||||
if (method == 'xworkmate.tasks.get') {
|
||||
snapshotPolls += 1;
|
||||
final completed = snapshotPolls >= 3;
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
@ -1080,96 +1048,6 @@ void main() {
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
'uses long task budget for default OpenClaw SSE recovery polling',
|
||||
() async {
|
||||
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
||||
addTearDown(() => server.close(force: true));
|
||||
var snapshotPolls = 0;
|
||||
server.listen((request) async {
|
||||
final body = await utf8.decoder.bind(request).join();
|
||||
final decoded = jsonDecode(body) as Map<String, dynamic>;
|
||||
final method = decoded['method']?.toString() ?? '';
|
||||
final id = decoded['id']?.toString() ?? 'request-id';
|
||||
if (method == 'session.start') {
|
||||
final event = jsonEncode(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'method': 'xworkmate.bridge.accepted',
|
||||
'params': <String, dynamic>{'sessionId': 'unit-fixture-task-d'},
|
||||
});
|
||||
request.response.headers.set(
|
||||
HttpHeaders.contentTypeHeader,
|
||||
'text/event-stream',
|
||||
);
|
||||
request.response.write('data: $event\n\n');
|
||||
await request.response.close();
|
||||
return;
|
||||
}
|
||||
if (method == 'xworkmate.sessions.get') {
|
||||
snapshotPolls += 1;
|
||||
final completed = snapshotPolls >= 301;
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': id,
|
||||
'result': <String, dynamic>{
|
||||
'status': completed ? 'completed' : 'running',
|
||||
'sessionId': 'unit-fixture-task-d',
|
||||
'threadId': 'unit-fixture-task-d',
|
||||
'task': <String, dynamic>{
|
||||
'state': completed ? 'completed' : 'running',
|
||||
'turnId': 'turn-recovered-long',
|
||||
},
|
||||
if (completed)
|
||||
'result': <String, dynamic>{
|
||||
'success': true,
|
||||
'output': 'recovered after long polling window',
|
||||
'turnId': 'turn-recovered-long',
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
await request.response.close();
|
||||
return;
|
||||
}
|
||||
request.response.statusCode = HttpStatus.badRequest;
|
||||
await request.response.close();
|
||||
});
|
||||
final endpoint = Uri.parse('http://127.0.0.1:${server.port}');
|
||||
final transport = ExternalCodeAgentAcpDesktopTransport(
|
||||
client: GatewayAcpClient(endpointResolver: () => endpoint),
|
||||
endpointResolver: (_) => endpoint,
|
||||
taskEndpointResolver: (_) => endpoint,
|
||||
recoveryPollDelay: const Duration(microseconds: 1),
|
||||
);
|
||||
addTearDown(transport.dispose);
|
||||
|
||||
final result = await transport.executeTask(
|
||||
const GoTaskServiceRequest(
|
||||
sessionId: 'unit-fixture-task-d',
|
||||
threadId: 'unit-fixture-task-d',
|
||||
target: AssistantExecutionTarget.gateway,
|
||||
provider: SingleAgentProvider.openclaw,
|
||||
prompt: '生成封面图 png 和短视频 mp4',
|
||||
workingDirectory: '/tmp/workspace',
|
||||
model: '',
|
||||
thinking: 'off',
|
||||
selectedSkills: <String>[],
|
||||
inlineAttachments: <GatewayChatAttachmentPayload>[],
|
||||
localAttachments: <CollaborationAttachment>[],
|
||||
agentId: '',
|
||||
metadata: <String, dynamic>{},
|
||||
),
|
||||
onUpdate: (_) {},
|
||||
);
|
||||
|
||||
expect(snapshotPolls, 301);
|
||||
expect(result.success, isTrue);
|
||||
expect(result.message, 'recovered after long polling window');
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
'recovers terminal failed OpenClaw snapshot without displayable result',
|
||||
() async {
|
||||
@ -1198,7 +1076,7 @@ void main() {
|
||||
socket.destroy();
|
||||
return;
|
||||
}
|
||||
if (method == 'xworkmate.sessions.get') {
|
||||
if (method == 'xworkmate.tasks.get') {
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<String, dynamic>{
|
||||
@ -1957,7 +1835,7 @@ void main() {
|
||||
},
|
||||
);
|
||||
|
||||
test('task submit uses dynamic HTTP response timeout budgets', () {
|
||||
test('task submit uses short HTTP response timeout', () {
|
||||
final openClawEndpoint = Uri.parse(
|
||||
'https://xworkmate-bridge.svc.plus/acp/rpc',
|
||||
);
|
||||
@ -1971,7 +1849,7 @@ void main() {
|
||||
'session.start',
|
||||
const <String, dynamic>{'requestedExecutionTarget': 'gateway'},
|
||||
),
|
||||
const Duration(minutes: 10),
|
||||
const Duration(seconds: 120),
|
||||
);
|
||||
expect(
|
||||
gatewayAcpHttpResponseTimeoutFor(
|
||||
@ -1982,11 +1860,11 @@ void main() {
|
||||
'requestedExecutionTarget': 'gateway',
|
||||
},
|
||||
),
|
||||
const Duration(minutes: 30),
|
||||
const Duration(seconds: 120),
|
||||
);
|
||||
expect(
|
||||
gatewayAcpHttpResponseTimeoutFor(acpEndpoint, 'session.start'),
|
||||
const Duration(minutes: 2),
|
||||
const Duration(seconds: 120),
|
||||
);
|
||||
expect(
|
||||
gatewayAcpHttpResponseTimeoutFor(openClawEndpoint, 'acp.capabilities'),
|
||||
|
||||
Loading…
Reference in New Issue
Block a user