diff --git a/docs/WINDOWS_NODE_TESTING.md b/docs/WINDOWS_NODE_TESTING.md index 2a2c1a888..52cb4815b 100644 --- a/docs/WINDOWS_NODE_TESTING.md +++ b/docs/WINDOWS_NODE_TESTING.md @@ -104,7 +104,7 @@ When the node connects, it advertises these capabilities: - `tts` - Windows speech synthesis or ElevenLabs playback, when enabled in Settings - `stt` - Local speech-to-text via Whisper.net, when enabled in Settings -Local MCP clients also see MCP-only `app.*` commands such as `app.navigate`, `app.status`, and `app.chat.snapshot`/`app.chat.send`/`app.chat.reset`. These are local testing and automation hooks registered with the tray's MCP server and are not advertised to the gateway WebSocket. +Local MCP clients also see MCP-only `app.*` commands such as `app.navigate`, `app.status`, `app.chat.snapshot`/`app.chat.send`/`app.chat.reset`, and `app.chat.queue.list`/`app.chat.queue.cancel`. These are local testing and automation hooks registered with the tray's MCP server and are not advertised to the gateway WebSocket. ## Security Features diff --git a/src/OpenClaw.Chat/ChatModels.cs b/src/OpenClaw.Chat/ChatModels.cs index 4de7d2151..da526cd1d 100644 --- a/src/OpenClaw.Chat/ChatModels.cs +++ b/src/OpenClaw.Chat/ChatModels.cs @@ -274,6 +274,7 @@ public interface IChatDataProvider : IAsyncDisposable Task SendMessageAsync(string threadId, string message, CancellationToken cancellationToken = default); Task SendMessageAsync(string threadId, string message, CancellationToken cancellationToken, IReadOnlyList? attachments) => SendMessageAsync(threadId, message, cancellationToken); + Task CancelQueuedMessageAsync(string threadId, string queuedMessageId, CancellationToken cancellationToken = default) => Task.FromResult(false); Task StopResponseAsync(string threadId, CancellationToken cancellationToken = default); Task SetThreadSuspendedAsync(string threadId, bool suspended, CancellationToken cancellationToken = default); Task DeleteThreadAsync(string threadId, CancellationToken cancellationToken = default); diff --git a/src/OpenClaw.Shared/Capabilities/AppCapability.cs b/src/OpenClaw.Shared/Capabilities/AppCapability.cs index 0746523c3..5faac7ccc 100644 --- a/src/OpenClaw.Shared/Capabilities/AppCapability.cs +++ b/src/OpenClaw.Shared/Capabilities/AppCapability.cs @@ -29,6 +29,8 @@ public class AppCapability : NodeCapabilityBase "app.chat.snapshot", "app.chat.send", "app.chat.reset", + "app.chat.queue.list", + "app.chat.queue.cancel", }; public override IReadOnlyList Commands => _commands; @@ -48,6 +50,8 @@ public class AppCapability : NodeCapabilityBase public Func>? ChatSnapshotHandler; public Func>? ChatSendHandler; public Func>? ChatResetHandler; + public Func>? ChatQueueListHandler; + public Func>? ChatQueueCancelHandler; public AppCapability(IOpenClawLogger logger) : base(logger) { } @@ -69,6 +73,8 @@ public override async Task ExecuteAsync(NodeInvokeRequest re "app.chat.snapshot" => await HandleChatSnapshot(request), "app.chat.send" => await HandleChatSend(request), "app.chat.reset" => await HandleChatReset(request), + "app.chat.queue.list" => await HandleChatQueueList(request), + "app.chat.queue.cancel" => await HandleChatQueueCancel(request), _ => Error($"Unknown command: {request.Command}") }; } @@ -203,7 +209,7 @@ private async Task HandleChatSnapshot(NodeInvokeRequest requ if (ChatSnapshotHandler == null) return Error("Chat snapshot handler not registered"); - var result = await ChatSnapshotHandler(GetOptionalThreadId(request)); + var result = await ChatSnapshotHandler(GetThreadIdOrSessionKeyArg(request)); return Success(result); } @@ -216,7 +222,7 @@ private async Task HandleChatSend(NodeInvokeRequest request) if (string.IsNullOrWhiteSpace(message)) return Error("Missing required arg: message"); - var result = await ChatSendHandler(GetOptionalThreadId(request), message); + var result = await ChatSendHandler(GetThreadIdOrSessionKeyArg(request), message); return Success(result); } @@ -225,11 +231,37 @@ private async Task HandleChatReset(NodeInvokeRequest request if (ChatResetHandler == null) return Error("Chat reset handler not registered"); - var result = await ChatResetHandler(GetOptionalThreadId(request)); + var result = await ChatResetHandler(GetThreadIdOrSessionKeyArg(request)); return Success(result); } - private string? GetOptionalThreadId(NodeInvokeRequest request) => + private async Task HandleChatQueueList(NodeInvokeRequest request) + { + if (ChatQueueListHandler == null) + return Error("Chat queue list handler not registered"); + + var result = await ChatQueueListHandler(GetThreadIdOrSessionKeyArg(request)); + return Success(result); + } + + private async Task HandleChatQueueCancel(NodeInvokeRequest request) + { + if (ChatQueueCancelHandler == null) + return Error("Chat queue cancel handler not registered"); + + var queuedMessageId = GetStringArg(request.Args, "queuedMessageId"); + if (string.IsNullOrWhiteSpace(queuedMessageId)) + return Error("Missing required arg: queuedMessageId"); + + var threadId = GetThreadIdOrSessionKeyArg(request); + if (string.IsNullOrWhiteSpace(threadId)) + return Error("Missing required arg: threadId"); + + var result = await ChatQueueCancelHandler(threadId, queuedMessageId); + return Success(result); + } + + private string? GetThreadIdOrSessionKeyArg(NodeInvokeRequest request) => GetStringArg(request.Args, "threadId") ?? GetStringArg(request.Args, "sessionKey"); } diff --git a/src/OpenClaw.Shared/Mcp/McpToolBridge.cs b/src/OpenClaw.Shared/Mcp/McpToolBridge.cs index 12f9eae31..525f6db5d 100644 --- a/src/OpenClaw.Shared/Mcp/McpToolBridge.cs +++ b/src/OpenClaw.Shared/Mcp/McpToolBridge.cs @@ -275,11 +275,15 @@ private object HandleToolsList() ["app.dashboard.url"] = "Build the same gateway dashboard URL the tray opens. Args: path (string, optional). Returns { url, credentialSource, usesSharedGatewayToken, hasTokenQuery }.", ["app.chat.snapshot"] = - "READ-ALL: Return the current native chat snapshot for local automation. Args: threadId/sessionKey (string, optional). Returns connection state, compose target, thread summaries, and recent timeline entries including chat text.", + "READ-ALL: Return the current native chat snapshot for local automation. Args: threadId/sessionKey (string, optional). Returns connection state, compose target, thread summaries, queued outgoing messages, and recent timeline entries including chat text.", ["app.chat.send"] = "Send a message through the native chat provider. Args: message (string, required), threadId/sessionKey (string, optional; defaults to the current compose/default thread). Returns { sent, threadId, entryCount, turnActive, error? }.", ["app.chat.reset"] = "Reset a chat session through the gateway sessions.reset path. Args: threadId/sessionKey (string, optional; defaults to the current compose/default thread, so no-arg reset clears the active chat). Returns { reset, threadId, error? }.", + ["app.chat.queue.list"] = + "READ-ALL: List native chat outgoing queue entries. Args: threadId/sessionKey (string, optional; omit to return all queued threads). Returns { defaultThreadId, requestedThreadId, totalCount, selectedThread, threads: [{ threadId, count, messages: [{ id, text, createdAt, sendState, errorText, canCancel }] }] }.", + ["app.chat.queue.cancel"] = + "Cancel/remove one native chat outgoing queue entry before it is sent. Args: queuedMessageId (string, required), threadId/sessionKey (string, required; use the threadId returned by app.chat.queue.list or app.chat.snapshot). Only Queued/Failed entries can be removed; Sending entries may already have reached the gateway. Returns { canceled, threadId, queuedMessageId, remainingCount, error? }.", // location.* ["location.get"] = diff --git a/src/OpenClaw.Shared/Models.cs b/src/OpenClaw.Shared/Models.cs index 07049059f..783208799 100644 --- a/src/OpenClaw.Shared/Models.cs +++ b/src/OpenClaw.Shared/Models.cs @@ -2015,6 +2015,7 @@ public static bool IsFailureStatus(string? status) || status.Equals("rejected", StringComparison.OrdinalIgnoreCase) || status.Equals("denied", StringComparison.OrdinalIgnoreCase) || status.Equals("aborted", StringComparison.OrdinalIgnoreCase) + || status.Equals("timeout", StringComparison.OrdinalIgnoreCase) || status.Equals("cancelled", StringComparison.OrdinalIgnoreCase) || status.Equals("canceled", StringComparison.OrdinalIgnoreCase); } diff --git a/src/OpenClaw.Shared/OpenClawGatewayClient.cs b/src/OpenClaw.Shared/OpenClawGatewayClient.cs index 1d5c139e6..66f1d316e 100644 --- a/src/OpenClaw.Shared/OpenClawGatewayClient.cs +++ b/src/OpenClaw.Shared/OpenClawGatewayClient.cs @@ -309,7 +309,12 @@ public async Task SendChatMessageAsync(string message, string? sessionKey = null _ = await SendChatMessageForRunAsync(message, sessionKey, sessionId, attachments).ConfigureAwait(false); } - public async Task SendChatMessageForRunAsync(string message, string? sessionKey = null, string? sessionId = null, IReadOnlyList? attachments = null) + public async Task SendChatMessageForRunAsync( + string message, + string? sessionKey = null, + string? sessionId = null, + IReadOnlyList? attachments = null, + string? idempotencyKey = null) { if (!IsConnected) throw new InvalidOperationException("Gateway connection is not open"); @@ -317,9 +322,12 @@ public async Task SendChatMessageForRunAsync(string message, str var hasAttachments = attachments is { Count: > 0 }; if (string.IsNullOrWhiteSpace(message) && !hasAttachments) throw new ArgumentException("Message or attachment is required", nameof(message)); + if (idempotencyKey is not null && string.IsNullOrWhiteSpace(idempotencyKey)) + throw new ArgumentException("Idempotency key cannot be whitespace", nameof(idempotencyKey)); var effectiveSessionKey = ResolveEffectiveSessionKey( sessionKey, Volatile.Read(ref _mainSessionKey), "chat.send"); + var effectiveIdempotencyKey = idempotencyKey?.Trim() ?? Guid.NewGuid().ToString(); var requestId = Guid.NewGuid().ToString(); var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); @@ -336,7 +344,7 @@ public async Task SendChatMessageForRunAsync(string message, str { ["sessionKey"] = effectiveSessionKey, ["message"] = message, - ["idempotencyKey"] = Guid.NewGuid().ToString() + ["idempotencyKey"] = effectiveIdempotencyKey }; if (hasAttachments) diff --git a/src/OpenClaw.Tray.WinUI/App.CapabilityHandlers.cs b/src/OpenClaw.Tray.WinUI/App.CapabilityHandlers.cs index ff15d7ed2..a675f431b 100644 --- a/src/OpenClaw.Tray.WinUI/App.CapabilityHandlers.cs +++ b/src/OpenClaw.Tray.WinUI/App.CapabilityHandlers.cs @@ -230,6 +230,8 @@ private void WireAppCapabilityHandlers() app.ChatSnapshotHandler = BuildChatSnapshotForMcpAsync; app.ChatSendHandler = SendChatMessageForMcpAsync; app.ChatResetHandler = ResetChatSessionForMcpAsync; + app.ChatQueueListHandler = ListQueuedChatMessagesForMcpAsync; + app.ChatQueueCancelHandler = CancelQueuedChatMessageForMcpAsync; connection.ApplySetupCodeHandler = async setupCode => { @@ -506,6 +508,79 @@ private object BuildPendingApprovalsPayload( }; } + private async Task ListQueuedChatMessagesForMcpAsync(string? threadId) + { + var provider = _chatCoordinator?.Provider; + if (provider == null) + return new { error = "Chat provider is not initialized" }; + + var snapshot = await provider.LoadAsync(); + var resolvedThreadId = ResolveChatThreadId(snapshot, threadId); + return BuildChatQueuePayload(snapshot, resolvedThreadId, filterToThread: !string.IsNullOrWhiteSpace(threadId)); + } + + private async Task CancelQueuedChatMessageForMcpAsync(string? threadId, string queuedMessageId) + { + var provider = _chatCoordinator?.Provider; + if (provider == null) + return new { canceled = false, error = "Chat provider is not initialized" }; + + var snapshot = await provider.LoadAsync(); + var resolvedThreadId = ResolveChatThreadId(snapshot, threadId); + if (string.IsNullOrWhiteSpace(resolvedThreadId)) + return new { canceled = false, queuedMessageId, error = "Chat compose target is not ready" }; + + if (!TryGetQueuedMessage(snapshot, resolvedThreadId, queuedMessageId, out var queuedMessage)) + { + return new + { + canceled = false, + threadId = resolvedThreadId, + queuedMessageId, + error = "Queued message was not found" + }; + } + + if (!CanCancelQueuedMessage(queuedMessage)) + { + return new + { + canceled = false, + threadId = resolvedThreadId, + queuedMessageId, + sendState = queuedMessage.SendState.ToString(), + error = "Queued message is already sending and cannot be canceled" + }; + } + + bool canceled; + try + { + canceled = await provider.CancelQueuedMessageAsync(resolvedThreadId, queuedMessageId); + } + catch (Exception ex) + { + Logger.Warn($"App: Chat queue cancel for '{resolvedThreadId}' message '{queuedMessageId}' failed: {ex.Message}"); + return new { canceled = false, threadId = resolvedThreadId, queuedMessageId, error = ex.Message }; + } + + var updated = await provider.LoadAsync(); + var stillQueued = TryGetQueuedMessage(updated, resolvedThreadId, queuedMessageId, out var remaining); + var remainingCount = GetQueuedMessagesForThread(updated, resolvedThreadId).Length; + return new + { + canceled, + threadId = resolvedThreadId, + queuedMessageId, + remainingCount, + error = canceled + ? null + : stillQueued + ? $"Queued message is still present with state '{remaining!.SendState}'." + : "Queued message was not canceled; it may have started sending before cancellation was processed." + }; + } + private static string? ResolveChatThreadId(ChatDataSnapshot snapshot, string? requestedThreadId) { if (!string.IsNullOrWhiteSpace(requestedThreadId)) @@ -548,6 +623,7 @@ private static object BuildChatSnapshotPayload(ChatDataSnapshot snapshot, string t.TotalTokens, t.ContextTokens }).ToArray(), + queue = BuildChatQueuePayload(snapshot, resolvedThreadId, filterToThread: false), selectedTimeline = selectedTimeline is null ? null : new { turnActive = selectedTimeline.TurnActive, @@ -580,6 +656,88 @@ private static object BuildChatSnapshotPayload(ChatDataSnapshot snapshot, string }; } + private static object BuildChatQueuePayload( + ChatDataSnapshot snapshot, + string? resolvedThreadId, + bool filterToThread) + { + var queued = snapshot.QueuedMessagesByThread ?? new Dictionary>(); + var threadQueues = filterToThread && !string.IsNullOrWhiteSpace(resolvedThreadId) + ? new[] + { + new KeyValuePair>( + resolvedThreadId, + GetQueuedMessagesForThread(snapshot, resolvedThreadId)) + } + : queued + .OrderBy(kvp => kvp.Key, StringComparer.Ordinal) + .ToArray(); + + var selectedMessages = !string.IsNullOrWhiteSpace(resolvedThreadId) + ? GetQueuedMessagesForThread(snapshot, resolvedThreadId) + : Array.Empty(); + + return new + { + defaultThreadId = snapshot.DefaultThreadId, + requestedThreadId = resolvedThreadId, + totalCount = threadQueues.Sum(kvp => kvp.Value.Count), + selectedThread = string.IsNullOrWhiteSpace(resolvedThreadId) + ? null + : new + { + threadId = resolvedThreadId, + count = selectedMessages.Length, + messages = selectedMessages.Select(ToMcpQueuedMessage).ToArray() + }, + threads = threadQueues.Select(kvp => new + { + threadId = kvp.Key, + count = kvp.Value.Count, + messages = kvp.Value.Select(ToMcpQueuedMessage).ToArray() + }).ToArray() + }; + } + + private static ChatQueuedMessage[] GetQueuedMessagesForThread(ChatDataSnapshot snapshot, string threadId) + { + if (snapshot.QueuedMessagesByThread?.TryGetValue(threadId, out var messages) == true) + return messages.ToArray(); + return Array.Empty(); + } + + private static bool TryGetQueuedMessage( + ChatDataSnapshot snapshot, + string threadId, + string queuedMessageId, + out ChatQueuedMessage queuedMessage) + { + foreach (var message in GetQueuedMessagesForThread(snapshot, threadId)) + { + if (string.Equals(message.Id, queuedMessageId, StringComparison.Ordinal)) + { + queuedMessage = message; + return true; + } + } + + queuedMessage = null!; + return false; + } + + private static bool CanCancelQueuedMessage(ChatQueuedMessage message) => + message.SendState != ChatQueuedMessageSendState.Sending; + + private static object ToMcpQueuedMessage(ChatQueuedMessage message) => new + { + id = message.Id, + text = message.Text, + createdAt = message.CreatedAt, + sendState = message.SendState.ToString(), + errorText = message.ErrorText, + canCancel = CanCancelQueuedMessage(message) + }; + private async Task WaitForAppStateUpdateAsync(string propertyName, Func requestAsync) { if (_appState == null) diff --git a/src/OpenClaw.Tray.WinUI/Chat/IChatGatewayBridge.cs b/src/OpenClaw.Tray.WinUI/Chat/IChatGatewayBridge.cs index bdbdb34bb..1a2b29348 100644 --- a/src/OpenClaw.Tray.WinUI/Chat/IChatGatewayBridge.cs +++ b/src/OpenClaw.Tray.WinUI/Chat/IChatGatewayBridge.cs @@ -35,7 +35,12 @@ public interface IChatGatewayBridge : IDisposable void StartProactiveBootstrap(); Task SendChatMessageAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null); - Task SendChatMessageForRunAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null); + Task SendChatMessageForRunAsync( + string message, + string? sessionKey, + string? sessionId, + IReadOnlyList? attachments = null, + string? idempotencyKey = null); /// /// Fetches the gateway command catalog (commands.list) via the typed /// protocol API. Returns a whose @@ -170,8 +175,13 @@ public void StartProactiveBootstrap() public Task SendChatMessageAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null) => _client.SendChatMessageAsync(message, sessionKey, sessionId, attachments); - public Task SendChatMessageForRunAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null) => - _client.SendChatMessageForRunAsync(message, sessionKey, sessionId, attachments); + public Task SendChatMessageForRunAsync( + string message, + string? sessionKey, + string? sessionId, + IReadOnlyList? attachments = null, + string? idempotencyKey = null) => + _client.SendChatMessageForRunAsync(message, sessionKey, sessionId, attachments, idempotencyKey); public Task PatchSessionModelAsync(string sessionKey, string model) => _client.PatchSessionAsync(sessionKey, new SessionPatch { Model = model }); diff --git a/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatDataProvider.cs b/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatDataProvider.cs index 52d43c896..f323959ab 100644 --- a/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatDataProvider.cs +++ b/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatDataProvider.cs @@ -141,8 +141,8 @@ public sealed class OpenClawChatDataProvider : IChatDataProvider private readonly Dictionary> _queuedSendRequests = new(); private readonly Dictionary> _queuedMessageIdsByRunId = new(); private readonly Dictionary> _terminalRunIdsByThread = new(); + private readonly HashSet _queuedDrainScheduledThreads = new(StringComparer.Ordinal); private readonly HashSet _assistantFallbackPromotedThreads = new(StringComparer.Ordinal); - private readonly Dictionary _deferredAssistantMessages = new(StringComparer.Ordinal); private long _queuedMessageSequence; private int _keylessEventDiagnosticRaised; // Threads where we locally initiated the current turn (via SendMessageAsync). @@ -152,25 +152,31 @@ public sealed class OpenClawChatDataProvider : IChatDataProvider // Per-thread retry count for LoadHistoryAsync to prevent unbounded retry loops. private readonly Dictionary _historyRetryCount = new(); private const int MaxHistoryRetries = 3; + private const int MaxDeferredAdmissionRetries = 8; private static readonly TimeSpan LocalEchoSuppressionWindow = TimeSpan.FromSeconds(30); + private static readonly TimeSpan DeferredQueueDrainDelay = TimeSpan.FromMilliseconds(100); + private static readonly TimeSpan MaxDeferredAdmissionRetryDelay = TimeSpan.FromSeconds(1); private readonly record struct LocalSentText(string Text, DateTimeOffset SentAt, string QueuedMessageId); private sealed record QueuedSendRequest( string Id, + string SendRunId, string ThreadId, string Text, string DisplayText, string LocalNonce, - IReadOnlyList? Attachments); + IReadOnlyList? Attachments, + int DeferredAdmissionRetryCount = 0, + DateTimeOffset? DeferredAdmissionRetryAfter = null); private sealed record QueuedSendDispatch( QueuedSendRequest Request, string? SessionId, long ResetVersion, long StartedLifecycleSequence, - long StartedRunStartSequence); + long StartedRunStartSequence, + bool StartedDirectly); private enum AssistantQueueFrameDisposition { Render, - Defer, Drop, } private sealed record LocalInlineApproval( @@ -365,6 +371,7 @@ public async Task SendMessageAsync(string threadId, string message, Cancellation var request = new QueuedSendRequest( messageId, + Guid.NewGuid().ToString(), threadId, trimmed, displayText, @@ -383,7 +390,7 @@ public async Task SendMessageAsync(string threadId, string message, Cancellation DateTimeOffset.UtcNow, nonce)); AddQueuedSendRequestLocked(request); - dispatch = TryStartNextQueuedSendLocked(threadId, requireConnected: false); + dispatch = TryStartNextQueuedSendLocked(threadId, requireConnected: false, out _); } snapshot = BuildSnapshotLocked(); @@ -394,6 +401,30 @@ public async Task SendMessageAsync(string threadId, string message, Cancellation await DispatchQueuedSendAsync(dispatch, rethrow: true, cancellationToken); } + public Task CancelQueuedMessageAsync(string threadId, string queuedMessageId, CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + if (string.IsNullOrEmpty(threadId)) + throw new ArgumentException("Thread id is required.", nameof(threadId)); + if (string.IsNullOrEmpty(queuedMessageId)) + throw new ArgumentException("Queued message id is required.", nameof(queuedMessageId)); + + ChatDataSnapshot? snapshot = null; + var canceled = false; + lock (_gate) + { + ObjectDisposedException.ThrowIf(_disposed, this); + canceled = CancelQueuedMessageLocked(threadId, queuedMessageId); + if (canceled) + snapshot = BuildSnapshotLocked(); + } + + if (snapshot is not null) + Publish(snapshot); + + return Task.FromResult(canceled); + } + private async Task DispatchQueuedSendAsync( QueuedSendDispatch dispatch, bool rethrow, @@ -410,8 +441,15 @@ private async Task DispatchQueuedSendAsync( { if (_disposed) return; + if (GetResetVersionLocked(threadId) == dispatch.ResetVersion) + TrackQueuedMessageRunLocked(threadId, request.SendRunId, request.Id); } - var sendResult = await _bridge.SendChatMessageForRunAsync(request.Text, threadId, dispatch.SessionId, request.Attachments); + var sendResult = await _bridge.SendChatMessageForRunAsync( + request.Text, + threadId, + dispatch.SessionId, + request.Attachments, + idempotencyKey: request.SendRunId); if (sendResult.IsTerminalFailure) { var failure = !string.IsNullOrWhiteSpace(sendResult.Error) @@ -425,29 +463,68 @@ private async Task DispatchQueuedSendAsync( bool sendStillCurrent; string? staleRunIdToAbort = null; - ChatDataSnapshot? acceptedRunAlreadyStartedSnapshot = null; + ChatDataSnapshot? acceptedSnapshot = null; + ChatDataSnapshot? requeuedSnapshot = null; + var retryDeferredSend = false; + var deferredRetryDelay = DeferredQueueDrainDelay; + var acceptedRunId = string.IsNullOrWhiteSpace(sendResult.RunId) + ? null + : sendResult.RunId!; lock (_gate) { sendStillCurrent = GetResetVersionLocked(threadId) == dispatch.ResetVersion; if (!sendStillCurrent) { - if (!string.IsNullOrEmpty(sendResult.RunId)) + staleRunIdToAbort = acceptedRunId ?? request.SendRunId; + AddResetIgnoredRunIdLocked(threadId, staleRunIdToAbort); + } + else if (IsDeferredAdmissionStatus(sendResult.Status)) + { + var runAlreadyStarted = !string.IsNullOrEmpty(acceptedRunId) + && _activeRunIds.TryGetValue(threadId, out var activeRunId) + && _activeRunStartSequences.TryGetValue(threadId, out var activeStartSequence) + && string.Equals(activeRunId, acceptedRunId, StringComparison.Ordinal) + && activeStartSequence > dispatch.StartedRunStartSequence; + if (runAlreadyStarted) + { + TrackQueuedMessageRunLocked(threadId, acceptedRunId!, request.Id); + AddResetAcceptedRunIdLocked(threadId, acceptedRunId!); + if (PromoteQueuedMessageLocked(threadId, request.Id)) + { + acceptedSnapshot = BuildSnapshotLocked(); + } + else + { + RemoveQueuedRunMappingByMessageIdLocked(threadId, request.Id); + } + } + else if (RequeueDeferredAdmissionLocked(threadId, request.Id, out deferredRetryDelay)) + { + if (!string.IsNullOrEmpty(acceptedRunId)) + { + TrackQueuedMessageRunLocked(threadId, acceptedRunId, request.Id); + AddResetAcceptedRunIdLocked(threadId, acceptedRunId); + } + requeuedSnapshot = BuildSnapshotLocked(); + retryDeferredSend = true; + } + else if (dispatch.StartedDirectly) { - AddResetIgnoredRunIdLocked(threadId, sendResult.RunId!); - staleRunIdToAbort = sendResult.RunId; + throw new InvalidOperationException( + $"Gateway returned chat.send status {sendResult.Status} before admitting the direct send."); } } - else if (!string.IsNullOrEmpty(sendResult.RunId)) + else if (!string.IsNullOrEmpty(acceptedRunId)) { - TrackQueuedMessageRunLocked(threadId, sendResult.RunId!, request.Id); - AddResetAcceptedRunIdLocked(threadId, sendResult.RunId!); + TrackQueuedMessageRunLocked(threadId, acceptedRunId, request.Id); + AddResetAcceptedRunIdLocked(threadId, acceptedRunId); var runAlreadyStarted = _activeRunIds.TryGetValue(threadId, out var activeRunId) && _activeRunStartSequences.TryGetValue(threadId, out var activeStartSequence) - && string.Equals(activeRunId, sendResult.RunId, StringComparison.Ordinal) + && string.Equals(activeRunId, acceptedRunId, StringComparison.Ordinal) && activeStartSequence > dispatch.StartedRunStartSequence; - if (runAlreadyStarted && PromoteQueuedMessageLocked(threadId, request.Id)) + if (PromoteQueuedMessageLocked(threadId, request.Id)) { - acceptedRunAlreadyStartedSnapshot = BuildSnapshotLocked(); + acceptedSnapshot = BuildSnapshotLocked(); } else if (runAlreadyStarted) { @@ -456,16 +533,33 @@ private async Task DispatchQueuedSendAsync( } else if (_resetAwaitingUserMessage.Contains(threadId)) { + RemoveQueuedRunMappingByRunIdLocked(threadId, request.SendRunId); _resetLocalSendWithoutRunVersions[threadId] = dispatch.ResetVersion; _resetLocalSendWithoutRunStartSequences[threadId] = dispatch.StartedLifecycleSequence; TryOpenResetGateFromPendingLifecycleLocked(threadId, acceptedRunId: null); + if (PromoteQueuedMessageLocked(threadId, request.Id)) + { + acceptedSnapshot = BuildSnapshotLocked(); + } + } + else if (PromoteQueuedMessageLocked(threadId, request.Id)) + { + RemoveQueuedRunMappingByRunIdLocked(threadId, request.SendRunId); + acceptedSnapshot = BuildSnapshotLocked(); } } - if (acceptedRunAlreadyStartedSnapshot is not null) + if (acceptedSnapshot is not null) + { + Publish(acceptedSnapshot); + } + if (requeuedSnapshot is not null) + { + Publish(requeuedSnapshot); + } + if (retryDeferredSend) { - Publish(acceptedRunAlreadyStartedSnapshot); - ReplayDeferredAssistantMessage(threadId); + ScheduleQueuedSendDrain(threadId, deferredRetryDelay); } if (staleRunIdToAbort is not null) @@ -493,9 +587,10 @@ private async Task DispatchQueuedSendAsync( sendStillCurrent = GetResetVersionLocked(threadId) == dispatch.ResetVersion; if (sendStillCurrent) { - RemovePendingLocalEchoLocked(threadId, request.Text); + RemovePendingLocalEchoLocked(threadId, request.Id); MarkQueuedMessageFailedLocked(threadId, request.Id, ex.Message); RemoveQueuedSendRequestLocked(threadId, request.Id); + RemoveQueuedRunMappingByMessageIdLocked(threadId, request.Id); if (!HasSendingQueuedMessagesLocked(threadId)) _locallyInitiatedThreads.Remove(threadId); failureSnapshot = ApplyEventLocked( @@ -512,6 +607,7 @@ private async Task DispatchQueuedSendAsync( if (!sendStillCurrent) return; + Logger.Warn($"[Queue] chat.send failed threadId='{threadId}' queuedMessageId='{request.Id}' sendRunId='{request.SendRunId}': {ex.Message}"); // Surface as an error in the timeline + notification, while the // failed queue card keeps the attempted text visible for retry/edit. Publish(failureSnapshot!); @@ -1556,9 +1652,9 @@ public ValueTask DisposeAsync() _localInlineApprovals.Clear(); _queuedMessages.Clear(); _queuedSendRequests.Clear(); + _queuedDrainScheduledThreads.Clear(); _queuedMessageIdsByRunId.Clear(); _terminalRunIdsByThread.Clear(); - _deferredAssistantMessages.Clear(); _localSentTexts.Clear(); _locallyInitiatedThreads.Clear(); _resetSubmittedLocalEchoTexts.Clear(); @@ -1658,8 +1754,8 @@ private void OnStatusChanged(object? sender, ConnectionStatus status) _localSentTexts.Clear(); _queuedMessages.Clear(); _queuedSendRequests.Clear(); + _queuedDrainScheduledThreads.Clear(); _assistantFallbackPromotedThreads.Clear(); - _deferredAssistantMessages.Clear(); _queuedMessageIdsByRunId.Clear(); _terminalRunIdsByThread.Clear(); _resetSubmittedLocalEchoTexts.Clear(); @@ -1719,6 +1815,7 @@ private void OnSessionsUpdated(object? sender, SessionInfo[] sessions) { ChatDataSnapshot snapshot; string[] newThreadsToLoad; + string[] queuedThreadsToDrain; lock (_gate) { var previousUsage = _sessions @@ -1754,10 +1851,12 @@ private void OnSessionsUpdated(object? sender, SessionInfo[] sessions) toLoad.Add(key); } newThreadsToLoad = toLoad.Count > 0 ? toLoad.ToArray() : Array.Empty(); + queuedThreadsToDrain = _queuedMessages.Keys.ToArray(); } else { newThreadsToLoad = Array.Empty(); + queuedThreadsToDrain = Array.Empty(); } } Publish(snapshot); @@ -1766,6 +1865,10 @@ private void OnSessionsUpdated(object? sender, SessionInfo[] sessions) { _ = LoadHistoryAsync(threadId, force: false); } + foreach (var threadId in queuedThreadsToDrain) + { + TryDispatchNextQueuedSend(threadId); + } } private void OnSessionCommandCompleted(object? sender, SessionCommandResult result) @@ -1938,7 +2041,6 @@ private void OnChatMessageReceived(object? sender, ChatMessageInfo message) if (resetLocalEchoSnapshot is not null) { Publish(resetLocalEchoSnapshot); - ReplayDeferredAssistantMessage(msgThreadId); } if (requestRemoteBackfillAfterReset) _ = FetchRemoteUserMessageAsync(msgThreadId, openResetGateOnSuccess: true); @@ -2025,7 +2127,6 @@ private void OnChatMessageReceived(object? sender, ChatMessageInfo message) if (echoSnapshot is not null) { Publish(echoSnapshot); - ReplayDeferredAssistantMessage(msgThreadId); } return; } @@ -2100,16 +2201,10 @@ private void OnChatMessageReceived(object? sender, ChatMessageInfo message) cappedAssistantText, message.OpenClawId, message.OpenClawSeq); - if (assistantDisposition == AssistantQueueFrameDisposition.Defer) - _deferredAssistantMessages[threadId] = message; - else - _deferredAssistantMessages.Remove(threadId); } if (assistantDisposition != AssistantQueueFrameDisposition.Render) { - Logger.Debug(assistantDisposition == AssistantQueueFrameDisposition.Defer - ? $"[Queue] Deferring identity-less assistant frame until the queued user boundary is confirmed threadId='{threadId}'" - : $"[Queue] Dropping retransmitted assistant frame before queued user boundary threadId='{threadId}'"); + Logger.Debug($"[Queue] Dropping retransmitted assistant frame around queued user boundary threadId='{threadId}'"); return; } @@ -2186,7 +2281,7 @@ private void OnChatMessageReceived(object? sender, ChatMessageInfo message) ApplyEventAndPublish(threadId, new ChatTurnEndEvent()); RaiseNotification(new ChatProviderNotification( ChatProviderNotificationKind.TurnComplete, threadId, LocalizationHelper.GetString("Chat_Notification_AssistantReplied"))); - TryDispatchNextQueuedSend(threadId); + ScheduleQueuedSendDrain(threadId); } } @@ -2293,7 +2388,7 @@ private void OnAgentEventReceived(object? sender, AgentEventInfo evt) if (suppressRendering) { if (isTerminalRunEvent) - TryDispatchNextQueuedSend(threadId); + ScheduleQueuedSendDrain(threadId); return; } @@ -2370,7 +2465,7 @@ private void OnAgentEventReceived(object? sender, AgentEventInfo evt) } } if (isTerminalRunEvent) - TryDispatchNextQueuedSend(threadId); + ScheduleQueuedSendDrain(threadId); return; } @@ -2388,7 +2483,7 @@ private void OnAgentEventReceived(object? sender, AgentEventInfo evt) ApplyEventAndPublish(threadId, mapped, meta); if (isTerminalRunEvent) - TryDispatchNextQueuedSend(threadId); + ScheduleQueuedSendDrain(threadId); } private void RaiseKeylessEventDiagnosticOnce() @@ -2475,7 +2570,8 @@ private void UpdateActiveRunId(AgentEventInfo evt, string threadId) // sends can queue runs behind the current one; treating the // next lifecycle.start as remote would orphan those queued // cards and let assistant fallback promote the wrong item. - if (!HasSendingQueuedMessagesLocked(threadId)) + RemoveQueuedRunMappingByRunIdLocked(threadId, evt.RunId); + if (!HasPendingQueuedMessagesLocked(threadId)) _locallyInitiatedThreads.Remove(threadId); // Edge case: if we have pending aborts but never saw lifecycle.start @@ -2504,6 +2600,7 @@ private void UpdateActiveRunId(AgentEventInfo evt, string threadId) _abortedRunIds.Remove(evt.RunId); _activeRunIds.Remove(threadId); _activeRunStartSequences.Remove(threadId); + RemoveQueuedRunMappingByRunIdLocked(threadId, evt.RunId); } } } @@ -2592,27 +2689,37 @@ private bool TryConsumeLocalEchoLocked(string threadId, Queue que return false; } - var kept = new Queue(queue.Count); var matched = false; - while (queue.Count > 0) + string matchedMessageId = string.Empty; + var pendingEchoes = queue.ToArray(); + queue.Clear(); + foreach (var pending in pendingEchoes) { - var pending = queue.Dequeue(); - if (!matched && string.Equals(pending.Text, text, StringComparison.Ordinal)) - { - queuedMessageId = pending.QueuedMessageId; - matched = true; + if (matched || !string.Equals(pending.Text, text, StringComparison.Ordinal)) continue; - } - kept.Enqueue(pending); + queuedMessageId = pending.QueuedMessageId; + matchedMessageId = pending.QueuedMessageId; + matched = true; } + var kept = new Queue(pendingEchoes.Length); if (!matched) { + foreach (var pending in pendingEchoes) + kept.Enqueue(pending); StoreLocalEchoQueueLocked(threadId, kept); return false; } + foreach (var pending in pendingEchoes) + { + if (string.Equals(pending.QueuedMessageId, matchedMessageId, StringComparison.Ordinal)) + continue; + + kept.Enqueue(pending); + } + StoreLocalEchoQueueLocked(threadId, kept); return true; } @@ -2721,7 +2828,7 @@ private bool CanSendDirectlyLocked(string threadId) return false; if (_timelines.TryGetValue(threadId, out var timeline) && timeline.TurnActive) return false; - return !_queuedMessages.TryGetValue(threadId, out var queuedMessages) || queuedMessages.Count == 0; + return !HasPendingQueuedMessagesLocked(threadId); } private bool CanClearAssistantFallbackPromotionLocked(string threadId) @@ -2756,11 +2863,16 @@ private QueuedSendDispatch StartDirectSendLocked(QueuedSendRequest request) sessionId, resetVersion, startedLifecycleSequence, - startedRunStartSequence); + startedRunStartSequence, + StartedDirectly: true); } - private QueuedSendDispatch? TryStartNextQueuedSendLocked(string threadId, bool requireConnected) + private QueuedSendDispatch? TryStartNextQueuedSendLocked( + string threadId, + bool requireConnected, + out TimeSpan? delayedRetry) { + delayedRetry = null; if (requireConnected && _status != ConnectionStatus.Connected) return null; if (_activeRunIds.ContainsKey(threadId)) @@ -2779,6 +2891,19 @@ private QueuedSendDispatch StartDirectSendLocked(QueuedSendRequest request) if (request is null) continue; + var now = DateTimeOffset.UtcNow; + if (request.DeferredAdmissionRetryAfter is { } retryAfter) + { + if (retryAfter > now) + { + delayedRetry = retryAfter - now; + return null; + } + + request = request with { DeferredAdmissionRetryAfter = null }; + AddQueuedSendRequestLocked(request); + } + // Each dispatched prompt gets one opportunity for assistant-frame // fallback promotion before its lifecycle/user echo arrives. _assistantFallbackPromotedThreads.Remove(threadId); @@ -2796,7 +2921,8 @@ private QueuedSendDispatch StartDirectSendLocked(QueuedSendRequest request) sessionId, resetVersion, startedLifecycleSequence, - startedRunStartSequence); + startedRunStartSequence, + StartedDirectly: false); } return null; @@ -2804,6 +2930,7 @@ private QueuedSendDispatch StartDirectSendLocked(QueuedSendRequest request) private void EnqueueLocalEchoLocked(string threadId, string text, string messageId) { + RemovePendingLocalEchoLocked(threadId, messageId); if (!_localSentTexts.TryGetValue(threadId, out var localEchoQueue)) { localEchoQueue = new Queue(); @@ -2819,11 +2946,12 @@ private void TryDispatchNextQueuedSend(string threadId) { ChatDataSnapshot? snapshot = null; QueuedSendDispatch? dispatch; + TimeSpan? delayedRetry; lock (_gate) { if (_disposed) return; - dispatch = TryStartNextQueuedSendLocked(threadId, requireConnected: true); + dispatch = TryStartNextQueuedSendLocked(threadId, requireConnected: true, out delayedRetry); if (dispatch is not null) snapshot = BuildSnapshotLocked(); } @@ -2832,6 +2960,56 @@ private void TryDispatchNextQueuedSend(string threadId) Publish(snapshot); if (dispatch is not null) _ = DispatchQueuedSendAsync(dispatch, rethrow: false); + else if (delayedRetry is { } delay) + ScheduleQueuedSendDrain(threadId, delay); + } + + private void ScheduleQueuedSendDrain(string threadId) + => ScheduleQueuedSendDrain(threadId, DeferredQueueDrainDelay); + + private void ScheduleQueuedSendDrain(string threadId, TimeSpan delay) + { + lock (_gate) + { + if (_disposed || !_queuedMessages.ContainsKey(threadId)) + return; + if (!_queuedDrainScheduledThreads.Add(threadId)) + return; + } + + _ = Task.Run(async () => + { + try + { + await Task.Delay(delay).ConfigureAwait(false); + } + finally + { + lock (_gate) + { + _queuedDrainScheduledThreads.Remove(threadId); + } + } + + try + { + TryDispatchNextQueuedSend(threadId); + } + catch (Exception ex) + { + Logger.Warn($"[Queue] Scheduled queued send drain failed for threadId='{threadId}': {ex.Message}"); + } + }); + } + + private static bool IsDeferredAdmissionStatus(string? status) => + string.Equals(status, "in_flight", StringComparison.OrdinalIgnoreCase); + + private static TimeSpan DeferredAdmissionRetryDelay(int retryCount) + { + var exponent = Math.Min(Math.Max(retryCount - 1, 0), 5); + var delayMs = DeferredQueueDrainDelay.TotalMilliseconds * (1 << exponent); + return TimeSpan.FromMilliseconds(Math.Min(delayMs, MaxDeferredAdmissionRetryDelay.TotalMilliseconds)); } private void TrackQueuedMessageRunLocked(string threadId, string runId, string messageId) @@ -2857,10 +3035,42 @@ private bool RemoveQueuedMessageLocked(string threadId, string messageId) RemoveQueuedSendRequestLocked(threadId, messageId); } if (list.Count == 0) + { _queuedMessages.Remove(threadId); + ClearQueuedDrainScheduleLocked(threadId); + ClearLocallyInitiatedIfIdleLocked(threadId); + } return removed; } + private bool CancelQueuedMessageLocked(string threadId, string messageId) + { + if (!_queuedMessages.TryGetValue(threadId, out var list)) + return false; + + var index = list.FindIndex(message => string.Equals(message.Id, messageId, StringComparison.Ordinal)); + if (index < 0) + return false; + + if (list[index].SendState == ChatQueuedMessageSendState.Sending) + return false; + + list.RemoveAt(index); + RemovePendingLocalEchoLocked(threadId, messageId); + RemoveQueuedRunMappingByMessageIdLocked(threadId, messageId); + RemoveQueuedSendRequestLocked(threadId, messageId); + if (list.Count == 0) + { + _queuedMessages.Remove(threadId); + ClearQueuedDrainScheduleLocked(threadId); + ClearLocallyInitiatedIfIdleLocked(threadId); + } + return true; + } + + private void ClearQueuedDrainScheduleLocked(string threadId) + => _queuedDrainScheduledThreads.Remove(threadId); + private bool PromoteQueuedMessageLocked( string threadId, string messageId, @@ -2890,13 +3100,27 @@ private bool PromoteQueuedMessageLocked( list.RemoveAt(index); _assistantFallbackPromotedThreads.Add(threadId); - RemoveQueuedRunMappingByMessageIdLocked(threadId, messageId); RemoveQueuedSendRequestLocked(threadId, messageId); if (list.Count == 0) + { _queuedMessages.Remove(threadId); + ClearQueuedDrainScheduleLocked(threadId); + } return true; } + private void ClearLocallyInitiatedIfIdleLocked(string threadId) + { + if (_activeRunIds.ContainsKey(threadId)) + return; + if (_timelines.TryGetValue(threadId, out var timeline) && timeline.TurnActive) + return; + if (HasPendingQueuedMessagesLocked(threadId)) + return; + + _locallyInitiatedThreads.Remove(threadId); + } + private bool ReconcileQueuedMessageEchoLocked( string threadId, string messageId, @@ -2942,6 +3166,25 @@ private void RemoveQueuedRunMappingByMessageIdLocked(string threadId, string mes _queuedMessageIdsByRunId.Remove(threadId); } + private void RemoveQueuedRunMappingByRunIdLocked(string threadId, string runId) + { + if (!_queuedMessageIdsByRunId.TryGetValue(threadId, out var byRunId)) + return; + + if (byRunId.TryGetValue(runId, out var messageId)) + { + foreach (var aliasRunId in byRunId.Where(kvp => kvp.Value == messageId).Select(kvp => kvp.Key).ToArray()) + byRunId.Remove(aliasRunId); + } + else + { + byRunId.Remove(runId); + } + + if (byRunId.Count == 0) + _queuedMessageIdsByRunId.Remove(threadId); + } + private void MarkQueuedMessageFailedLocked(string threadId, string messageId, string error) { if (!_queuedMessages.TryGetValue(threadId, out var list)) @@ -2961,6 +3204,81 @@ private void MarkQueuedMessageFailedLocked(string threadId, string messageId, st } } + private bool RequeueDeferredAdmissionLocked(string threadId, string messageId, out TimeSpan retryDelay) + { + retryDelay = DeferredQueueDrainDelay; + var hasActiveRun = _activeRunIds.ContainsKey(threadId); + if (!_queuedMessages.TryGetValue(threadId, out var list)) + return false; + + for (var i = 0; i < list.Count; i++) + { + if (list[i].Id != messageId || + list[i].SendState != ChatQueuedMessageSendState.Sending) + { + continue; + } + + var retryCount = IncrementDeferredAdmissionRetryCountLocked(threadId, messageId); + if (retryCount > MaxDeferredAdmissionRetries) + { + throw new InvalidOperationException( + $"Gateway kept chat.send status in_flight after {MaxDeferredAdmissionRetries} retries."); + } + + list[i] = list[i] with + { + SendState = ChatQueuedMessageSendState.Queued, + ErrorText = null + }; + retryDelay = DeferredAdmissionRetryDelay(retryCount); + SetDeferredAdmissionRetryAfterLocked(threadId, messageId, DateTimeOffset.UtcNow + retryDelay); + _assistantFallbackPromotedThreads.Remove(threadId); + if (!hasActiveRun) + { + _timelines[threadId] = ChatTimelineReducer.Apply( + GetOrCreateTimelineLocked(threadId), + new ChatTurnEndEvent()); + } + return true; + } + + return false; + } + + private void SetDeferredAdmissionRetryAfterLocked(string threadId, string messageId, DateTimeOffset retryAfter) + { + if (!_queuedSendRequests.TryGetValue(threadId, out var requests)) + return; + + for (var i = 0; i < requests.Count; i++) + { + if (!string.Equals(requests[i].Id, messageId, StringComparison.Ordinal)) + continue; + + requests[i] = requests[i] with { DeferredAdmissionRetryAfter = retryAfter }; + return; + } + } + + private int IncrementDeferredAdmissionRetryCountLocked(string threadId, string messageId) + { + if (!_queuedSendRequests.TryGetValue(threadId, out var requests)) + return MaxDeferredAdmissionRetries + 1; + + for (var i = 0; i < requests.Count; i++) + { + if (!string.Equals(requests[i].Id, messageId, StringComparison.Ordinal)) + continue; + + var retryCount = requests[i].DeferredAdmissionRetryCount + 1; + requests[i] = requests[i] with { DeferredAdmissionRetryCount = retryCount }; + return retryCount; + } + + return MaxDeferredAdmissionRetries + 1; + } + private void ClearQueuedMessageOnLocalTurnStart(AgentEventInfo evt, string threadId) { if (!IsLifecycleStart(evt)) @@ -2976,7 +3294,6 @@ private void ClearQueuedMessageOnLocalTurnStart(AgentEventInfo evt, string threa if (snapshot is not null) { Publish(snapshot); - ReplayDeferredAssistantMessage(threadId); } } @@ -2988,12 +3305,9 @@ private bool TryPromoteQueuedMessageOnLocalTurnStartLocked(AgentEventInfo evt, s var runId = evt.RunId; if (!string.IsNullOrEmpty(runId) && _queuedMessageIdsByRunId.TryGetValue(threadId, out var byRunId) && - byRunId.Remove(runId, out var queuedMessageId)) + byRunId.TryGetValue(runId, out var queuedMessageId)) { - var removed = PromoteQueuedMessageLocked(threadId, queuedMessageId); - if (byRunId.Count == 0) - _queuedMessageIdsByRunId.Remove(threadId); - return removed; + return PromoteQueuedMessageLocked(threadId, queuedMessageId); } if (string.IsNullOrEmpty(runId) && @@ -3005,28 +3319,22 @@ private bool TryPromoteQueuedMessageOnLocalTurnStartLocked(AgentEventInfo evt, s return false; } - private void RemovePendingLocalEchoLocked(string threadId, string text) + private void RemovePendingLocalEchoLocked(string threadId, string messageId) { if (!_localSentTexts.TryGetValue(threadId, out var queue)) return; var kept = new Queue(queue.Count); - var removed = false; while (queue.Count > 0) { var pending = queue.Dequeue(); - if (!removed && string.Equals(pending.Text, text, StringComparison.Ordinal)) - { - removed = true; + if (string.Equals(pending.QueuedMessageId, messageId, StringComparison.Ordinal)) continue; - } + kept.Enqueue(pending); } - if (kept.Count == 0) - _localSentTexts.Remove(threadId); - else - _localSentTexts[threadId] = kept; + StoreLocalEchoQueueLocked(threadId, kept); } /// @@ -4393,10 +4701,10 @@ private ResetClearPersistence ClearThreadHistoryAfterResetLocked(string threadId _localSentTexts.Remove(threadId); _queuedMessages.Remove(threadId); _queuedSendRequests.Remove(threadId); + ClearQueuedDrainScheduleLocked(threadId); _queuedMessageIdsByRunId.Remove(threadId); _terminalRunIdsByThread.Remove(threadId); _assistantFallbackPromotedThreads.Remove(threadId); - _deferredAssistantMessages.Remove(threadId); _resetAcceptedRunIds.Remove(threadId); _resetLocalSendWithoutRunVersions.Remove(threadId); _resetLocalSendWithoutRunStartSequences.Remove(threadId); @@ -4606,6 +4914,13 @@ private AssistantQueueFrameDisposition ClassifyAssistantQueueFrameLocked( return AssistantQueueFrameDisposition.Drop; } + if (string.IsNullOrEmpty(gatewayMessageId) && + openClawSeq is null && + IsIdentitylessAssistantRetransmitAcrossLocalUserBoundaryLocked(threadId, assistantText)) + { + return AssistantQueueFrameDisposition.Drop; + } + if (!_locallyInitiatedThreads.Contains(threadId) || !TryGetSingleSendingQueuedMessageLocked(threadId, out _) || _activeRunIds.ContainsKey(threadId) || @@ -4623,7 +4938,9 @@ private AssistantQueueFrameDisposition ClassifyAssistantQueueFrameLocked( if (entry.IsStreaming || !string.Equals(entry.Text, assistantText, StringComparison.Ordinal)) return AssistantQueueFrameDisposition.Render; if (string.IsNullOrEmpty(gatewayMessageId) && openClawSeq is null) - return AssistantQueueFrameDisposition.Defer; + // In this queue-boundary window, an identity-less same-text frame cannot be tied + // to the queued prompt; replaying it can attach stale output to the next prompt. + return AssistantQueueFrameDisposition.Drop; if (!_entryMeta.TryGetValue(threadId, out var threadMeta) || !threadMeta.TryGetValue(entry.Id, out var existing)) { @@ -4642,6 +4959,42 @@ private AssistantQueueFrameDisposition ClassifyAssistantQueueFrameLocked( return AssistantQueueFrameDisposition.Render; } + private bool IsIdentitylessAssistantRetransmitAcrossLocalUserBoundaryLocked(string threadId, string assistantText) + { + if (!_locallyInitiatedThreads.Contains(threadId) || + _activeRunIds.ContainsKey(threadId) || + !_timelines.TryGetValue(threadId, out var timeline) || + !_entryMeta.TryGetValue(threadId, out var threadMeta)) + { + return false; + } + + var sawLatestLocalUserBoundary = false; + for (var i = timeline.Entries.Count - 1; i >= 0; i--) + { + var entry = timeline.Entries[i]; + if (!sawLatestLocalUserBoundary) + { + if (entry.Kind == ChatTimelineItemKind.Assistant) + return false; + if (entry.Kind == ChatTimelineItemKind.User && + threadMeta.TryGetValue(entry.Id, out var meta) && + meta.IsLocalQueuedSend) + { + sawLatestLocalUserBoundary = true; + } + continue; + } + + if (entry.Kind == ChatTimelineItemKind.Assistant) + return !entry.IsStreaming && string.Equals(entry.Text, assistantText, StringComparison.Ordinal); + if (entry.Kind == ChatTimelineItemKind.User) + return false; + } + + return false; + } + private bool IsIdentifiedCompletedAssistantDuplicateLocked( string threadId, string assistantText, @@ -4689,22 +5042,14 @@ openClawSeq is not null && return false; } - private void ReplayDeferredAssistantMessage(string threadId) - { - ChatMessageInfo? deferred; - lock (_gate) - { - if (!_deferredAssistantMessages.Remove(threadId, out deferred)) - return; - } - - OnChatMessageReceived(this, deferred); - } - private bool HasSendingQueuedMessagesLocked(string threadId) => _queuedMessages.TryGetValue(threadId, out var queued) && queued.Any(message => message.SendState == ChatQueuedMessageSendState.Sending); + private bool HasPendingQueuedMessagesLocked(string threadId) + => _queuedMessages.TryGetValue(threadId, out var queued) && + queued.Any(message => message.SendState is ChatQueuedMessageSendState.Queued or ChatQueuedMessageSendState.Sending); + private bool TryGetSingleSendingQueuedMessageLocked(string threadId, out ChatQueuedMessage message) { message = default!; diff --git a/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatRoot.cs b/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatRoot.cs index 232846bee..4c0cc5a0b 100644 --- a/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatRoot.cs +++ b/src/OpenClaw.Tray.WinUI/Chat/OpenClawChatRoot.cs @@ -301,6 +301,8 @@ Element BuildLoadingElement() { queuedMessages = queuedForThread.ToArray(); } + var hasPendingQueuedSend = queuedMessages.Any(message => + message.SendState is ChatQueuedMessageSendState.Queued or ChatQueuedMessageSendState.Sending); var entries = (IReadOnlyList)timeline.Entries; var connectedRaw = snapshot.ConnectionStatus; @@ -567,6 +569,7 @@ Element BuildLoadingElement() CurrentModel: composerThread.Model, CurrentModelProvider: composerThread.ModelProvider, CurrentThinkingLevel: composerThread.ThinkingLevel, + MessageOptionsDisabled: turnActiveOverride || hasPendingQueuedSend, ModelChoices: snapshot.ModelChoices, OnSend: (msg, attachments) => { @@ -587,6 +590,7 @@ Element BuildLoadingElement() OnAttachClick: _onAttachClick, PendingAttachments: pendingAttachments.Value, QueuedMessages: queuedMessages, + OnQueuedMessageCancel: queuedMessageId => RunFireAndForget(ct => _provider.CancelQueuedMessageAsync(composerThread.Id!, queuedMessageId, ct)), OnAttachmentRemoved: attachment => SetPendingAttachments(RemoveAttachment(pendingAttachmentsRef.Current, attachment)), IsSpeakerMuted: speakerMuted.Value, OnSpeakerToggle: () => diff --git a/src/OpenClaw.Tray.WinUI/Chat/OpenClawComposer.cs b/src/OpenClaw.Tray.WinUI/Chat/OpenClawComposer.cs index e47b3961c..75d64696a 100644 --- a/src/OpenClaw.Tray.WinUI/Chat/OpenClawComposer.cs +++ b/src/OpenClaw.Tray.WinUI/Chat/OpenClawComposer.cs @@ -50,6 +50,7 @@ public record OpenClawComposerProps( string? CurrentModel, string? CurrentModelProvider, string? CurrentThinkingLevel, + bool MessageOptionsDisabled, Action> OnSend, Action OnStop, Action OnChannelChanged, @@ -60,6 +61,7 @@ public record OpenClawComposerProps( Action? OnAttachClick = null, IReadOnlyList? PendingAttachments = null, IReadOnlyList? QueuedMessages = null, + Action? OnQueuedMessageCancel = null, Action? OnAttachmentRemoved = null, bool IsSpeakerMuted = false, Action? OnSpeakerToggle = null, @@ -233,8 +235,8 @@ public override Element Render() }; var sendActionRef = UseRef(sendAction); sendActionRef.Current = sendAction; - var isConnected = Props.ConnectionState == "connected"; + var messageOptionControlsEnabled = !Props.MessageOptionsDisabled; var placeholder = Props.ConnectionState switch { "connected" => LocalizationHelper.GetString("Chat_Composer_Placeholder_Connected"), @@ -371,6 +373,7 @@ public override Element Render() CornerRadius = composerCornerRadius, HorizontalAlignment = HorizontalAlignment.Stretch, VerticalAlignment = VerticalAlignment.Center, + IsEnabled = messageOptionControlsEnabled, }; Microsoft.UI.Xaml.Automation.AutomationProperties.SetName( cb, @@ -430,6 +433,7 @@ public override Element Render() cb.Padding = new Thickness(8, 0, 4, 0); cb.CornerRadius = composerCornerRadius; cb.HorizontalAlignment = HorizontalAlignment.Stretch; + cb.IsEnabled = messageOptionControlsEnabled; }).VAlign(VerticalAlignment.Center); Element dropdownsRow = Grid([GridSize.Star(1.2), GridSize.Star(), GridSize.Star(0.62)], [GridSize.Auto], @@ -926,8 +930,58 @@ Element RenderQueuedMessages() CultureInfo.CurrentCulture, LocalizationHelper.GetString("Chat_Composer_QueuedCountFormat"), queuedMessages.Count); + Element RenderQueueCancelButton(ChatQueuedMessage message, int index) + { + if (message.SendState == ChatQueuedMessageSendState.Sending || Props.OnQueuedMessageCancel is null) + return Empty(); - Element RenderQueuedCard(ChatQueuedMessage message) + var failed = message.SendState == ChatQueuedMessageSendState.Failed; + var cancelTooltip = LocalizationHelper.GetString(failed + ? "Chat_Composer_QueuedMessageRemoveFailed" + : "Chat_Composer_QueuedMessageCancel"); + var buttonName = string.Format( + CultureInfo.CurrentCulture, + LocalizationHelper.GetString(failed + ? "Chat_Composer_QueuedMessageRemoveFailedAutomationFormat" + : "Chat_Composer_QueuedMessageCancelAutomationFormat"), + index + 1, + FormatQueuedMessageAutomationSnippet(message.Text)); + return Button( + TextBlock("\uE711") + .Set(t => + { + t.FontFamily = FluentIconCatalog.SymbolThemeFontFamily; + t.FontSize = 11; + t.HorizontalAlignment = HorizontalAlignment.Center; + t.VerticalAlignment = VerticalAlignment.Center; + }), + () => Props.OnQueuedMessageCancel?.Invoke(message.Id)) + .Set(b => + { + b.Width = 28; + b.Height = 28; + b.MinWidth = 0; + b.MinHeight = 0; + b.Padding = new Thickness(0); + b.CornerRadius = new CornerRadius(4); + Microsoft.UI.Xaml.Automation.AutomationProperties.SetAutomationId( + b, + $"{(failed ? "ChatQueuedMessageRemoveFailed" : "ChatQueuedMessageCancel")}_{message.Id}"); + }) + .Resources(r => + { + r.Set("ButtonBackground", new SolidColorBrush(Colors.Transparent)); + r.Set("ButtonBackgroundPointerOver", Ref("SubtleFillColorSecondaryBrush")); + r.Set("ButtonBackgroundPressed", Ref("SubtleFillColorTertiaryBrush")); + r.Set("ButtonBorderBrush", new SolidColorBrush(Colors.Transparent)); + r.Set("ButtonBorderBrushPointerOver", new SolidColorBrush(Colors.Transparent)); + r.Set("ButtonBorderBrushPressed", new SolidColorBrush(Colors.Transparent)); + }) + .AutomationName(buttonName) + .SetToolTip(cancelTooltip); + } + + Element RenderQueuedCard(ChatQueuedMessage message, int index) { var failed = message.SendState == ChatQueuedMessageSendState.Failed; var automationName = string.Format( @@ -961,17 +1015,24 @@ Element RenderQueuedCard(ChatQueuedMessage message) } } + var messageBody = VStack(6, + stateLabel, + TextBlock(message.Text) + .Set(t => + { + t.TextWrapping = TextWrapping.Wrap; + t.IsTextSelectionEnabled = true; + t.Foreground = textFg; + }), + details + ); + return Border( - VStack(6, - stateLabel, - TextBlock(message.Text) - .Set(t => - { - t.TextWrapping = TextWrapping.Wrap; - t.IsTextSelectionEnabled = true; - t.Foreground = textFg; - }), - details + Grid([GridSize.Star(), GridSize.Auto], [GridSize.Auto], + messageBody.Grid(row: 0, column: 0), + RenderQueueCancelButton(message, index) + .Grid(row: 0, column: 1) + .VAlign(VerticalAlignment.Top) ) ) .Background(cardBg) @@ -1378,6 +1439,15 @@ private static double ComputeQueuedMessagesMaxHeight(bool isCompact, double? ava ExpandedQueuedMessagesMaxHeight); } + private static string FormatQueuedMessageAutomationSnippet(string text) + { + var normalized = string.Join(" ", text.Split((char[]?)null, StringSplitOptions.RemoveEmptyEntries)); + const int maxLength = 80; + return normalized.Length <= maxLength + ? normalized + : normalized[..maxLength] + "…"; + } + private const int SlashMenuMaxItems = 8; /// diff --git a/src/OpenClaw.Tray.WinUI/Strings/en-us/Resources.resw b/src/OpenClaw.Tray.WinUI/Strings/en-us/Resources.resw index a332f9b8b..09429b6c2 100644 --- a/src/OpenClaw.Tray.WinUI/Strings/en-us/Resources.resw +++ b/src/OpenClaw.Tray.WinUI/Strings/en-us/Resources.resw @@ -3062,6 +3062,18 @@ Use one of these options: Failed + + Cancel queued message + + + Cancel queued message {0}: {1} + + + Remove failed queued message + + + Remove failed queued message {0}: {1} + Copy message diff --git a/src/OpenClaw.Tray.WinUI/Strings/fr-fr/Resources.resw b/src/OpenClaw.Tray.WinUI/Strings/fr-fr/Resources.resw index daacb46bb..11bbab385 100644 --- a/src/OpenClaw.Tray.WinUI/Strings/fr-fr/Resources.resw +++ b/src/OpenClaw.Tray.WinUI/Strings/fr-fr/Resources.resw @@ -3015,6 +3015,18 @@ Utilisez l'une de ces options : Échec + + Annuler le message en file d’attente + + + Annuler le message en file d’attente {0} : {1} + + + Supprimer le message en file d’attente en échec + + + Supprimer le message en file d’attente en échec {0} : {1} + Copier le message diff --git a/src/OpenClaw.Tray.WinUI/Strings/nl-nl/Resources.resw b/src/OpenClaw.Tray.WinUI/Strings/nl-nl/Resources.resw index 027effb4e..cd4c980f7 100644 --- a/src/OpenClaw.Tray.WinUI/Strings/nl-nl/Resources.resw +++ b/src/OpenClaw.Tray.WinUI/Strings/nl-nl/Resources.resw @@ -3016,6 +3016,18 @@ Gebruik een van deze opties: Mislukt + + Bericht in wachtrij annuleren + + + Bericht {0} in wachtrij annuleren: {1} + + + Mislukt bericht in wachtrij verwijderen + + + Mislukt bericht {0} in wachtrij verwijderen: {1} + Bericht kopiëren diff --git a/src/OpenClaw.Tray.WinUI/Strings/zh-cn/Resources.resw b/src/OpenClaw.Tray.WinUI/Strings/zh-cn/Resources.resw index cb67066e2..f611240c9 100644 --- a/src/OpenClaw.Tray.WinUI/Strings/zh-cn/Resources.resw +++ b/src/OpenClaw.Tray.WinUI/Strings/zh-cn/Resources.resw @@ -3015,6 +3015,18 @@ 失败 + + 取消已排队消息 + + + 取消已排队消息 {0}:{1} + + + 移除失败的已排队消息 + + + 移除失败的已排队消息 {0}:{1} + 复制消息 diff --git a/src/OpenClaw.Tray.WinUI/Strings/zh-tw/Resources.resw b/src/OpenClaw.Tray.WinUI/Strings/zh-tw/Resources.resw index daf97d636..6466184a6 100644 --- a/src/OpenClaw.Tray.WinUI/Strings/zh-tw/Resources.resw +++ b/src/OpenClaw.Tray.WinUI/Strings/zh-tw/Resources.resw @@ -3015,6 +3015,18 @@ 失敗 + + 取消已排入佇列的訊息 + + + 取消已排入佇列的訊息 {0}:{1} + + + 移除失敗的已排入佇列訊息 + + + 移除失敗的已排入佇列訊息 {0}:{1} + 複製訊息 diff --git a/src/OpenClaw.WinNode.Cli/skill.md b/src/OpenClaw.WinNode.Cli/skill.md index bf61068cd..8df1b5909 100644 --- a/src/OpenClaw.WinNode.Cli/skill.md +++ b/src/OpenClaw.WinNode.Cli/skill.md @@ -366,11 +366,12 @@ Returns `{ url, credentialSource, usesSharedGatewayToken, hasTokenQuery }`. ### app.chat.snapshot Read the current native chat snapshot for local automation and diagnostics. -**READ-ALL:** returns recent chat text from the selected timeline. +**READ-ALL:** returns recent chat text from the selected timeline and outgoing +queue text. ``` {"threadId": "string"} // optional; "sessionKey" alias also accepted ``` -Returns `{ connectionStatus, defaultThreadId, composeTarget, threads, selectedTimeline }`. +Returns `{ connectionStatus, defaultThreadId, composeTarget, threads, queue, selectedTimeline }`. ### app.chat.send Send a message through the same native chat provider used by the Chat UI. @@ -391,6 +392,31 @@ When `threadId`/`sessionKey` is omitted this resets the current compose/default thread, which usually means the active chat session. Returns `{ reset, threadId, error? }`. +### app.chat.queue.list +List native chat outgoing queue entries. +**READ-ALL:** returns queued message text. +``` +{"threadId": "string"} // optional; "sessionKey" alias also accepted +``` +When `threadId`/`sessionKey` is omitted this returns all queued threads. Returns: +`{ defaultThreadId, requestedThreadId, totalCount, selectedThread, threads }` +where each thread has `{ threadId, count, messages }`, and each message has +`{ id, text, createdAt, sendState, errorText, canCancel }`. Queue message IDs +are ephemeral UI/provider IDs; read them from the current `app.chat.queue.list` +or `app.chat.snapshot` response and do not cache them across app lifetimes. + +### app.chat.queue.cancel +Cancel/remove one native chat outgoing queue entry before it is sent. +``` +{ + "queuedMessageId": "string", // required; from the current app.chat.queue.list or app.chat.snapshot queue + "threadId": "string" // required; "sessionKey" alias also accepted +} +``` +Only `Queued` and `Failed` entries can be removed. `Sending` entries may already +have reached the gateway and are not canceled. Returns +`{ canceled, threadId, queuedMessageId, remainingCount, error? }`. + ## Location (location.*) ### location.get diff --git a/tests/OpenClaw.Shared.Tests/AppCapabilityTests.cs b/tests/OpenClaw.Shared.Tests/AppCapabilityTests.cs index 1d2642ab4..1c71bf587 100644 --- a/tests/OpenClaw.Shared.Tests/AppCapabilityTests.cs +++ b/tests/OpenClaw.Shared.Tests/AppCapabilityTests.cs @@ -28,7 +28,8 @@ public void Commands_ContainsAllExpectedTools() var cap = new AppCapability(NullLogger.Instance); var expected = new[] { "app.navigate", "app.status", "app.sessions", "app.agents", "app.nodes", "app.config.get", "app.settings.get", "app.settings.set", "app.menu", "app.search", - "app.dashboard.url", "app.chat.snapshot", "app.chat.send", "app.chat.reset" }; + "app.dashboard.url", "app.chat.snapshot", "app.chat.send", "app.chat.reset", + "app.chat.queue.list", "app.chat.queue.cancel" }; foreach (var cmd in expected) { Assert.Contains(cmd, cap.Commands); @@ -202,4 +203,99 @@ public async Task ChatSend_WithoutMessage_ReturnsError() Assert.False(res.Ok); Assert.Contains("message", res.Error); } + + [Fact] + public async Task ChatQueueList_WithSessionKeyAlias_ForwardsThreadId() + { + var cap = new AppCapability(NullLogger.Instance); + string? capturedThreadId = null; + cap.ChatQueueListHandler = threadId => + { + capturedThreadId = threadId; + return Task.FromResult(new { totalCount = 1 }); + }; + + var req = new NodeInvokeRequest + { + Id = "1", + Command = "app.chat.queue.list", + Args = ParseArgs("{\"sessionKey\":\"agent:main:default\"}") + }; + + var res = await cap.ExecuteAsync(req); + + Assert.True(res.Ok); + Assert.Equal("agent:main:default", capturedThreadId); + } + + [Fact] + public async Task ChatQueueCancel_WithHandler_ForwardsThreadAndQueuedMessage() + { + var cap = new AppCapability(NullLogger.Instance); + string? capturedThreadId = null; + string? capturedQueuedMessageId = null; + cap.ChatQueueCancelHandler = (threadId, queuedMessageId) => + { + capturedThreadId = threadId; + capturedQueuedMessageId = queuedMessageId; + return Task.FromResult(new { canceled = true }); + }; + + var req = new NodeInvokeRequest + { + Id = "1", + Command = "app.chat.queue.cancel", + Args = ParseArgs("{\"threadId\":\"main\",\"queuedMessageId\":\"q7\"}") + }; + + var res = await cap.ExecuteAsync(req); + + Assert.True(res.Ok); + Assert.Equal("main", capturedThreadId); + Assert.Equal("q7", capturedQueuedMessageId); + } + + [Fact] + public async Task ChatQueueCancel_WithoutQueuedMessageId_ReturnsError() + { + var cap = new AppCapability(NullLogger.Instance); + cap.ChatQueueCancelHandler = (_, _) => Task.FromResult(new { canceled = true }); + + var req = new NodeInvokeRequest + { + Id = "1", + Command = "app.chat.queue.cancel", + Args = ParseArgs("{\"threadId\":\"main\"}") + }; + + var res = await cap.ExecuteAsync(req); + + Assert.False(res.Ok); + Assert.Contains("queuedMessageId", res.Error); + } + + [Fact] + public async Task ChatQueueCancel_WithoutThreadId_ReturnsError() + { + var cap = new AppCapability(NullLogger.Instance); + var handlerCalled = false; + cap.ChatQueueCancelHandler = (_, _) => + { + handlerCalled = true; + return Task.FromResult(new { canceled = true }); + }; + + var req = new NodeInvokeRequest + { + Id = "1", + Command = "app.chat.queue.cancel", + Args = ParseArgs("{\"queuedMessageId\":\"q7\"}") + }; + + var res = await cap.ExecuteAsync(req); + + Assert.False(res.Ok); + Assert.Contains("threadId", res.Error); + Assert.False(handlerCalled); + } } diff --git a/tests/OpenClaw.Shared.Tests/GatewayProtocolLiveRoundTripTests.cs b/tests/OpenClaw.Shared.Tests/GatewayProtocolLiveRoundTripTests.cs index 68d7d4992..b5e6469b7 100644 --- a/tests/OpenClaw.Shared.Tests/GatewayProtocolLiveRoundTripTests.cs +++ b/tests/OpenClaw.Shared.Tests/GatewayProtocolLiveRoundTripTests.cs @@ -176,6 +176,44 @@ public async Task CronRunDetailed_FallsBackToLegacyIdPayload_WhenJobIdPayloadIsR } } + [Fact] + public async Task ChatSend_UsesCallerProvidedIdempotencyKeyOnWire() + { + using var server = new LoopbackGatewayServer(); + server.OnMethod("chat.send", parameters => + { + Assert.Equal("agent:main:main", parameters.GetProperty("sessionKey").GetString()); + Assert.Equal("Hello", parameters.GetProperty("message").GetString()); + Assert.Equal("send-run-123", parameters.GetProperty("idempotencyKey").GetString()); + return new { runId = "send-run-123", sessionKey = "agent:main:main", status = "started" }; + }); + + var logger = new TestLogger(); + var client = new OpenClawGatewayClient( + server.WebSocketUrl, "test-token", logger, + tokenIsBootstrapToken: false, bootstrapPairAsNode: false, + identityPath: _identityDir); + + try + { + await ConnectAndWaitAsync(client, server); + + var result = await client.SendChatMessageForRunAsync( + "Hello", + sessionKey: "agent:main:main", + idempotencyKey: "send-run-123"); + + Assert.Equal("send-run-123", result.RunId); + Assert.Equal("started", result.Status); + var frame = await server.WaitFrameAsync("chat.send", occurrence: 0, timeoutMs: 20000); + Assert.Contains("\"idempotencyKey\":\"send-run-123\"", frame); + } + finally + { + await client.DisconnectAsync(); + } + } + private static void ConfigureResponders(LoopbackGatewayServer server) { // NOTE: intentionally NO hello-ok responder. The new methods only require diff --git a/tests/OpenClaw.Shared.Tests/ModelsTests.cs b/tests/OpenClaw.Shared.Tests/ModelsTests.cs index 14374ca9c..9f9f33c7b 100644 --- a/tests/OpenClaw.Shared.Tests/ModelsTests.cs +++ b/tests/OpenClaw.Shared.Tests/ModelsTests.cs @@ -5,6 +5,41 @@ namespace OpenClaw.Shared.Tests; +public class ChatSendResultTests +{ + [Theory] + [InlineData("failed")] + [InlineData("failure")] + [InlineData("error")] + [InlineData("rejected")] + [InlineData("denied")] + [InlineData("aborted")] + [InlineData("timeout")] + [InlineData("cancelled")] + [InlineData("canceled")] + public void IsTerminalFailure_ReturnsTrue_ForTerminalStatus(string status) + { + var result = new ChatSendResult { Status = status }; + + Assert.True(result.IsTerminalFailure); + Assert.True(ChatSendResult.IsFailureStatus(status)); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData("started")] + [InlineData("in_flight")] + [InlineData("ok")] + public void IsTerminalFailure_ReturnsFalse_ForNonTerminalStatus(string? status) + { + var result = new ChatSendResult { Status = status }; + + Assert.False(result.IsTerminalFailure); + Assert.False(ChatSendResult.IsFailureStatus(status)); + } +} + public class AgentActivityTests { [Fact] diff --git a/tests/OpenClaw.Tray.Tests/AppRefactorContractTests.cs b/tests/OpenClaw.Tray.Tests/AppRefactorContractTests.cs index 870abb364..e9ec3d0d3 100644 --- a/tests/OpenClaw.Tray.Tests/AppRefactorContractTests.cs +++ b/tests/OpenClaw.Tray.Tests/AppRefactorContractTests.cs @@ -230,6 +230,34 @@ public void AppMenu_StatusItemIncludesManagerSnapshotState() Assert.Contains("nodeError = snapshot?.NodeError", method); } + [Fact] + public void AppChatQueueMcpHandlers_AreWiredToAppCapability() + { + var source = ReadAppSources(); + var method = ExtractMethod(source, "WireAppCapabilityHandlers"); + + Assert.Contains("app.ChatQueueListHandler = ListQueuedChatMessagesForMcpAsync;", method); + Assert.Contains("app.ChatQueueCancelHandler = CancelQueuedChatMessageForMcpAsync;", method); + } + + [Fact] + public void AppChatSnapshot_IncludesQueuePayload() + { + var source = ReadAppSources(); + var snapshotMethod = ExtractMethod(source, "BuildChatSnapshotPayload"); + var queueMethod = ExtractMethod(source, "BuildChatQueuePayload"); + var queueMessageMethod = ExtractMethod(source, "ToMcpQueuedMessage"); + var cancelMethod = ExtractMethod(source, "CancelQueuedChatMessageForMcpAsync"); + + Assert.Contains("queue = BuildChatQueuePayload(snapshot, resolvedThreadId, filterToThread: false)", snapshotMethod); + Assert.Contains("snapshot.QueuedMessagesByThread", queueMethod); + Assert.Contains("sendState = message.SendState.ToString()", queueMessageMethod); + Assert.Contains("canCancel = CanCancelQueuedMessage(message)", queueMessageMethod); + Assert.Contains("canceled = await provider.CancelQueuedMessageAsync(resolvedThreadId, queuedMessageId)", cancelMethod); + Assert.Contains("Queued message is already sending and cannot be canceled", cancelMethod); + Assert.Contains("it may have started sending before cancellation was processed", cancelMethod); + } + [Fact] public void Startup_NodeOnlyReconnect_UsesNodeCredentialAndLegacyIdentityFallback() { @@ -950,7 +978,7 @@ private static string ExtractMethod(string source, string methodName) { var match = Regex.Match( source, - $@"(?m)^\s*(?:private|protected|public|internal)\s+(?:async\s+)?(?:Task(?:<[^>]+>)?|System\.Threading\.Tasks\.Task|void|bool|int|string\??|IntPtr|OpenClaw\.Connection\.GatewayCredential\?)\s+{Regex.Escape(methodName)}\s*\("); + $@"(?m)^\s*(?:private|protected|public|internal)\s+(?:static\s+)?(?:async\s+)?(?:Task(?:<[^>]+>)?|System\.Threading\.Tasks\.Task|void|bool|int|string\??|object\??|IntPtr|OpenClaw\.Connection\.GatewayCredential\?)\s+{Regex.Escape(methodName)}\s*\("); Assert.True(match.Success, $"Could not find method {methodName}."); var brace = source.IndexOf('{', match.Index); diff --git a/tests/OpenClaw.Tray.Tests/ChatTimelineRenderIdentityContractTests.cs b/tests/OpenClaw.Tray.Tests/ChatTimelineRenderIdentityContractTests.cs index a55adaa9a..40e8ed252 100644 --- a/tests/OpenClaw.Tray.Tests/ChatTimelineRenderIdentityContractTests.cs +++ b/tests/OpenClaw.Tray.Tests/ChatTimelineRenderIdentityContractTests.cs @@ -58,11 +58,22 @@ public void QueuedMessages_RenderInComposerAboveInput() Assert.Contains("QueuedMessagesByThread: queuedMessagesCopy", provider); Assert.Contains("snapshot.QueuedMessagesByThread", root); Assert.Contains("QueuedMessages: queuedMessages", root); + Assert.Contains("OnQueuedMessageCancel:", root); Assert.Contains("IReadOnlyList? QueuedMessages = null", composer); + Assert.Contains("Action? OnQueuedMessageCancel = null", composer); Assert.Contains("AvailableHeight: chatSurfaceHeight.Value", root); Assert.Contains("queuedPanel", composer); Assert.Contains("composerInput", composer); Assert.Contains("RenderQueuedMessages", composer); + Assert.Contains("Chat_Composer_QueuedMessageCancel", composer); + Assert.Contains("RenderQueueCancelButton", composer); + Assert.Contains("ChatQueuedMessageSendState.Sending", composer); + Assert.Contains("Chat_Composer_QueuedMessageCancelAutomationFormat", composer); + Assert.Contains("Chat_Composer_QueuedMessageRemoveFailed", composer); + Assert.Contains("Chat_Composer_QueuedMessageRemoveFailedAutomationFormat", composer); + Assert.Contains("ChatQueuedMessageRemoveFailed", composer); + Assert.Contains("ChatQueuedMessageCancel", composer); + Assert.Contains("}_{message.Id}", composer); Assert.Contains("Chat_Composer_QueuedCountFormat", composer); Assert.Contains("Chat_Composer_QueuedMessageAutomationFormat", composer); Assert.Contains("Chat_Composer_QueuedMessageFailedAutomationFormat", composer); @@ -77,6 +88,21 @@ public void QueuedMessages_RenderInComposerAboveInput() Assert.DoesNotContain("Steer", composer); } + [Fact] + public void Composer_DisablesMessageOptionDropdownsWhileTurnOrPendingQueueSendIsActive() + { + var composer = Read("src", "OpenClaw.Tray.WinUI", "Chat", "OpenClawComposer.cs"); + var root = Read("src", "OpenClaw.Tray.WinUI", "Chat", "OpenClawChatRoot.cs"); + + Assert.Contains("var messageOptionControlsEnabled = !Props.MessageOptionsDisabled;", composer); + Assert.Contains("MessageOptionsDisabled: turnActiveOverride || hasPendingQueuedSend", root); + Assert.Contains("message.SendState is ChatQueuedMessageSendState.Queued or ChatQueuedMessageSendState.Sending", root); + Assert.Equal(2, Regex.Matches(composer, "IsEnabled = messageOptionControlsEnabled").Count); + Assert.DoesNotMatch( + new Regex(@"var\s+channelCombo[\s\S]*?IsEnabled\s*=\s*messageOptionControlsEnabled[\s\S]*?//\s+── Model picker", RegexOptions.Multiline), + composer); + } + [Fact] public void Timeline_DoesNotRenderTemporaryDebugMetadata() { diff --git a/tests/OpenClaw.Tray.Tests/OpenClawChatDataProviderTests.cs b/tests/OpenClaw.Tray.Tests/OpenClawChatDataProviderTests.cs index d7a2d831d..8e821404e 100644 --- a/tests/OpenClaw.Tray.Tests/OpenClawChatDataProviderTests.cs +++ b/tests/OpenClaw.Tray.Tests/OpenClawChatDataProviderTests.cs @@ -17,6 +17,7 @@ private sealed class FakeBridge : IChatGatewayBridge public List SentSessionKeys { get; } = new(); public List SentSessionIds { get; } = new(); public List?> SentAttachments { get; } = new(); + public List SentIdempotencyKeys { get; } = new(); public Queue SendResults { get; } = new(); public List AbortedRunIds { get; } = new(); public Func? SendBehavior { get; set; } @@ -47,12 +48,18 @@ public Task ListCommandsAsync(CommandCatalogQuery? query = null) public Task SendChatMessageAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null) => SendChatMessageForRunAsync(message, sessionKey, sessionId, attachments); - public async Task SendChatMessageForRunAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null) + public async Task SendChatMessageForRunAsync( + string message, + string? sessionKey, + string? sessionId, + IReadOnlyList? attachments = null, + string? idempotencyKey = null) { SentMessages.Add(message); SentSessionKeys.Add(sessionKey); SentSessionIds.Add(sessionId); SentAttachments.Add(attachments?.ToArray()); + SentIdempotencyKeys.Add(idempotencyKey); if (SendBehavior is not null) await SendBehavior(message, sessionKey, sessionId); @@ -293,10 +300,11 @@ public async Task SendMessageAsync_DuringActiveTurn_QueuesFollowUpsLocallyUntilT }); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-1")); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); - var sending = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); - Assert.Equal("second", sending.Text); - Assert.Equal(ChatQueuedMessageSendState.Sending, sending.SendState); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-2")); @@ -324,23 +332,22 @@ public async Task AgentEvent_DuplicateTerminalForCompletedRun_DoesNotDispatchAdd Assert.Equal(new[] { "a" }, bridge.SentMessages); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-1")); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); Assert.Equal(new[] { "a", "b" }, bridge.SentMessages); bridge.RaiseAgent(MakeAgentEvent("job", """{"state":"done"}""", runId: "run-1")); + await Task.Delay(150); Assert.Equal(new[] { "a", "b" }, bridge.SentMessages); Assert.Collection( GetQueuedMessages(snapshots[^1], "main"), - sending => - { - Assert.Equal("b", sending.Text); - Assert.Equal(ChatQueuedMessageSendState.Sending, sending.SendState); - }, queued => { Assert.Equal("c", queued.Text); Assert.Equal(ChatQueuedMessageSendState.Queued, queued.SendState); }); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "b"); } [Fact] @@ -389,6 +396,7 @@ public async Task LifecycleStart_DoesNotClearNextQueuedMessageWhenEarlierEchoAlr State = "final" }); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-1")); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); Assert.Equal(new[] { "first queued", "second queued" }, bridge.SentMessages); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-2")); @@ -501,9 +509,10 @@ public async Task LifecycleEnd_KeepsLocalInitiatedStateWhenQueuedMessagesRemain( State = "final" }); - Assert.Equal("second", Assert.Single(GetQueuedMessages(snapshots[^1], "main")).Text); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); - Assert.DoesNotContain(snapshots[^1].Timelines["main"].Entries, e => + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => e.Kind == ChatTimelineItemKind.User && e.Text == "second"); Assert.Single(snapshots[^1].Timelines["main"].Entries, e => e.Kind == ChatTimelineItemKind.Assistant && e.Text == "first run response"); @@ -653,6 +662,66 @@ public async Task LifecycleEnd_IgnoresFailedQueuedMessagesWhenClearingLocalIniti Assert.True(historyCalls > 0); } + [Fact] + public async Task LifecycleEnd_IgnoresFailedQueuedMessagesAfterLaterAcceptedSend() + { + var historyCalls = 0; + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-a", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { Status = "failed", Error = "second failed" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-next", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "a"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-a")); + await provider.SendMessageAsync("main", "failed card"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-a")); + for (var i = 0; i < 20; i++) + { + if (GetQueuedMessages(snapshots[^1], "main").Any(q => q.SendState == ChatQueuedMessageSendState.Failed)) + break; + await Task.Delay(10); + } + + await provider.SendMessageAsync("main", "next local"); + Assert.Contains(GetQueuedMessages(snapshots[^1], "main"), q => + q.Text == "failed card" && + q.SendState == ChatQueuedMessageSendState.Failed); + Assert.Contains(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "next local"); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-next")); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-next")); + bridge.HistoryBehavior = _ => + { + historyCalls++; + return Task.FromResult(new ChatHistoryInfo + { + SessionKey = "main", + Messages = new[] + { + new ChatMessageInfo + { + SessionKey = "main", + Role = "user", + Text = "remote after completed local", + Ts = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), + OpenClawSeq = 31 + } + } + }); + }; + snapshots.Clear(); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "remote-run")); + + for (var i = 0; i < 20 && historyCalls == 0; i++) + await Task.Delay(10); + + Assert.True(historyCalls > 0); + } + [Fact] public async Task Status_ReconnectClearsUncorrelatedQueuedMessagesBeforeHistoryReload() { @@ -1102,11 +1171,15 @@ public async Task AgentEvent_JobError_DispatchesNextQueuedMessage() for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) await Task.Delay(10); + await WaitForConditionAsync(() => + GetQueuedMessages(snapshots[^1], "main").Count == 0 && + snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second") == 1); Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); - var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); - Assert.Equal("second", queued.Text); - Assert.Equal(ChatQueuedMessageSendState.Sending, queued.SendState); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second"); } [Fact] @@ -1506,6 +1579,7 @@ public async Task SessionResetCompletion_AbortsAndSuppressesLiveFramesFromPreRes await provider.SendMessageAsync("main", "c"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-a")); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); Assert.Equal(new[] { "a", "b" }, bridge.SentMessages); bridge.RaiseSessionCommandCompleted(new SessionCommandResult @@ -1613,6 +1687,7 @@ public async Task SessionResetCompletion_ShowsPersistedSubmittedRunFromForcedHis bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-a")); await provider.SendMessageAsync("main", "b"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-a")); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); Assert.Equal(new[] { "a", "b" }, bridge.SentMessages); bridge.RaiseSessionCommandCompleted(new SessionCommandResult @@ -2438,91 +2513,923 @@ public async Task AgentEvent_AssistantDelta_AppendsStreamingAssistantEntry() } [Fact] - public async Task AgentEvent_AssistantContent_IsIgnoredBecauseChatMessageCarriesFinalText() + public async Task AgentEvent_AssistantContent_IsIgnoredBecauseChatMessageCarriesFinalText() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + await provider.LoadAsync(); + snapshots.Clear(); + + bridge.RaiseAgent(MakeAgentEvent("assistant", + """{"content":"Final answer."}""")); + + Assert.Empty(snapshots); + } + + [Fact] + public async Task AgentEvent_LifecycleStart_SetsTurnActive() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + await provider.LoadAsync(); + snapshots.Clear(); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", + """{"phase":"start"}""", runId: "run-1")); + + Assert.True(snapshots[^1].Timelines["main"].TurnActive); + } + + [Fact] + public async Task AgentEvent_LifecycleEnd_ClearsTurnActiveAndAssistant() + { + var (bridge, provider, _, _) = CreateProvider(new[] { MainSession() }); + await provider.LoadAsync(); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + bridge.RaiseAgent(MakeAgentEvent("assistant", """{"delta":"hi"}""")); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-1")); + + var snap = await provider.LoadAsync(); + var timeline = snap.Timelines["main"]; + Assert.False(timeline.TurnActive); + Assert.Null(timeline.ActiveAssistantId); + } + + [Fact] + public async Task AgentEvent_LifecycleError_AppendsErrorStatusEntry() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + await provider.LoadAsync(); + snapshots.Clear(); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + var evt = MakeAgentEvent("lifecycle", """{"phase":"error","message":"model unreachable"}""", runId: "run-1"); + bridge.RaiseAgent(evt); + + var timeline = snapshots[^1].Timelines["main"]; + Assert.False(timeline.TurnActive); + Assert.Contains(timeline.Entries, e => + e.Kind == ChatTimelineItemKind.Status && e.Text.Contains("model unreachable")); + } + + [Fact] + public async Task AgentEvent_LifecycleError_DispatchesNextQueuedMessage() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + snapshots.Clear(); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "second"); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"error","message":"model unreachable"}""", runId: "run-1")); + + await WaitForConditionAsync(() => + bridge.SentMessages.Count >= 2 && + GetQueuedMessages(snapshots[^1], "main").Count == 0 && + snapshots[^1].Timelines["main"].Entries.Any(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second")); + + Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second"); + } + + [Fact] + public async Task AssistantFinal_DispatchesNextQueuedMessageWithoutLifecycleEnd() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "second"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => + bridge.SentMessages.Count >= 2 && + GetQueuedMessages(snapshots[^1], "main").Count == 0 && + snapshots[^1].Timelines["main"].Entries.Any(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second")); + + Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second"); + } + + [Fact] + public async Task QueuedDuplicateUserMessages_AcceptedAckPromotesEachPromptByIdentity() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-3", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-4", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + snapshots.Clear(); + + await provider.SendMessageAsync("main", "Hello"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "Hello"); + await provider.SendMessageAsync("main", "Hello"); + await provider.SendMessageAsync("main", "Hello"); + + Assert.Equal(new[] { "Hello" }, bridge.SentMessages); + Assert.Equal(3, GetQueuedMessages(snapshots[^1], "main").Count); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "reply 1", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) + await Task.Delay(10); + await WaitForConditionAsync(() => + GetQueuedMessages(snapshots[^1], "main").Count == 2 && + snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello") == 2); + + Assert.Equal(2, bridge.SentMessages.Count); + Assert.Equal(2, GetQueuedMessages(snapshots[^1], "main").Count); + Assert.Equal(2, snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello")); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "reply 2", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 3; i++) + await Task.Delay(10); + await WaitForConditionAsync(() => + GetQueuedMessages(snapshots[^1], "main").Count == 1 && + snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello") == 3); + + Assert.Equal(3, bridge.SentMessages.Count); + Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal(3, snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello")); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "reply 3", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 4; i++) + await Task.Delay(10); + await WaitForConditionAsync(() => + GetQueuedMessages(snapshots[^1], "main").Count == 0 && + snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello") == 4); + + Assert.Equal(new[] { "Hello", "Hello", "Hello", "Hello" }, bridge.SentMessages); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.All(bridge.SentIdempotencyKeys, key => Assert.False(string.IsNullOrWhiteSpace(key))); + Assert.Equal(4, bridge.SentIdempotencyKeys.Distinct(StringComparer.Ordinal).Count()); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "reply 4", + State = "final", + }); + + Assert.Equal( + new[] { "Hello", "reply 1", "Hello", "reply 2", "Hello", "reply 3", "Hello", "reply 4" }, + snapshots[^1].Timelines["main"].Entries + .Where(e => e.Kind is ChatTimelineItemKind.User or ChatTimelineItemKind.Assistant) + .Select(e => e.Text) + .ToArray()); + } + + [Fact] + public async Task CancelQueuedMessageAsync_RemovesOneDuplicateQueuedItemAndPreventsDispatch() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-active", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-next-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-next-2", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + snapshots.Clear(); + + await provider.SendMessageAsync("main", "active"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-active")); + await provider.SendMessageAsync("main", "same"); + await provider.SendMessageAsync("main", "same"); + await provider.SendMessageAsync("main", "same"); + + var queuedBeforeCancel = GetQueuedMessages(snapshots[^1], "main"); + Assert.Equal(3, queuedBeforeCancel.Count); + Assert.All(queuedBeforeCancel, message => Assert.Equal("same", message.Text)); + var canceledId = queuedBeforeCancel[1].Id; + + var canceled = await provider.CancelQueuedMessageAsync("main", canceledId); + + var queuedAfterCancel = GetQueuedMessages(snapshots[^1], "main"); + Assert.True(canceled); + Assert.Equal(2, queuedAfterCancel.Count); + Assert.DoesNotContain(queuedAfterCancel, message => message.Id == canceledId); + Assert.All(queuedAfterCancel, message => Assert.Equal("same", message.Text)); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "active response", + State = "final", + }); + await WaitForConditionAsync(() => + bridge.SentMessages.Count >= 2 && + GetQueuedMessages(snapshots[^1], "main").Count == 1); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "response 1", + State = "final", + }); + await WaitForConditionAsync(() => + bridge.SentMessages.Count >= 3 && + GetQueuedMessages(snapshots[^1], "main").Count == 0); + + Assert.Equal(new[] { "active", "same", "same" }, bridge.SentMessages); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal(2, snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "same")); + } + + [Fact] + public async Task CancelQueuedMessageAsync_RemovesFailedQueuedCard() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-active", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { Status = "failed", Error = "gateway rejected queued send" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "active"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-active")); + await provider.SendMessageAsync("main", "failed queued"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "active response", + State = "final", + }); + await WaitForConditionAsync(() => HasFailedQueuedMessage(snapshots[^1], "main", "failed queued")); + + var failed = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal(ChatQueuedMessageSendState.Failed, failed.SendState); + + var canceled = await provider.CancelQueuedMessageAsync("main", failed.Id); + + Assert.True(canceled); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.False(snapshots[^1].QueuedMessagesByThread?.ContainsKey("main") == true); + } + + [Fact] + public async Task CancelQueuedMessageAsync_RemovingLastQueuedMessageClearsStaleDrainGuard() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-active", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "active"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-active")); + await provider.SendMessageAsync("main", "queued"); + + var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + var scheduled = GetQueuedDrainScheduledThreads(provider); + scheduled.Add("main"); + + var canceled = await provider.CancelQueuedMessageAsync("main", queued.Id); + + Assert.True(canceled); + Assert.DoesNotContain("main", scheduled); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + } + + [Fact] + public async Task CancelQueuedMessageAsync_ReturnsFalseForSendingQueuedCard() + { + var queuedSendStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseQueuedSend = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-active", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-queued", Status = "started" }); + bridge.SendBehavior = async (message, _, _) => + { + if (message == "queued") + { + queuedSendStarted.SetResult(); + await releaseQueuedSend.Task; + } + }; + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "active"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-active")); + await provider.SendMessageAsync("main", "queued"); + var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "active response", + State = "final", + }); + await queuedSendStarted.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await WaitForConditionAsync(() => + GetQueuedMessages(snapshots[^1], "main").Single().SendState == ChatQueuedMessageSendState.Sending); + + var canceled = await provider.CancelQueuedMessageAsync("main", queued.Id); + + Assert.False(canceled); + Assert.Equal(ChatQueuedMessageSendState.Sending, GetQueuedMessages(snapshots[^1], "main").Single().SendState); + releaseQueuedSend.SetResult(); + } + + [Fact] + public async Task CancelQueuedMessageAsync_DoesNotTurnActiveLocalRunIntoRemoteRun() + { + var historyCalls = 0; + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-active", Status = "started" }); + bridge.HistoryBehavior = _ => + { + historyCalls++; + return Task.FromResult(new ChatHistoryInfo { SessionKey = "main" }); + }; + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "active"); + await provider.SendMessageAsync("main", "queued"); + + var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + await provider.CancelQueuedMessageAsync("main", queued.Id); + + historyCalls = 0; + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-active")); + await Task.Delay(50); + + Assert.Equal(0, historyCalls); + } + + [Fact] + public async Task CancelQueuedMessageAsync_LastQueuedAfterLifecycleEndAllowsNextRemoteRunBackfill() + { + var historyCalls = 0; + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-active", Status = "started" }); + bridge.HistoryBehavior = _ => + { + historyCalls++; + return Task.FromResult(new ChatHistoryInfo + { + SessionKey = "main", + Messages = new[] + { + new ChatMessageInfo { SessionKey = "main", Role = "user", Text = "remote prompt" }, + }, + }); + }; + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "active"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-active")); + await provider.SendMessageAsync("main", "queued"); + var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-active")); + var canceled = await provider.CancelQueuedMessageAsync("main", queued.Id); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-remote")); + + Assert.True(canceled); + await WaitForConditionAsync(() => + historyCalls > 0 && + snapshots[^1].Timelines["main"].Entries.Any(entry => + entry.Kind == ChatTimelineItemKind.User && entry.Text == "remote prompt")); + Assert.True(historyCalls > 0); + Assert.Contains(snapshots[^1].Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.User && entry.Text == "remote prompt"); + Assert.DoesNotContain(snapshots[^1].Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.User && entry.Text == "queued"); + } + + [Fact] + public async Task QueuedSend_LifecycleStartBeforeAck_PromotesByIdempotencyKey() + { + var secondSendGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var sendCount = 0; + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { Status = "started" }); + bridge.SendBehavior = (_, _, _) => + { + sendCount++; + if (sendCount == 2) + { + var preAckRunId = Assert.Single(bridge.SentIdempotencyKeys.Skip(1)); + Assert.False(string.IsNullOrWhiteSpace(preAckRunId)); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: preAckRunId)); + return secondSendGate.Task; + } + + return Task.CompletedTask; + }; + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "second"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) + await Task.Delay(10); + + Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second"); + + secondSendGate.SetResult(); + } + + [Fact] + public async Task QueuedSend_InFlightAckWithoutLifecycle_RequeuesAndRetriesSameIdempotencyKey() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "in_flight" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "Hello"); + + var postFinalSnapshotStart = snapshots.Count; + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + for (var i = 0; i < 30 && bridge.SentMessages.Count < 2; i++) + await Task.Delay(10); + + Assert.Equal(2, bridge.SentMessages.Count); + var firstAttemptKey = bridge.SentIdempotencyKeys[1]; + ChatDataSnapshot? requeuedSnapshot = null; + for (var i = 0; i < 30 && requeuedSnapshot is null; i++) + { + requeuedSnapshot = snapshots + .Skip(postFinalSnapshotStart) + .LastOrDefault(snapshot => + { + var queued = GetQueuedMessages(snapshot, "main"); + return queued.Count == 1 && + queued[0].Text == "Hello" && + queued[0].SendState == ChatQueuedMessageSendState.Queued && + snapshot.Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello") == 0; + }); + if (requeuedSnapshot is null) + await Task.Delay(10); + } + + Assert.NotNull(requeuedSnapshot); + + bridge.RaiseSessions(new[] { MainSession() }); + for (var i = 0; i < 30 && bridge.SentMessages.Count < 3; i++) + await Task.Delay(10); + + Assert.Equal(3, bridge.SentMessages.Count); + Assert.Equal(firstAttemptKey, bridge.SentIdempotencyKeys[2]); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello"); + } + + [Fact] + public async Task CancelQueuedMessageAsync_DeferredInFlightRetryRemovesQueuedCardWithoutAbortOrResend() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "in_flight" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "deferred"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => + { + var queued = GetQueuedMessages(snapshots[^1], "main"); + return bridge.SentMessages.Count == 2 && + queued.Count == 1 && + queued[0].Text == "deferred" && + queued[0].SendState == ChatQueuedMessageSendState.Queued; + }); + + var deferred = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + var canceled = await provider.CancelQueuedMessageAsync("main", deferred.Id); + await Task.Delay(250); + + Assert.True(canceled); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal(new[] { "first", "deferred" }, bridge.SentMessages); + Assert.Empty(bridge.AbortedRunIds); + Assert.DoesNotContain(snapshots[^1].Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.User && entry.Text == "deferred"); + } + + [Fact] + public async Task QueuedSend_InFlightAckThenLifecycleBeforeRetry_PromotesWithoutResend() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "in_flight" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + var lifecycleRaisedBeforeRetry = false; + provider.Changed += (_, e) => + { + if (lifecycleRaisedBeforeRetry || bridge.SentMessages.Count != 2) + return; + + var queued = GetQueuedMessages(e.Snapshot, "main"); + if (queued.Count != 1 || + queued[0].Text != "Hello" || + queued[0].SendState != ChatQueuedMessageSendState.Queued) + { + return; + } + + lifecycleRaisedBeforeRetry = true; + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-2")); + }; + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "Hello"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => + { + var queued = GetQueuedMessages(snapshots[^1], "main"); + return bridge.SentMessages.Count == 2 && + queued.Count == 1 && + queued[0].Text == "Hello" && + queued[0].SendState == ChatQueuedMessageSendState.Queued; + }); + + await WaitForConditionAsync(() => + GetQueuedMessages(snapshots[^1], "main").Count == 0 && + snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello") == 1); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "user", + Text = "Hello", + }); + await Task.Delay(250); + + Assert.True(lifecycleRaisedBeforeRetry); + Assert.Equal(2, bridge.SentMessages.Count); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "Hello"); + } + + [Fact] + public async Task QueuedSend_InFlightRetry_DoesNotSuppressLaterSameTextRemoteUserMessage() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "in_flight" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "Hello"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => + { + var queued = GetQueuedMessages(snapshots[^1], "main"); + return bridge.SentMessages.Count >= 2 && + queued.Count == 1 && + queued[0].Text == "Hello" && + queued[0].SendState == ChatQueuedMessageSendState.Queued; + }); + + await WaitForConditionAsync(() => + bridge.SentMessages.Count >= 3 && + GetQueuedMessages(snapshots[^1], "main").Count == 0 && + snapshots[^1].Timelines["main"].Entries.Count(entry => + entry.Kind == ChatTimelineItemKind.User && entry.Text == "Hello") == 1, + attempts: 200); + + Assert.Equal(bridge.SentIdempotencyKeys[1], bridge.SentIdempotencyKeys[2]); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "user", + Text = "Hello", + }); + Assert.Single(snapshots[^1].Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.User && entry.Text == "Hello"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "user", + Text = "Hello", + }); + await WaitForConditionAsync(() => + snapshots[^1].Timelines["main"].Entries.Count(entry => + entry.Kind == ChatTimelineItemKind.User && entry.Text == "Hello") == 2); + } + + [Fact] + public async Task QueuedSend_InFlightAckWhileDifferentRunActive_RequeuesWithoutPromoting() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { Status = "in_flight" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + var releaseInFlightAck = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var delayFirstStuckAttempt = true; + bridge.SendBehavior = (message, _, _) => + { + if (message == "stuck" && delayFirstStuckAttempt) + { + delayFirstStuckAttempt = false; + return releaseInFlightAck.Task; + } + + return Task.CompletedTask; + }; + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "stuck"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "remote-run")); + releaseInFlightAck.SetResult(); + await WaitForConditionAsync(() => + { + var queued = GetQueuedMessages(snapshots[^1], "main"); + return queued.Count == 1 && + queued[0].Text == "stuck" && + queued[0].SendState == ChatQueuedMessageSendState.Queued && + snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "stuck") == 0; + }); + + Assert.Equal(2, bridge.SentMessages.Count); + var firstAttemptKey = bridge.SentIdempotencyKeys[1]; + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "remote-run")); + await WaitForConditionAsync(() => + bridge.SentMessages.Count >= 3 && + GetQueuedMessages(snapshots[^1], "main").Count == 0 && + snapshots[^1].Timelines["main"].Entries.Count(e => + e.Kind == ChatTimelineItemKind.User && e.Text == "stuck") == 1, + attempts: 200); + + Assert.Equal(firstAttemptKey, bridge.SentIdempotencyKeys[2]); + } + + [Fact] + public async Task DirectSend_InFlightAckWithoutLifecycle_FailsInsteadOfLeavingTurnActive() { var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { Status = "in_flight" }); await provider.LoadAsync(); - snapshots.Clear(); + bridge.RaiseStatus(ConnectionStatus.Connected); - bridge.RaiseAgent(MakeAgentEvent("assistant", - """{"content":"Final answer."}""")); + var ex = await Assert.ThrowsAsync(() => + provider.SendMessageAsync("main", "direct")); - Assert.Empty(snapshots); + Assert.Contains("in_flight", ex.Message, StringComparison.Ordinal); + var snapshot = snapshots[^1]; + Assert.False(snapshot.Timelines["main"].TurnActive); + Assert.Contains(snapshot.Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.Status && + entry.Text.Contains("in_flight", StringComparison.Ordinal)); } [Fact] - public async Task AgentEvent_LifecycleStart_SetsTurnActive() + public async Task DirectSend_TimeoutAckWithRunId_FailsInsteadOfPromoting() { var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-timeout", Status = "timeout" }); await provider.LoadAsync(); - snapshots.Clear(); + bridge.RaiseStatus(ConnectionStatus.Connected); - bridge.RaiseAgent(MakeAgentEvent("lifecycle", - """{"phase":"start"}""", runId: "run-1")); + var ex = await Assert.ThrowsAsync(() => + provider.SendMessageAsync("main", "direct timeout")); - Assert.True(snapshots[^1].Timelines["main"].TurnActive); + Assert.Contains("timeout", ex.Message, StringComparison.Ordinal); + var snapshot = snapshots[^1]; + Assert.False(snapshot.Timelines["main"].TurnActive); + Assert.Empty(GetQueuedMessages(snapshot, "main")); + Assert.Contains(snapshot.Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.Status && + entry.Text.Contains("timeout", StringComparison.Ordinal)); } [Fact] - public async Task AgentEvent_LifecycleEnd_ClearsTurnActiveAndAssistant() + public async Task QueuedSend_TimeoutAckWithRunId_KeepsFailedQueuedMessage() { - var (bridge, provider, _, _) = CreateProvider(new[] { MainSession() }); + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-timeout", Status = "timeout" }); await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + await provider.SendMessageAsync("main", "first"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); - bridge.RaiseAgent(MakeAgentEvent("assistant", """{"delta":"hi"}""")); - bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "queued timeout"); - var snap = await provider.LoadAsync(); - var timeline = snap.Timelines["main"]; - Assert.False(timeline.TurnActive); - Assert.Null(timeline.ActiveAssistantId); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => HasFailedQueuedMessage(snapshots[^1], "main", "queued timeout")); + + var failed = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal("queued timeout", failed.Text); + Assert.Equal(ChatQueuedMessageSendState.Failed, failed.SendState); + Assert.Contains("timeout", failed.ErrorText, StringComparison.Ordinal); + Assert.Equal(new[] { "first", "queued timeout" }, bridge.SentMessages); + Assert.DoesNotContain(snapshots[^1].Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.User && + entry.Text == "queued timeout"); } [Fact] - public async Task AgentEvent_LifecycleError_AppendsErrorStatusEntry() + public async Task SendMessageAsync_FailedQueuedCardDoesNotForceNextSendToQueue() { var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { Status = "failed", Error = "queued failed" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-3", Status = "started" }); await provider.LoadAsync(); - snapshots.Clear(); + bridge.RaiseStatus(ConnectionStatus.Connected); + await provider.SendMessageAsync("main", "first"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); - var evt = MakeAgentEvent("lifecycle", """{"phase":"error","message":"model unreachable"}""", runId: "run-1"); - bridge.RaiseAgent(evt); + await provider.SendMessageAsync("main", "failed queued"); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => HasFailedQueuedMessage(snapshots[^1], "main", "failed queued")); - var timeline = snapshots[^1].Timelines["main"]; - Assert.False(timeline.TurnActive); - Assert.Contains(timeline.Entries, e => - e.Kind == ChatTimelineItemKind.Status && e.Text.Contains("model unreachable")); + var beforeNextSendSnapshotCount = snapshots.Count; + await provider.SendMessageAsync("main", "after failure"); + + var queued = GetQueuedMessages(snapshots[^1], "main"); + Assert.Single(queued, message => + message.Text == "failed queued" && + message.SendState == ChatQueuedMessageSendState.Failed); + Assert.DoesNotContain(snapshots.Skip(beforeNextSendSnapshotCount), snapshot => + GetQueuedMessages(snapshot, "main").Any(message => message.Text == "after failure")); + Assert.Contains(snapshots[^1].Timelines["main"].Entries, entry => + entry.Kind == ChatTimelineItemKind.User && + entry.Text == "after failure"); } [Fact] - public async Task AgentEvent_LifecycleError_DispatchesNextQueuedMessage() + public async Task QueuedSend_RepeatedInFlightAckWithoutLifecycle_EventuallyFails() { var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); - bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-2", Status = "started" }); + for (var i = 0; i < 13; i++) + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-stuck", Status = "in_flight" }); await provider.LoadAsync(); bridge.RaiseStatus(ConnectionStatus.Connected); - snapshots.Clear(); await provider.SendMessageAsync("main", "first"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); - await provider.SendMessageAsync("main", "second"); + await provider.SendMessageAsync("main", "stuck"); - bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"error","message":"model unreachable"}""", runId: "run-1")); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "first response", + State = "final", + }); + await WaitForConditionAsync(() => bridge.SentMessages.Count >= 2); - for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) + for (var i = 0; i < 20 && !HasFailedQueuedMessage(snapshots[^1], "main", "stuck"); i++) + { + bridge.RaiseSessions(new[] { MainSession() }); await Task.Delay(10); + } + await WaitForConditionAsync(() => HasFailedQueuedMessage(snapshots[^1], "main", "stuck"), attempts: 1000); - Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); - var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); - Assert.Equal("second", queued.Text); - Assert.Equal(ChatQueuedMessageSendState.Sending, queued.SendState); + var failed = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal("stuck", failed.Text); + Assert.Equal(ChatQueuedMessageSendState.Failed, failed.SendState); + Assert.Contains("in_flight", failed.ErrorText, StringComparison.Ordinal); + Assert.Single(bridge.SentIdempotencyKeys.Skip(1).Distinct(StringComparer.Ordinal)); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.Status && + e.Text.Contains("in_flight", StringComparison.Ordinal)); } [Fact] - public async Task AssistantFinal_DispatchesNextQueuedMessageWithoutLifecycleEnd() + public async Task QueuedSend_AckPromotionKeepsRunMappingUntilTerminalCleanup() { var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); @@ -2533,7 +3440,6 @@ public async Task AssistantFinal_DispatchesNextQueuedMessageWithoutLifecycleEnd( await provider.SendMessageAsync("main", "first"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); await provider.SendMessageAsync("main", "second"); - bridge.RaiseChat(new ChatMessageInfo { SessionKey = "main", @@ -2543,14 +3449,25 @@ public async Task AssistantFinal_DispatchesNextQueuedMessageWithoutLifecycleEnd( }); for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) await Task.Delay(10); + await WaitForConditionAsync(() => GetQueuedMessages(snapshots[^1], "main").Count == 0); - Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); - var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); - Assert.Equal(ChatQueuedMessageSendState.Sending, queued.SendState); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-2")); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"end"}""", runId: "run-2")); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "second response", + State = "final", + }); + + Assert.Contains(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.Assistant && e.Text == "second response"); } [Fact] - public async Task IdentitylessIdenticalAssistant_DefersUntilQueuedUserBoundaryIsConfirmed() + public async Task IdentitylessIdenticalAssistant_DropsBeforeQueuedUserBoundaryButAllowsCurrentRunResponse() { var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); @@ -2570,6 +3487,7 @@ public async Task IdentitylessIdenticalAssistant_DefersUntilQueuedUserBoundaryIs }); for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) await Task.Delay(10); + await WaitForConditionAsync(() => GetQueuedMessages(snapshots[^1], "main").Count == 0); bridge.RaiseChat(new ChatMessageInfo { @@ -2579,20 +3497,183 @@ public async Task IdentitylessIdenticalAssistant_DefersUntilQueuedUserBoundaryIs State = "final", }); - Assert.Single(GetQueuedMessages(snapshots[^1], "main")); - Assert.Single(snapshots[^1].Timelines["main"].Entries, e => - e.Kind == ChatTimelineItemKind.Assistant && e.Text == "OK"); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal( + new[] { "first", "OK", "second" }, + snapshots[^1].Timelines["main"].Entries + .Where(e => e.Kind is ChatTimelineItemKind.User or ChatTimelineItemKind.Assistant) + .Select(e => e.Text) + .ToArray()); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-2")); var entries = snapshots[^1].Timelines["main"].Entries; Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); Assert.Equal( - new[] { "first", "OK", "second", "OK" }, + new[] { "first", "OK", "second" }, entries .Where(e => e.Kind is ChatTimelineItemKind.User or ChatTimelineItemKind.Assistant) .Select(e => e.Text) .ToArray()); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "OK", + State = "final", + }); + + Assert.Equal( + new[] { "first", "OK", "second", "OK" }, + snapshots[^1].Timelines["main"].Entries + .Where(e => e.Kind is ChatTimelineItemKind.User or ChatTimelineItemKind.Assistant) + .Select(e => e.Text) + .ToArray()); + } + + [Fact] + public async Task QueuedReplies_IgnoreIdentitylessRetransmitsAndStayMatchedToPrompts() + { + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-a", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-b", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-c", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-d", Status = "started" }); + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + snapshots.Clear(); + + await provider.SendMessageAsync("main", "a"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-a")); + await provider.SendMessageAsync("main", "b"); + await provider.SendMessageAsync("main", "c"); + await provider.SendMessageAsync("main", "d"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "ack - a", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) + await Task.Delay(10); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "ack - a", + State = "final", + }); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-b")); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "ack - b", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 3; i++) + await Task.Delay(10); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "ack - b", + State = "final", + }); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-c")); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "ack - c", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 4; i++) + await Task.Delay(10); + await WaitForConditionAsync(() => GetQueuedMessages(snapshots[^1], "main").Count == 0); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "ack - c", + State = "final", + }); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-d")); + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "ack - d", + State = "final", + }); + + Assert.Equal(new[] { "a", "b", "c", "d" }, bridge.SentMessages); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Equal( + new[] { "a", "ack - a", "b", "ack - b", "c", "ack - c", "d", "ack - d" }, + snapshots[^1].Timelines["main"].Entries + .Where(e => e.Kind is ChatTimelineItemKind.User or ChatTimelineItemKind.Assistant) + .Select(e => e.Text) + .ToArray()); + } + + [Fact] + public async Task DroppedIdentitylessAssistant_IsNotReplayedAfterSendingPromptFails() + { + var secondSendGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var sendCount = 0; + var (bridge, provider, snapshots, _) = CreateProvider(new[] { MainSession() }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-1", Status = "started" }); + bridge.SendResults.Enqueue(new ChatSendResult { Status = "failed", Error = "boom" }); + bridge.SendResults.Enqueue(new ChatSendResult { RunId = "run-3", Status = "started" }); + bridge.SendBehavior = (_, _, _) => + { + sendCount++; + return sendCount == 2 ? secondSendGate.Task : Task.CompletedTask; + }; + await provider.LoadAsync(); + bridge.RaiseStatus(ConnectionStatus.Connected); + + await provider.SendMessageAsync("main", "first"); + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-1")); + await provider.SendMessageAsync("main", "second"); + await provider.SendMessageAsync("main", "third"); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "OK", + State = "final", + }); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 2; i++) + await Task.Delay(10); + + bridge.RaiseChat(new ChatMessageInfo + { + SessionKey = "main", + Role = "assistant", + Text = "OK", + State = "final", + }); + + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.Assistant && e.Text == "OK"); + + secondSendGate.SetResult(); + for (var i = 0; i < 20 && bridge.SentMessages.Count < 3; i++) + await Task.Delay(10); + + bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-3")); + + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.Assistant && e.Text == "OK"); } [Fact] @@ -2622,12 +3703,10 @@ public async Task IdentifiedDuplicateAssistant_DoesNotPromoteNextQueuedMessage() bridge.RaiseChat(firstFinal); - var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); - Assert.Equal("second", queued.Text); - Assert.Equal(ChatQueuedMessageSendState.Sending, queued.SendState); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); Assert.Single(snapshots[^1].Timelines["main"].Entries, e => e.Kind == ChatTimelineItemKind.Assistant && e.Text == "OK"); - Assert.DoesNotContain(snapshots[^1].Timelines["main"].Entries, e => + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => e.Kind == ChatTimelineItemKind.User && e.Text == "second"); bridge.RaiseAgent(MakeAgentEvent("lifecycle", """{"phase":"start"}""", runId: "run-2")); @@ -2855,9 +3934,9 @@ public async Task StopResponseAsync_WithQueuedFollowUp_WaitsForConfirmedTerminal Assert.Contains("run-1", bridge.AbortedRunIds); Assert.Equal(new[] { "first", "second" }, bridge.SentMessages); - var queued = Assert.Single(GetQueuedMessages(snapshots[^1], "main")); - Assert.Equal("second", queued.Text); - Assert.Equal(ChatQueuedMessageSendState.Sending, queued.SendState); + Assert.Empty(GetQueuedMessages(snapshots[^1], "main")); + Assert.Single(snapshots[^1].Timelines["main"].Entries, e => + e.Kind == ChatTimelineItemKind.User && e.Text == "second"); } [Fact] @@ -5933,6 +7012,12 @@ public async Task LocalExecApproval_TimeoutExpiresEntryAndCompletesAsTimedOutDen Assert.Null(timedOutTimeline.PendingPermission); } + private static async Task WaitForConditionAsync(Func condition, int attempts = 50) + { + for (var i = 0; i < attempts && !condition(); i++) + await Task.Delay(10); + } + private static AgentEventInfo MakeApprovalResolvedEvent( string approvalId, string phase, @@ -5974,6 +7059,20 @@ private static IReadOnlyList GetQueuedMessages(ChatDataSnapsh ? queued : Array.Empty(); + private static ISet GetQueuedDrainScheduledThreads(OpenClawChatDataProvider provider) + { + var field = typeof(OpenClawChatDataProvider).GetField( + "_queuedDrainScheduledThreads", + System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic); + Assert.NotNull(field); + return Assert.IsAssignableFrom>(field.GetValue(provider)); + } + + private static bool HasFailedQueuedMessage(ChatDataSnapshot snapshot, string threadId, string text) => + GetQueuedMessages(snapshot, threadId).Any(message => + message.Text == text && + message.SendState == ChatQueuedMessageSendState.Failed); + private static void AssertNoQueuedTranscriptDuplicate( IEnumerable snapshots, string threadId, diff --git a/tests/OpenClaw.Tray.Tests/ToolMetaCacheTests.cs b/tests/OpenClaw.Tray.Tests/ToolMetaCacheTests.cs index 92116b283..06e02dcaf 100644 --- a/tests/OpenClaw.Tray.Tests/ToolMetaCacheTests.cs +++ b/tests/OpenClaw.Tray.Tests/ToolMetaCacheTests.cs @@ -365,7 +365,12 @@ private sealed class FakeBridge : IChatGatewayBridge public void StartProactiveBootstrap() { } public Task ListCommandsAsync(CommandCatalogQuery? query = null) => Task.FromResult(new CommandCatalog { IsSupported = true }); public Task SendChatMessageAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null) => Task.CompletedTask; - public Task SendChatMessageForRunAsync(string message, string? sessionKey, string? sessionId, IReadOnlyList? attachments = null) => Task.FromResult(new ChatSendResult()); + public Task SendChatMessageForRunAsync( + string message, + string? sessionKey, + string? sessionId, + IReadOnlyList? attachments = null, + string? idempotencyKey = null) => Task.FromResult(new ChatSendResult()); public Task PatchSessionModelAsync(string sessionKey, string model) => Task.CompletedTask; public Task ClearSessionModelAsync(string sessionKey) => Task.CompletedTask; public Task PatchSessionThinkingLevelAsync(string sessionKey, string thinkingLevel) => Task.CompletedTask;