From 02ec1b6166f19bbb3c76a38742811e854e0a9ea3 Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Mon, 17 Aug 2026 17:33:40 -0700 Subject: [PATCH] fix(ui): keep acknowledged sends pending during stale history (#125426) --- ui/src/pages/chat/chat-outbox-drain.ts | 45 +++++++++- ui/src/pages/chat/chat-outbox-owner.ts | 32 ++++++- ui/src/pages/chat/chat-send.test.ts | 117 +++++++++++++++++++++++++ 3 files changed, 191 insertions(+), 3 deletions(-) diff --git a/ui/src/pages/chat/chat-outbox-drain.ts b/ui/src/pages/chat/chat-outbox-drain.ts index f97f3900daf0..0df25cfea60a 100644 --- a/ui/src/pages/chat/chat-outbox-drain.ts +++ b/ui/src/pages/chat/chat-outbox-drain.ts @@ -17,7 +17,9 @@ import { type ChatCommandResetOptions, } from "./chat-commands.ts"; import { loadChatHistory, type ChatHistoryResult } from "./chat-history.ts"; +import { chatOutboxOwner } from "./chat-outbox-owner.ts"; import { + anyChatOutboxPaneMatches, excludeComposerAttachments, readQueuedMessageById, removeQueuedMessageWithoutReleasing, @@ -91,6 +93,7 @@ type StoredChatOutboxClientState = { retryTimers: Map>; }; +const STORED_OUTBOX_CONFIRMATION_GRACE_MS = 5_000; const STORED_OUTBOX_RETRY_DEFAULT_MS = 500; const STORED_OUTBOX_RETRY_MIN_MS = 100; const STORED_OUTBOX_RETRY_MAX_MS = 30_000; @@ -106,7 +109,10 @@ function getStoredChatOutboxClientState(client: GatewayBrowserClient): StoredCha if (existing) { return existing; } - const created = { lanes: new Map(), retryTimers: new Map() }; + const created: StoredChatOutboxClientState = { + lanes: new Map(), + retryTimers: new Map(), + }; storedChatOutboxClients.set(client, created); return created; } @@ -164,6 +170,10 @@ function sameQueuedDeliveryVersion(left: ChatQueueItem, right: ChatQueueItem): b ); } +function queuedDeliveryVersion(item: ChatQueueItem): string { + return `${item.id}\u0000${item.sendRunId ?? ""}\u0000${item.sendAttempts ?? 0}`; +} + async function readCurrentStoredChatHistory( host: ChatHost, outbox: StoredChatOutbox, @@ -282,21 +292,54 @@ async function reconcileStoredChatOutboxHead( return "blocked"; } } + const outboxOwner = chatOutboxOwner(host); + const clearConfirmationGrace = () => outboxOwner.clearConfirmationGrace(outbox, item.id); const historyArgs = [host, outbox, item, client, connectionEpoch, dependencies] as const; const history = await readCurrentStoredChatHistory(...historyArgs); // Keyed unknown sends reach history only for exact proof; absence stays blocked. if (history === "blocked" || history === "continue" || item.sendState === "unconfirmed") { + clearConfirmationGrace(); return history === "continue" ? "continue" : "blocked"; } if (visibleSessionMatches(host, outbox.sessionKey, outbox.agentId) && isChatBusy(host)) { + clearConfirmationGrace(); return "blocked"; } if ((item.sendAttempts ?? 0) > 0) { // History and run metadata are non-atomic; verify idle before parking unknown. const verifiedHistory = await readCurrentStoredChatHistory(...historyArgs); if (verifiedHistory === "blocked" || verifiedHistory === "continue") { + clearConfirmationGrace(); return verifiedHistory; } + const liveSendCurrent = anyChatOutboxPaneMatches(host, (pane) => { + const liveItem = pane.chatQueue.find((entry) => entry.id === item.id); + return ( + liveItem?.sendState === "sending" && + sameQueuedDeliveryVersion(liveItem, { ...item, sendState: "sending" }) + ); + }); + if (liveSendCurrent) { + // Start the bound only after idle is verified; a valid run can outlive the send request. + const now = Date.now(); + const deadlineMs = outboxOwner.confirmationDeadline( + outbox, + item.id, + queuedDeliveryVersion(item), + now, + STORED_OUTBOX_CONFIRMATION_GRACE_MS, + ); + if (deadlineMs !== null && now < deadlineMs) { + scheduleStoredChatOutboxRetry( + host, + outbox, + Math.min(STORED_OUTBOX_RETRY_DEFAULT_MS, deadlineMs - now), + dependencies, + ); + return "blocked"; + } + } + clearConfirmationGrace(); const parked = updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ ...entry, sendError: UNCONFIRMED_CHAT_SEND_ERROR, diff --git a/ui/src/pages/chat/chat-outbox-owner.ts b/ui/src/pages/chat/chat-outbox-owner.ts index 4605b7d2d2c8..38606fbd1efa 100644 --- a/ui/src/pages/chat/chat-outbox-owner.ts +++ b/ui/src/pages/chat/chat-outbox-owner.ts @@ -15,7 +15,11 @@ type HostProjection = { durableSeen: Set; retryable: Set; }; -type LiveProjection = { item: ChatQueueItem; owner: Host }; +type LiveProjection = { + confirmationGrace?: { deadlineMs: number; deliveryVersion: string }; + item: ChatQueueItem; + owner: Host; +}; const LIVE_VERSION_KEYS = ["sendRunId", "sendAttempts", "sendState", "sendError"] as const; const routePresentation = new WeakSet(); const storageIds = new WeakMap(); @@ -233,6 +237,30 @@ class ChatOutboxGatewayOwner { hasVolatile(host: Host, id: string): boolean { return this.hosts.get(host)?.retryable.has(id) ?? false; } + confirmationDeadline( + scope: Scope, + itemId: string, + deliveryVersion: string, + now: number, + graceMs: number, + ): number | null { + const live = this.live.get(storedChatOutboxScopeKey(scope))?.get(itemId); + if (!live) { + return null; + } + if (live.confirmationGrace?.deliveryVersion === deliveryVersion) { + return live.confirmationGrace.deadlineMs; + } + const deadlineMs = now + graceMs; + live.confirmationGrace = { deadlineMs, deliveryVersion }; + return deadlineMs; + } + clearConfirmationGrace(scope: Scope, itemId: string): void { + const live = this.live.get(storedChatOutboxScopeKey(scope))?.get(itemId); + if (live) { + delete live.confirmationGrace; + } + } // Panes share this outbox and its drain while composer state stays per pane, so // a pane-local fact that blocks delivery has to be answerable from any of them. anyPane(matches: (host: Host) => boolean): boolean { @@ -257,7 +285,7 @@ class ChatOutboxGatewayOwner { const key = storedChatOutboxScopeKey(scope); const live = this.live.get(key) ?? new Map(); if (item) { - live.set(id, { item, owner: host }); + live.set(id, { confirmationGrace: live.get(id)?.confirmationGrace, item, owner: host }); this.live.set(key, live); } else { live.delete(id); diff --git a/ui/src/pages/chat/chat-send.test.ts b/ui/src/pages/chat/chat-send.test.ts index 3c59fda55a3c..e5bf658e2eed 100644 --- a/ui/src/pages/chat/chat-send.test.ts +++ b/ui/src/pages/chat/chat-send.test.ts @@ -32,6 +32,7 @@ import { refreshChatAvatar } from "./chat-avatar.ts"; import * as chatCommandExecutor from "./chat-command-executor.ts"; import type { executeSlashCommand } from "./chat-command-executor.ts"; import { makeChatHost, makeRequestMock } from "./chat-host.test-support.ts"; +import { UNCONFIRMED_CHAT_SEND_ERROR } from "./chat-outbox-drain.ts"; import { renderChatPaneComposerControls } from "./chat-pane-session-controls.ts"; import type { ChatHost } from "./chat-send-contract.ts"; import { @@ -149,6 +150,7 @@ let loadChatHistory: typeof import("./chat-history.ts").loadChatHistory; let clearPendingQueueItemsForRun: typeof import("./chat-queue.ts").clearPendingQueueItemsForRun; let admitQueuedMessageForSession: typeof import("./chat-queue.ts").admitQueuedMessageForSession; let removeQueuedMessage: typeof import("./chat-queue.ts").removeQueuedMessage; +let setTransientQueuedMessageProjection: typeof import("./chat-queue.ts").setTransientQueuedMessageProjection; let removeDeliveredQueuedChatSendForRun: typeof import("./chat-queue.ts").removeDeliveredQueuedChatSendForRun; let removeVisibleOrScopedQueuedMessageWithoutReleasing: typeof import("./chat-queue.ts").removeVisibleOrScopedQueuedMessageWithoutReleasing; let markQueuedChatSendsWaitingForReconnect: typeof import("./chat-queue.ts").markQueuedChatSendsWaitingForReconnect; @@ -182,6 +184,7 @@ async function loadChatHelpers(): Promise { clearPendingQueueItemsForRun, removeDeliveredQueuedChatSendForRun, removeQueuedMessage, + setTransientQueuedMessageProjection, markQueuedChatSendsWaitingForReconnect, removeVisibleOrScopedQueuedMessageWithoutReleasing, readChatQueueForScope, @@ -5604,6 +5607,120 @@ describe("handleSendChat", () => { expect(host.chatMessages).toStrictEqual([]); }); + it("keeps an acknowledged live send pending while durable history is briefly stale", async () => { + let historyRequests = 0; + let runId: string | undefined; + const host = makeChatHost({ + requestHandlers: { + "chat.history": () => { + historyRequests += 1; + return Promise.resolve({ + messages: + historyRequests > 2 + ? [{ role: "user", __openclaw: { idempotencyKey: `${runId}:user` } }] + : [], + sessionInfo: row("agent:main", { hasActiveRun: false, status: "done" }), + }); + }, + "chat.send": (params: unknown) => { + const payload = requireRecord(params, "live send payload"); + runId = String(payload.idempotencyKey); + return Promise.resolve({ runId, status: "started" }); + }, + }, + }); + + await handleSendChat(host, "history will catch up"); + expect(host.chatQueue).toEqual([ + expect.objectContaining({ sendState: "sending", text: "history will catch up" }), + ]); + expect(loadChatComposerSnapshot(host, host.sessionKey)?.queue).toEqual([ + expect.objectContaining({ sendAttempts: 1, sendState: "waiting-reconnect" }), + ]); + + host.chatRunId = null; + await flushChatQueueForEvent(host); + + expect(host.chatQueue[0]?.sendState).toBe("sending"); + expect(host.lastError).toBeNull(); + await waitForFast(() => expect(listStoredChatOutboxes(host)).toStrictEqual([]), { + timeout: 1_000, + }); + expect(historyRequests).toBeGreaterThanOrEqual(3); + expect(host.chatQueue).toStrictEqual([]); + expect(host.lastError).toBeNull(); + }); + + it("marks an acknowledged live send unconfirmed after history stays missing", async () => { + let now = 1_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + const host = makeChatHost({ + requestHandlers: { + "chat.history": () => idleChatHistory(), + "chat.send": (params: unknown) => { + const payload = requireRecord(params, "live send payload"); + const runId = String(payload.idempotencyKey); + return Promise.resolve({ runId, status: "started" }); + }, + }, + }); + + await handleSendChat(host, "history never catches up"); + host.chatRunId = null; + await flushChatQueueForEvent(host); + + expect(host.chatQueue[0]?.sendState).toBe("sending"); + expect(host.lastError).toBeNull(); + + now += 5_001; + await flushChatQueueForEvent(host); + + expect(host.chatQueue[0]).toMatchObject({ + sendError: UNCONFIRMED_CHAT_SEND_ERROR, + sendState: "unconfirmed", + text: "history never catches up", + }); + expect(host.lastError).toBe(UNCONFIRMED_CHAT_SEND_ERROR); + }); + + it("starts a fresh confirmation grace after the prior outbox scope is removed", async () => { + let now = 1_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + const host = makeChatHost({ + requestHandlers: { + "chat.history": () => idleChatHistory(), + "chat.send": (params: unknown) => { + const payload = requireRecord(params, "live send payload"); + const runId = String(payload.idempotencyKey); + return Promise.resolve({ runId, status: "started" }); + }, + }, + }); + + await handleSendChat(host, "reuse this delivery version"); + host.chatRunId = null; + await flushChatQueueForEvent(host); + + const item = expectDefined(host.chatQueue[0], "queued live send"); + const sessionKey = expectDefined(item.sessionKey, "queued session key"); + expect(removeQueuedMessage(host, item.id)).toBe("removed"); + + now = 5_500; + expect(admitQueuedMessageForSession(host, sessionKey, item)).toBe(true); + expect(setTransientQueuedMessageProjection(host, sessionKey, item)).toBe(true); + await flushChatQueueForEvent(host); + + now = 6_001; + await flushChatQueueForEvent(host); + + expect(host.chatQueue[0]).toMatchObject({ + id: item.id, + sendState: "sending", + text: "reuse this delivery version", + }); + expect(host.lastError).toBeNull(); + }); + it("coalesces duplicate queued local commands while the first command is running", async () => { const command = createDeferred<{ content: string }>(); executeSlashCommandMock.mockImplementation(() => command.promise);