From 429068d5e8f02ba5140074f802ace94995d39a45 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Tue, 22 Sep 2026 15:24:36 +0800 Subject: [PATCH] fix(chat): hide live transmissions from queue recovery panel --- .../src/styles/components/chat.scss | 4 +- src/mobile-web/src/styles/host-queue.scss | 2 +- .../tests/host-dialog-queue.test.mjs | 42 ++++++++++++++ src/shared/dialog-queue/HostDialogQueue.ts | 33 +++++++---- .../components/HostPendingQueuePanel.test.tsx | 58 +++++++++++++++++++ 5 files changed, 126 insertions(+), 13 deletions(-) create mode 100644 src/web-ui/src/flow_chat/components/HostPendingQueuePanel.test.tsx diff --git a/src/mobile-web/src/styles/components/chat.scss b/src/mobile-web/src/styles/components/chat.scss index cfe59c1fc9..c9e3996a0d 100644 --- a/src/mobile-web/src/styles/components/chat.scss +++ b/src/mobile-web/src/styles/components/chat.scss @@ -397,7 +397,7 @@ background: var(--openbitfun-color-surface-subtle); color: var(--openbitfun-color-content-secondary); font-size: var(--openbitfun-type-body-sm-font-size); - line-height: 1.5; + line-height: var(--openbitfun-type-body-sm-line-height); white-space: pre-wrap; overflow-wrap: anywhere; } @@ -405,7 +405,7 @@ .chat-msg__rollback-input { min-height: 96px; font-size: var(--openbitfun-type-body-md-font-size); - line-height: 1.5; + line-height: var(--openbitfun-type-body-md-line-height); } .chat-msg__menu-note { diff --git a/src/mobile-web/src/styles/host-queue.scss b/src/mobile-web/src/styles/host-queue.scss index 0a4a1fe008..db0dda3f45 100644 --- a/src/mobile-web/src/styles/host-queue.scss +++ b/src/mobile-web/src/styles/host-queue.scss @@ -3,7 +3,7 @@ max-height: 28vh; overflow: auto; flex-shrink: 0; - font-size: 12px; + font-size: var(--openbitfun-type-body-xs-font-size); ul { list-style: none; margin: 0; padding: 0; } li { padding-block: 8px; } p { margin-block: 4px; } diff --git a/src/mobile-web/tests/host-dialog-queue.test.mjs b/src/mobile-web/tests/host-dialog-queue.test.mjs index 417311b626..0c7247cadf 100644 --- a/src/mobile-web/tests/host-dialog-queue.test.mjs +++ b/src/mobile-web/tests/host-dialog-queue.test.mjs @@ -101,3 +101,45 @@ test('a serialized promote keeps the turn observed at click time',async()=>{ await Promise.all([first,second]); assert.equal(f.calls.find(call=>call.action==='promote').expectedActiveTurnId,'running'); }); + +function deferred() { + let resolve, reject; + const promise=new Promise((ok,fail)=>{resolve=ok;reject=fail;}); + return {promise,resolve,reject}; +} + +test('a live send never becomes unknown delivery, including observer refreshes before acknowledgement',async()=>{ + const f=fixture();const entered=deferred();const reply=deferred();const views=[]; + const q=new HostDialogQueue('account/host/session-a','session-a',async request=>{ + if(request.action==='submit'){entered.resolve();await reply.promise;} + return f.invoke(request); + },f.storage); + q.subscribe(()=>views.push(q.getSnapshot())); + const sending=q.submit(message);await entered.promise; + assert.equal(f.records.size,1,'persist before waiting for the host'); + await q.refresh();await q.refresh(); + assert.ok(views.every(view=>view.pending.length===0 && view.error===null)); + // Another page has no live request and must still recover the committed outbox. + const reopened=f.create();await reopened.refresh(); + assert.equal(reopened.getSnapshot().pending.length,1); + reply.resolve();await sending; + assert.ok(views.every(view=>view.pending.length===0 && view.error===null)); + assert.equal(q.getSnapshot().snapshot.items.length,1,'real host queue remains visible'); +}); + +test('a live send becomes recoverable only when its acknowledgement fails',async()=>{ + const f=fixture();const entered=deferred();const reply=deferred(); + const q=new HostDialogQueue('account/host/session-a','session-a',async request=>{ + if(request.action==='submit'){entered.resolve();await reply.promise;} + return f.invoke(request); + },f.storage); + const sending=q.submit(message);const rejected=assert.rejects(sending,/Connection lost/); + await entered.promise;await q.refresh();assert.equal(q.getSnapshot().pending.length,0); + reply.reject(new Error('Connection lost'));await rejected; + assert.equal(q.getSnapshot().pending.length,1); + assert.match(q.getSnapshot().error,/Connection lost/); + const id=q.getSnapshot().pending[0].request.message.turnId; + const reopened=f.create();await reopened.refresh();await reopened.retry(reopened.getSnapshot().pending[0]); + assert.equal(f.calls.find(call=>call.action==='submit').message.turnId,id); + assert.equal(f.executions,1); +}); diff --git a/src/shared/dialog-queue/HostDialogQueue.ts b/src/shared/dialog-queue/HostDialogQueue.ts index 3a21b40a6f..9f615818c8 100644 --- a/src/shared/dialog-queue/HostDialogQueue.ts +++ b/src/shared/dialog-queue/HostDialogQueue.ts @@ -73,6 +73,10 @@ export class HostDialogQueue { private listeners = new Set<() => void>(); private refreshing: Promise | null = null; private mutation: Promise = Promise.resolve(); + // Durable outbox entries also exist during healthy RPCs. Only unresolved + // delivery needs recovery UI; a new page has no live RPCs and shows all of it. + private inFlight = new Set(); + private pendingRead = 0; constructor(readonly scope: string, readonly sessionId: string, private invoke: (request: QueueRequest) => Promise, private storage: QueueStorage = queueStorage) {} getSnapshot = (): QueueView => this.view; @@ -82,6 +86,13 @@ export class HostDialogQueue { private publish(patch: Partial): void { this.view = { ...this.view, ...patch }; for (const listener of this.listeners) listener(); } + private async publishPending(): Promise { + const read = ++this.pendingRead; + const records = await this.storage.list(this.scope); + if (read !== this.pendingRead) return; + this.publish({ pending: records.filter(record => !this.inFlight.has(record.key) + && (!record.accepted || record.restoreIntent)) }); + } private accept(snapshot: QueueSnapshot, authoritative: boolean): void { if (snapshot.sessionId !== this.sessionId || !snapshot.queueEpoch || !Number.isSafeInteger(snapshot.revision)) { throw new Error('Invalid host queue snapshot'); @@ -100,7 +111,7 @@ export class HostDialogQueue { const records = await this.storage.list(this.scope); const current = this.view.snapshot!; for (const record of records) { - if (!record.accepted || record.restoreIntent || record.request.action !== 'submit') continue; + if (this.inFlight.has(record.key) || !record.accepted || record.restoreIntent || record.request.action !== 'submit') continue; if (record.request.queueEpoch !== current.queueEpoch) { // A restarted owner cannot vouch for an accepted in-memory message. // Preserve the original draft and require an explicit user decision. @@ -109,7 +120,7 @@ export class HostDialogQueue { await this.storage.remove(record.key); } } - this.publish({ pending: (await this.storage.list(this.scope)).filter(record => !record.accepted || record.restoreIntent) }); + await this.publishPending(); return current; } catch (error) { this.publish({ error: String(error) }); throw error; } finally { this.refreshing = null; } @@ -122,10 +133,11 @@ export class HostDialogQueue { return result; } private async transmit(record: QueueOutboxRecord): Promise { - // The transaction commits before an RPC is permitted to leave this client. - await this.storage.put(record); - this.publish({ pending: (await this.storage.list(this.scope)).filter(record => !record.accepted || record.restoreIntent) }); + this.inFlight.add(record.key); try { + // The transaction commits before an RPC is permitted to leave this client. + await this.storage.put(record); + await this.publishPending(); const snapshot = await this.invoke(record.request); if (snapshot.queueEpoch !== record.request.queueEpoch) throw new Error('queue_scope_expired: host queue owner changed'); if (record.request.action === 'submit' && snapshot.receipt?.turnId !== record.request.message.turnId) { @@ -134,15 +146,16 @@ export class HostDialogQueue { this.accept(snapshot, false); if (record.request.action === 'submit') await this.storage.put({ ...record, accepted: true }); else await this.storage.remove(record.key); - this.publish({ pending: (await this.storage.list(this.scope)).filter(record => !record.accepted || record.restoreIntent) }); return snapshot; } catch (error) { const text = String(error); if (/queue_conflict:|too_late:|idempotency_conflict:|Message queue is full|Queue is blocked by interrupted/.test(text)) { await this.storage.remove(record.key); - this.publish({ pending: (await this.storage.list(this.scope)).filter(record => !record.accepted || record.restoreIntent) }); } this.publish({ error: text }); throw error; + } finally { + this.inFlight.delete(record.key); + await this.publishPending(); } } submit(message: Omit, draft?: unknown, requestedTurnId?: string): Promise { @@ -190,7 +203,7 @@ export class HostDialogQueue { if (result.receipt) { this.accept(result, false); await this.storage.put({ ...record, accepted: true }); - this.publish({ pending: (await this.storage.list(this.scope)).filter(record => !record.accepted || record.restoreIntent) }); + await this.publishPending(); return result; } } @@ -201,7 +214,7 @@ export class HostDialogQueue { async prepareRestore(record: QueueOutboxRecord): Promise { if (record.scope !== this.scope) throw new Error('Queue target changed'); await this.storage.put({ ...record, restoreIntent: true }); - this.publish({ pending: (await this.storage.list(this.scope)).filter(record => !record.accepted || record.restoreIntent) }); + await this.publishPending(); } async receipt(turnId: string, expectedEpoch?: string): Promise { const snapshot = await this.refresh(); @@ -218,7 +231,7 @@ export class HostDialogQueue { async dismiss(record: QueueOutboxRecord): Promise { if (record.scope !== this.scope) throw new Error('Queue target changed'); await this.storage.remove(record.key); - this.publish({ pending: (await this.storage.list(this.scope)).filter(record => !record.accepted || record.restoreIntent) }); + await this.publishPending(); } } diff --git a/src/web-ui/src/flow_chat/components/HostPendingQueuePanel.test.tsx b/src/web-ui/src/flow_chat/components/HostPendingQueuePanel.test.tsx new file mode 100644 index 0000000000..348b570de4 --- /dev/null +++ b/src/web-ui/src/flow_chat/components/HostPendingQueuePanel.test.tsx @@ -0,0 +1,58 @@ +/** @vitest-environment jsdom */ +import { act } from 'react'; +import { createRoot } from 'react-dom/client'; +import { expect, it, vi } from 'vitest'; +import { HostDialogQueue, type QueueOutboxRecord, type QueueSnapshot } from '../../../../shared/dialog-queue/HostDialogQueue'; +import { HostPendingQueuePanel } from './HostPendingQueuePanel'; + +vi.mock('@/infrastructure/i18n', () => ({ useI18n: () => ({ t: (key: string) => key }) })); +vi.mock('../services/flow-chat-manager/PendingQueueModule', () => ({ pendingQueueManager: { enqueue: vi.fn() } })); + +it.each([false, true])('keeps a normal send hidden and shows recovery only on failure (failure=%s)', async failure => { + const records = new Map(); + let finish!: () => void; + let entered!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const acknowledgement = new Promise(resolve => { finish = resolve; }); + const snapshot: QueueSnapshot = { sessionId: 'session', queueEpoch: 'epoch', revision: 0, + activeTurnId: null, items: [], capacity: 20, used: 0, receipt: null }; + const queue = new HostDialogQueue('scope', 'session', async request => { + if (request.action !== 'submit') return snapshot; + entered(); + await acknowledgement; + if (failure) throw new Error('Connection lost'); + return { ...snapshot, revision: 1, receipt: { turnId: request.message.turnId, + displayContent: request.message.content, status: 'started', previewTruncated: false, + attachmentCount: 0, agentType: 'Standard', createdAtMs: 1, reason: null, + targetTurnId: null, steeringId: null } }; + }, { + list: async () => [...records.values()], + put: async record => { records.set(record.key, record); }, + remove: async key => { records.delete(key); }, + }); + const container = document.createElement('div'); + document.body.append(container); + const root = createRoot(container); + try { + await act(async () => root.render( true} />)); + let sending!: Promise; + await act(async () => { + sending = queue.submit({ content: 'hello', agentType: 'Standard', attachments: [], metadata: {} }).catch(() => undefined); + await started; + await queue.refresh(); + }); + expect(records.size).toBe(1); + expect(container.textContent).toBe(''); + await act(async () => { finish(); await sending; }); + if (failure) { + expect(container.textContent).toContain('hostQueue.unknown'); + expect(container.textContent).toContain('hello'); + } else { + expect(container.textContent).toBe(''); + } + } finally { + finish(); + await act(async () => root.unmount()); + container.remove(); + } +});