From 0c6b296c1fe491d9ae374962a8698112961024d0 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sun, 16 Aug 2026 19:31:08 -0700 Subject: [PATCH] fix(peer): keep live turns attached after a device switch A surface switch is a view change, not a change of execution. Locally created sessions stay at historyState new, so attach must still run, and a turn already submitted to the host must not be re-queued or drained while the projection is being repaired. --- .../flow-chat-manager/MessageModule.test.ts | 15 ++ .../flow-chat-manager/MessageModule.ts | 9 +- .../PeerSessionRefreshModule.test.ts | 252 ++++++++++++++++++ .../PeerSessionRefreshModule.ts | 97 +++++-- .../PendingQueueModule.test.ts | 47 +++- .../flow-chat-manager/PendingQueueModule.ts | 62 ++++- .../dispatch/DispatchSessionDriver.ts | 1 + .../local/LocalSessionDriver.ts | 4 + .../src/flow_chat/session-drivers/types.ts | 7 + .../src/infrastructure/peer-device/README.md | 19 +- .../peer-device/runtimeSessionEventGate.ts | 7 + 11 files changed, 493 insertions(+), 27 deletions(-) diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts index 55274b5653..ad86866b91 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts @@ -1117,6 +1117,21 @@ describe('MessageModule device surface switch', () => { expect(mockNotificationError).not.toHaveBeenCalled(); }); + it('does not re-queue when startDialogTurn was sent and the invoke then fails after a switch', async () => { + const { context, switchSurface } = switchingContext('session-accepted'); + mockStartDialogTurn.mockImplementation(async () => { + switchSurface(); + throw new Error('invoke aborted after transport swap'); + }); + + await sendMessage(context, 'already running', 'session-accepted'); + + expect(mockStartDialogTurn).toHaveBeenCalledTimes(1); + expect(mockPendingEnqueueForSurface).not.toHaveBeenCalled(); + expect(context.flowChatStore.abandonOptimisticDialogTurn).not.toHaveBeenCalled(); + expect(mockNotificationError).not.toHaveBeenCalled(); + }); + it('still reports ordinary failures when the surface did not change', async () => { const { context } = switchingContext('session-normal'); mockStartDialogTurn.mockRejectedValue(new Error('backend exploded')); diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts index 8be883f5c1..1f42704860 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts @@ -11,6 +11,7 @@ import { stateMachineManager } from '../../state-machine'; import { SessionExecutionEvent, SessionExecutionState } from '../../state-machine/types'; import { createLogger } from '@/shared/utils/logger'; import { + getActiveSurfaceId, getActiveSurfaceScope, isSurfaceChangedError, type DeviceSurfaceId, @@ -19,6 +20,7 @@ import type { FlowChatContext } from './types'; import { isProjectedSessionEmpty } from '../../utils/flowChatTurnIdentity'; import type { ImageContextData as ImageInputContextData } from '@/infrastructure/api/service-api/ImageContextTypes'; import { pendingQueueManager } from './PendingQueueModule'; +import { isRuntimeSessionAttachmentInFlight } from '@/infrastructure/peer-device/runtimeSessionEventGate'; import { isSessionInUseError } from '@/infrastructure/api/errors/TauriCommandError'; import { i18nService } from '@/infrastructure/i18n'; import { driverForSession } from '../../session-drivers/registry'; @@ -117,7 +119,9 @@ function recoverSubmissionAfterSurfaceSwitch( options?: SendMessageOptions; }, ): void { - if (turnTracker.hostAcceptedTurn) { + const hostHasTurn = + turnTracker.hostAcceptedTurn || turnTracker.hostSubmitStarted === true; + if (hostHasTurn) { log.info('Device surface switched after the host accepted the turn; it keeps running there', { sessionId, }); @@ -523,6 +527,9 @@ export async function drainPendingQueue( sessionId: string, options?: { allowInterruptedRecoveryAbandon?: boolean }, ): Promise { + if (isRuntimeSessionAttachmentInFlight(getActiveSurfaceId(), sessionId)) { + return; + } const machineState = stateMachineManager.getCurrentState(sessionId); if (machineState !== SessionExecutionState.IDLE) { return; diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.test.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.test.ts index b6d57f68ef..b43e913e33 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.test.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.test.ts @@ -32,6 +32,7 @@ vi.mock('../AgenticEventListener', () => ({ import { installPeerSessionRefresh, + isSessionProjectionAttachable, PEER_SESSION_REFRESH_INTERVAL_MS, requestPeerSessionRefresh, } from './PeerSessionRefreshModule'; @@ -430,3 +431,254 @@ describe('PeerSessionRefreshModule dead subscription recovery', () => { cleanup(); }); }); + +describe('isSessionProjectionAttachable', () => { + const base = { + workspacePath: '/repo/BitFun', + isTransient: false, + isHistorical: false, + historyState: 'ready' as const, + }; + + it('accepts a hydrated live session', () => { + expect(isSessionProjectionAttachable(base)).toBe(true); + }); + + it('accepts a locally created session that never left historyState new', () => { + expect(isSessionProjectionAttachable({ ...base, historyState: 'new' })).toBe(true); + }); + + it('rejects metadata-only, hydrating, failed, historical, and transient shells', () => { + expect(isSessionProjectionAttachable({ ...base, historyState: 'metadata-only' })).toBe(false); + expect(isSessionProjectionAttachable({ ...base, historyState: 'hydrating' })).toBe(false); + expect(isSessionProjectionAttachable({ ...base, historyState: 'failed' })).toBe(false); + expect(isSessionProjectionAttachable({ ...base, isHistorical: true })).toBe(false); + expect(isSessionProjectionAttachable({ ...base, isTransient: true })).toBe(false); + expect(isSessionProjectionAttachable({ ...base, workspacePath: ' ' })).toBe(false); + expect(isSessionProjectionAttachable(null)).toBe(false); + }); +}); + +describe('PeerSessionRefreshModule attach eligibility after a surface switch', () => { + beforeEach(() => { + vi.useFakeTimers(); + peerModeMock.active = true; + resetRuntimeSessionEventGateForTest(); + stateMachineMock.get.mockReturnValue({ + getCurrentState: () => 'idle', + getContext: () => ({ lastUpdateTime: 0, version: 0 }), + }); + vi.stubGlobal('document', { + visibilityState: 'visible', + addEventListener: vi.fn(), + removeEventListener: vi.fn(), + }); + vi.stubGlobal('window', { addEventListener: vi.fn(), removeEventListener: vi.fn() }); + }); + + afterEach(() => { + vi.clearAllMocks(); + vi.unstubAllGlobals(); + vi.useRealTimers(); + }); + + it('attaches a locally created session that is still historyState new', async () => { + const refreshPeerSessionSnapshot = vi.fn(async () => ({ + applied: false, + backendState: 'Processing { current_turn_id: "turn-live", phase: Streaming }', + latestTurnId: 'turn-live', + latestTurnStatus: 'processing', + })); + const state = { + activeSessionId: 'session-new', + sessions: new Map([ + ['session-new', { + sessionId: 'session-new', + workspacePath: '/repo/BitFun', + historyState: 'new', + isHistorical: false, + isTransient: false, + dialogTurns: [], + }], + ]), + }; + const context = { + flowChatStore: { + getState: () => state, + subscribeSelector: vi.fn(() => () => {}), + refreshPeerSessionSnapshot, + }, + eventBatcher: { flushNow: vi.fn(), clear: vi.fn() }, + contentBuffers: new Map(), + activeTextItems: new Map(), + } as any; + + const cleanup = installPeerSessionRefresh(context); + await vi.advanceTimersByTimeAsync(1); + + expect(refreshPeerSessionSnapshot).toHaveBeenCalledWith( + 'session-new', + '/repo/BitFun', + expect.objectContaining({ requireActiveSession: false }), + ); + cleanup(); + }); + + it('does not attach a metadata-only historical shell', async () => { + const refreshPeerSessionSnapshot = vi.fn(); + const state = { + activeSessionId: 'session-meta', + sessions: new Map([ + ['session-meta', { + sessionId: 'session-meta', + workspacePath: '/repo/BitFun', + historyState: 'metadata-only', + isHistorical: true, + isTransient: false, + }], + ]), + }; + const context = { + flowChatStore: { + getState: () => state, + subscribeSelector: vi.fn(() => () => {}), + refreshPeerSessionSnapshot, + }, + eventBatcher: { flushNow: vi.fn(), clear: vi.fn() }, + contentBuffers: new Map(), + activeTextItems: new Map(), + } as any; + + const cleanup = installPeerSessionRefresh(context); + await vi.advanceTimersByTimeAsync(1); + + expect(refreshPeerSessionSnapshot).not.toHaveBeenCalled(); + cleanup(); + }); + + it('attaches a background session named by a dropped-event refresh', async () => { + const refreshPeerSessionSnapshot = vi.fn(async () => ({ + applied: false, + backendState: 'Processing { current_turn_id: "turn-bg", phase: Streaming }', + latestTurnId: 'turn-bg', + latestTurnStatus: 'processing', + })); + const state = { + activeSessionId: 'session-active', + sessions: new Map([ + ['session-active', { + sessionId: 'session-active', + workspacePath: '/repo/BitFun', + historyState: 'new', + isHistorical: false, + isTransient: false, + }], + ['session-bg', { + sessionId: 'session-bg', + workspacePath: '/repo/BitFun', + historyState: 'new', + isHistorical: false, + isTransient: false, + }], + ]), + }; + const context = { + flowChatStore: { + getState: () => state, + subscribeSelector: vi.fn(() => () => {}), + refreshPeerSessionSnapshot, + }, + eventBatcher: { flushNow: vi.fn(), clear: vi.fn() }, + contentBuffers: new Map(), + activeTextItems: new Map(), + } as any; + + const cleanup = installPeerSessionRefresh(context); + await vi.advanceTimersByTimeAsync(1); + refreshPeerSessionSnapshot.mockClear(); + + requestPeerSessionRefresh('session-bg'); + await vi.advanceTimersByTimeAsync(1); + + expect(refreshPeerSessionSnapshot).toHaveBeenCalledWith( + 'session-bg', + '/repo/BitFun', + expect.objectContaining({ requireActiveSession: false }), + ); + cleanup(); + }); + + it('repairs a second live session queued while the first attach is in flight', async () => { + let releaseFirst: ((value: { + applied: boolean; + backendState: string; + latestTurnId: string; + latestTurnStatus: string; + }) => void) | undefined; + const refreshPeerSessionSnapshot = vi.fn((sessionId: string) => { + if (sessionId === 'session-active') { + return new Promise(resolve => { + releaseFirst = resolve; + }); + } + return Promise.resolve({ + applied: false, + backendState: 'Processing { current_turn_id: "turn-bg", phase: Streaming }', + latestTurnId: 'turn-bg', + latestTurnStatus: 'processing', + }); + }); + const state = { + activeSessionId: 'session-active', + sessions: new Map([ + ['session-active', { + sessionId: 'session-active', + workspacePath: '/repo/BitFun', + historyState: 'new', + isHistorical: false, + isTransient: false, + }], + ['session-bg', { + sessionId: 'session-bg', + workspacePath: '/repo/BitFun', + historyState: 'new', + isHistorical: false, + isTransient: false, + }], + ]), + }; + const context = { + flowChatStore: { + getState: () => state, + subscribeSelector: vi.fn(() => () => {}), + refreshPeerSessionSnapshot, + }, + eventBatcher: { flushNow: vi.fn(), clear: vi.fn() }, + contentBuffers: new Map(), + activeTextItems: new Map(), + } as any; + + const cleanup = installPeerSessionRefresh(context); + await vi.advanceTimersByTimeAsync(1); + expect(refreshPeerSessionSnapshot).toHaveBeenCalledTimes(1); + + requestPeerSessionRefresh('session-bg'); + await vi.advanceTimersByTimeAsync(1); + expect(refreshPeerSessionSnapshot).toHaveBeenCalledTimes(1); + + releaseFirst?.({ + applied: false, + backendState: 'Processing { current_turn_id: "turn-active", phase: Streaming }', + latestTurnId: 'turn-active', + latestTurnStatus: 'processing', + }); + await vi.advanceTimersByTimeAsync(1); + + expect(refreshPeerSessionSnapshot).toHaveBeenCalledWith( + 'session-bg', + '/repo/BitFun', + expect.objectContaining({ requireActiveSession: false }), + ); + cleanup(); + }); +}); diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.ts index 40bc3f2394..2ba3599583 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.ts @@ -29,9 +29,10 @@ import { SessionExecutionEvent, SessionExecutionState, } from '../../state-machine/types'; -import type { AnyFlowItem, DialogTurn } from '../../types/flow-chat'; +import type { AnyFlowItem, DialogTurn, Session } from '../../types/flow-chat'; import { installLiveSessionInteractionMailbox } from '../liveSessionInteractionStore'; import { agenticEventListener } from '../AgenticEventListener'; +import { pendingQueueManager } from './PendingQueueModule'; import type { FlowChatContext } from './types'; const log = createLogger('PeerSessionRefresh'); @@ -39,6 +40,39 @@ const log = createLogger('PeerSessionRefresh'); export const PEER_SESSION_REFRESH_INTERVAL_MS = 3000; export const PEER_SESSION_STREAM_STALE_MS = 6000; +/** + * A session can attach to the host Runtime projection once it has a usable + * live shell on this surface. + * + * `historyState === 'ready'` is the hydrated-from-disk path. Locally created + * sessions stay at `'new'` for their whole window — they never pass through + * disk hydrate — and after a surface switch they are exactly the sessions + * that missed DialogTurnStarted. Treating `'new'` as unready permanently + * disables the only repair path: later chunks arrive against an idle machine + * and are dropped, and the composer queues follow-up messages behind a turn + * that the UI can no longer update. + */ +type AttachableSession = Pick< + Session, + 'workspacePath' | 'isTransient' | 'isHistorical' | 'historyState' +> & { workspacePath: string }; + +export function isSessionProjectionAttachable( + session: Pick< + Session, + 'workspacePath' | 'isTransient' | 'isHistorical' | 'historyState' + > | null | undefined, +): session is AttachableSession { + const workspacePath = session?.workspacePath?.trim(); + return Boolean( + session && + workspacePath && + !session.isTransient && + !session.isHistorical && + (session.historyState === 'ready' || session.historyState === 'new'), + ); +} + type RefreshRequester = (sessionId?: string) => void; let installedRefreshRequester: RefreshRequester | null = null; @@ -135,16 +169,29 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void installLiveSessionInteractionMailbox(); let disposed = false; let inFlight = false; - let queued = false; + const queuedSessionIds = new Set(); let immediateTimer: ReturnType | null = null; + function enqueueFollowUpRefresh(sessionId?: string): void { + queuedSessionIds.add(sessionId); + } + + function drainFollowUpRefresh(): void { + if (disposed || queuedSessionIds.size === 0) { + return; + } + const next = queuedSessionIds.values().next().value; + queuedSessionIds.delete(next); + scheduleRefresh(next); + } + async function runRefresh( requestedSessionId?: string, staleOnly = false, ): Promise { if (disposed || inFlight || !isSurfaceReconcileEnabled()) { if (inFlight) { - queued = true; + enqueueFollowUpRefresh(requestedSessionId); } return; } @@ -162,20 +209,22 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void const state = context.flowChatStore.getState(); const sessionId = requestedSessionId || state.activeSessionId; - if (!sessionId || state.activeSessionId !== sessionId) { + // Attach is per (surface, session), not per the tab currently focused. + // After a switch, every live session on this surface may have missed + // DialogTurnStarted; dropped-event refresh requests name those sessions + // explicitly even when they are not active. + if (!sessionId) { return; } const session = state.sessions.get(sessionId); - const workspacePath = session?.workspacePath?.trim(); - if ( - !session || - !workspacePath || - session.isTransient || - session.isHistorical || - session.historyState !== 'ready' - ) { + if (!isSessionProjectionAttachable(session)) { return; } + const workspacePath = session.workspacePath.trim(); + pendingQueueManager.reconcileAgainstLiveTurns( + sessionId, + state.sessions.get(sessionId)?.dialogTurns ?? [], + ); const machine = stateMachineManager.get(sessionId); const machineState = machine?.getCurrentState() ?? SessionExecutionState.IDLE; @@ -205,7 +254,10 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void workspacePath, { replaceRunningSnapshot, - requireActiveSession: true, + // A background session on this surface still owns its projection. + // Requiring the focused tab aborted the dropped-event repair for + // every non-active running chat after a multi-session switch. + requireActiveSession: false, shouldApply: () => { if (!isSurfaceReconcileEnabled()) { return false; @@ -218,6 +270,13 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void }, ); surfaceScope.assertCurrent('attachRuntimeSession'); + const restoredSession = context.flowChatStore.getState().sessions.get(sessionId); + if (restoredSession) { + pendingQueueManager.reconcileAgainstLiveTurns( + sessionId, + restoredSession.dialogTurns, + ); + } if (result.runtimeEventSnapshot) { if (result.runtimeEventReplayRequired === false) { @@ -344,10 +403,7 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void attachment.abort({ discard: !surfaceScope.isCurrent() }); } inFlight = false; - if (queued && !disposed) { - queued = false; - scheduleRefresh(); - } + drainFollowUpRefresh(); } } @@ -355,8 +411,11 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void if (disposed) { return; } - if (immediateTimer !== null) { - clearTimeout(immediateTimer); + // Do not replace a pending attach with a later request: two sessions + // that drop events in the same tick must both be repaired. + if (inFlight || immediateTimer !== null) { + enqueueFollowUpRefresh(sessionId); + return; } immediateTimer = setTimeout(() => { immediateTimer = null; diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts index a508841f69..33e784607e 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts @@ -1,7 +1,10 @@ // @vitest-environment jsdom import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { pendingQueueManager } from './PendingQueueModule'; +import { + pendingQueueManager, + queuedItemDuplicatesLiveTurn, +} from './PendingQueueModule'; import { LOCAL_SURFACE_ID, activateSurface, @@ -83,4 +86,46 @@ describe('PendingQueueModule', () => { pendingQueueManager.clearSurface('peer-b'); activateSurface(LOCAL_SURFACE_ID); }); + + it('drops a queued duplicate of a live turn after a surface switch', () => { + const sessionId = testSession(); + pendingQueueManager.enqueue({ + sessionId, + content: '详细分析项目，然后调用 askuserquestion 随便问我几个当前项目相关的问题吧', + initialStatus: 'failed', + }); + pendingQueueManager.enqueue({ + sessionId, + content: 'a later follow-up that should stay', + }); + + const removed = pendingQueueManager.reconcileAgainstLiveTurns(sessionId, [ + { + id: 'dialog_live', + status: 'processing', + userMessage: { + id: 'user-1', + content: '详细分析项目，然后调用 askuserquestion 随便问我几个当前项目相关的问题吧', + timestamp: Date.now(), + }, + }, + ]); + + expect(removed).toBe(1); + expect(pendingQueueManager.list(sessionId).map(item => item.content)).toEqual([ + 'a later follow-up that should stay', + ]); + expect(queuedItemDuplicatesLiveTurn( + { content: '详细分析项目，然后调用 askuserquestion 随便问我几个当前项目相关的问题吧' }, + [{ + id: 'dialog_live', + status: 'processing', + userMessage: { + id: 'user-1', + content: '详细分析项目，然后调用 askuserquestion 随便问我几个当前项目相关的问题吧', + timestamp: Date.now(), + }, + }], + )).toBe(true); + }); }); diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts index b488b75f49..7a2c218cea 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts @@ -13,7 +13,7 @@ */ import { createLogger } from '@/shared/utils/logger'; -import type { QueuedMessage } from '../../types/flow-chat'; +import type { DialogTurn, QueuedMessage } from '../../types/flow-chat'; import { getActiveSurfaceId, onSurfaceActivated, @@ -30,6 +30,36 @@ const STORAGE_PREFIX = 'flowChat.pendingQueue.'; const PEER_STORAGE_PREFIX = 'flowChat.peerPendingQueue.v1.'; const MAX_QUEUE_DEPTH = 20; +const LIVE_TURN_STATUSES = new Set([ + 'pending', + 'image_analyzing', + 'processing', + 'finishing', +]); + +function normalizeQueueText(value: string | undefined): string { + return value?.trim() ?? ''; +} + +export function queuedItemDuplicatesLiveTurn( + item: Pick, + turns: Array>, +): boolean { + const itemTexts = [item.displayMessage, item.content] + .map(normalizeQueueText) + .filter(text => text.length > 0); + return turns.some(turn => { + if (!LIVE_TURN_STATUSES.has(turn.status)) { + return false; + } + if (item.localDialogTurnId && item.localDialogTurnId === turn.id) { + return true; + } + const turnText = normalizeQueueText(turn.userMessage?.content); + return Boolean(turnText && itemTexts.includes(turnText)); + }); +} + export interface EnqueueInput { sessionId: string; content: string; @@ -214,6 +244,36 @@ class PendingQueueManager { return item; } + /** + * Drop queue items that are the same user message as a turn this surface + * is already projecting as live. A device switch must not keep a duplicate + * "pending send" of a turn the host is executing. + */ + reconcileAgainstLiveTurns( + sessionId: string, + turns: Array>, + surfaceId = getActiveSurfaceId(), + ): number { + const key = this.queueKey(sessionId, surfaceId); + const items = this.queues.get(key); + if (!items || items.length === 0) { + return 0; + } + const next = items.filter(item => !queuedItemDuplicatesLiveTurn(item, turns)); + const removed = items.length - next.length; + if (removed === 0) { + return 0; + } + if (next.length === 0) { + this.queues.delete(key); + } else { + this.queues.set(key, next); + } + this.persist(sessionId, surfaceId); + this.notifySurface(surfaceId, sessionId); + return removed; + } + update( sessionId: string, id: string, diff --git a/src/web-ui/src/flow_chat/session-drivers/dispatch/DispatchSessionDriver.ts b/src/web-ui/src/flow_chat/session-drivers/dispatch/DispatchSessionDriver.ts index 842078ea47..a5c2bfca46 100644 --- a/src/web-ui/src/flow_chat/session-drivers/dispatch/DispatchSessionDriver.ts +++ b/src/web-ui/src/flow_chat/session-drivers/dispatch/DispatchSessionDriver.ts @@ -743,6 +743,7 @@ export const dispatchSessionDriver: SessionDriver = { label: i18nService.t('flow-chat:chatInput.dispatch.transferInProgress'), }); let response: Awaited>; + tracker.hostSubmitStarted = true; try { response = await dispatchApi.submit({ target: targetRequest, diff --git a/src/web-ui/src/flow_chat/session-drivers/local/LocalSessionDriver.ts b/src/web-ui/src/flow_chat/session-drivers/local/LocalSessionDriver.ts index ba0ed33bce..9ec277a6f9 100644 --- a/src/web-ui/src/flow_chat/session-drivers/local/LocalSessionDriver.ts +++ b/src/web-ui/src/flow_chat/session-drivers/local/LocalSessionDriver.ts @@ -430,6 +430,7 @@ export const localSessionDriver: SessionDriver = { const projectWorkspacePath = sessionProjectWorkspacePath(updatedSession); if (acpClientId) { + tracker.hostSubmitStarted = true; await ACPClientAPI.startDialogTurn({ sessionId, clientId: acpClientId, @@ -447,6 +448,7 @@ export const localSessionDriver: SessionDriver = { context.flowChatStore.updateSessionLastSubmittedMode(sessionId, currentAgentType); } else { try { + tracker.hostSubmitStarted = true; await agentAPI.startDialogTurn({ sessionId: sessionId, userInput: message, @@ -470,6 +472,7 @@ export const localSessionDriver: SessionDriver = { sessionId: sessionId, dialogTurnsCount: updatedSession.dialogTurns.length }); + tracker.hostSubmitStarted = false; // Lazy import: SessionModule routes lifecycle calls through the // driver registry, so a static import would create a module cycle. @@ -479,6 +482,7 @@ export const localSessionDriver: SessionDriver = { await retryCreateBackendSession(context, sessionId); surfaceScope.assertCurrent('retry backend session creation'); + tracker.hostSubmitStarted = true; await agentAPI.startDialogTurn({ sessionId: sessionId, userInput: message, diff --git a/src/web-ui/src/flow_chat/session-drivers/types.ts b/src/web-ui/src/flow_chat/session-drivers/types.ts index 35bd9fdd57..60043c20c7 100644 --- a/src/web-ui/src/flow_chat/session-drivers/types.ts +++ b/src/web-ui/src/flow_chat/session-drivers/types.ts @@ -99,6 +99,13 @@ export interface TurnTracker { * must recover the message rather than assume it is running somewhere. */ hostAcceptedTurn: boolean; + /** + * Set immediately before `start_dialog_turn` / ACP start is invoked. + * The host may accept the turn before the client sees the ACK — a surface + * switch that aborts the invoke must not re-queue a message the host is + * already executing. + */ + hostSubmitStarted?: boolean; } /** diff --git a/src/web-ui/src/infrastructure/peer-device/README.md b/src/web-ui/src/infrastructure/peer-device/README.md index ecb63934ea..cd89b83104 100644 --- a/src/web-ui/src/infrastructure/peer-device/README.md +++ b/src/web-ui/src/infrastructure/peer-device/README.md @@ -32,11 +32,16 @@ Controller-side React/transport layer for Peer Device Mode. Architecture: inside that window made the submission resume against a missing session and throw `Session lost after adding dialog turn` — before `start_dialog_turn`, so the message reached no host at all (regression: - 2026-08-15). `resetProductSurface` therefore awaits + 2026-08-15). `resetProductSurface` therefore awaits `waitForInFlightSubmissions` first. `sendMessage` and its driver carry one `SurfaceScope`; after every host await, a stale epoch abandons without writing into the newly selected container, and an unaccepted message is - re-queued onto its original surface. Any new await inside `startTurn` + re-queued onto its original surface. Once `start_dialog_turn` has been + invoked, the host may already own the Turn before the client sees the + ACK — that submission must not be re-queued, and attach must drop any + pending-queue item that duplicates a live turn's user message. Drain + must not fire while a Runtime attach is resetting the state machine to + IDLE. Any new await inside `startTurn` widens that window and must keep the same scope checkpoint. - **Reconciliation repairs a projection, never guts it.** The wholesale replace path (`replaceRunningSnapshot`) skips the forward-progress @@ -139,9 +144,13 @@ Controller-side React/transport layer for Peer Device Mode. Architecture: `isSurfaceReconcileEnabled()`, **not** on Peer Mode: once a window has switched surface, a turn left running on the local device also needs the same attach, because its live events were dropped by surface routing while - another device was rendered. Attach is requested as soon as active Session - hydration becomes ready; the 3s loop is only a liveness retry and an - older-Host fallback. The Peer Host must + another device was rendered. Attach is requested as soon as a Session on + this surface has a usable live projection: `historyState === 'ready'` + after disk hydrate, or `historyState === 'new'` for a session created in + this window (those never become `ready` via hydrate). The gate is per + `(DeviceSurfaceId, SessionId)`, not per the focused tab — a dropped-event + refresh must still attach a background session that kept running here. + The 3s loop is only a liveness retry and an older-Host fallback. The Peer Host must overlay its live in-memory session state on the persisted view; otherwise an in-progress turn is normalized as interrupted history and later chunks are dropped by the controller state machine. Surface epoch checks reject a diff --git a/src/web-ui/src/infrastructure/peer-device/runtimeSessionEventGate.ts b/src/web-ui/src/infrastructure/peer-device/runtimeSessionEventGate.ts index a441fa8152..3d3b665522 100644 --- a/src/web-ui/src/infrastructure/peer-device/runtimeSessionEventGate.ts +++ b/src/web-ui/src/infrastructure/peer-device/runtimeSessionEventGate.ts @@ -132,6 +132,13 @@ export function subscribeRuntimeSessionEventGaps( return () => gapListeners.delete(listener); } +export function isRuntimeSessionAttachmentInFlight( + surfaceId: DeviceSurfaceId, + sessionId: string, +): boolean { + return attachments.has(attachmentKey(surfaceId, sessionId)); +} + export function beginRuntimeSessionAttachment( surfaceId: DeviceSurfaceId, sessionId: string,