fix(flow-chat): guard peer turn, cancel, queue and history state races

src/web-ui/src/flow_chat/components/modern/ModernFlowChatContainer.history-restore.test.tsx

@@ -14,20 +14,25 @@ const fixture = vi.hoisted(() => ({
   restore: vi.fn(),
   restoreTail: vi.fn(),
   loadTurnWindow: vi.fn(),
+  synchronize: vi.fn(),
+  extend: vi.fn(),
   present: null as ((sessionId: string, presentation: SessionHistoryPresentation) => void) | null,
 }));
 vi.mock('../../store/FlowChatStore', () => ({ flowChatStore: {
   getState: () => ({ activeSessionId: fixture.session?.sessionId, sessions: new Map(fixture.session ? [[fixture.session.sessionId, fixture.session]] : []) }),
   retainSessionHistory: () => () => {},
   subscribeSelector: () => () => {},
-  getSessionHistoryViewState: () => undefined,
+  getSessionHistoryViewState: () => ({
+    catalog: fixture.session?.turnCatalog ?? null, loadedRanges: [], activeRange: fixture.cached?.range ?? null,
+  }),
+  synchronizeSessionHistory: (...args: unknown[]) => fixture.synchronize(...args),
   reactivateSessionHistoryWindow: () => fixture.cached,
   restoreSessionHistoryWindow: (...args: unknown[]) => fixture.restore(...args),
   restoreSessionTailPresentation: (...args: unknown[]) => fixture.restoreTail(...args),
   hasPendingSessionHistoryCompletion: () => false,
   hasDeferredSessionHistoryProjection: () => false,
   releaseSessionHistoryCompletionAfterInitialPaint: () => false,
-  extendSessionHistoryWindow: () => null,
+  extendSessionHistoryWindow: (...args: unknown[]) => fixture.extend(...args),
   loadSessionTurnWindow: (...args: unknown[]) => fixture.loadTurnWindow(...args),
   loadPreviousSessionTurnWindow: (...args: unknown[]) => fixture.loadTurnWindow(...args),
   activateSessionHistoryWindow: () => fixture.cached,
@@ -45,6 +50,7 @@ vi.mock('./VirtualMessageList', () => ({ VirtualMessageList: React.forwardRef((p
   fixture.listProps = props;
   React.useImperativeHandle(ref, () => ({
     scrollToTurnEnd: () => true,
+    scrollToLatestEndPosition: () => {},
     captureViewportSnapshot: () => null,
     clearSearchMatch: () => {},
     prepareTurnNavigation: () => 'pending',
@@ -103,6 +109,14 @@ function session(totalTurnCount = 8, historyState: Session['historyState'] = 're
     }],
   } as Session;
 }
+function sessionWithCatalog(count: number, revision: string): Session {
+  return { ...session(count), turnCatalog: {
+    schemaVersion: 1, sessionId: 'session-a', revision, totalTurnCount: count,
+    complete: true, entries: Array.from({ length: count }, (_, ordinal) => ({
+      ordinal, storageTurnIndex: ordinal, turnId: `turn-${ordinal}`, previewTruncated: false,
+    })),
+  } };
+}
 function presentation(startOrdinal = 4, endOrdinalExclusive = 8): SessionHistoryPresentation {
   return {
     range: { startOrdinal, endOrdinalExclusive, targetTurnId: `turn-${startOrdinal}`, mode: 'history-window' },
@@ -151,6 +165,8 @@ describe('FlowChat reader restoration across history rewrites', () => {
     fixture.session = session();
     fixture.cached = presentation();
     fixture.restore.mockResolvedValue(null);
+    fixture.extend.mockReturnValue(null);
+    fixture.synchronize.mockResolvedValue(undefined);
     fixture.listProps = null;
     fixture.viewportSettled = true;
     fixture.present = null;
@@ -223,6 +239,109 @@ describe('FlowChat reader restoration across history rewrites', () => {
     expect(host.querySelector('[data-turn-id="turn-2"]')).not.toBeNull();
   });
 
+  it.each([['local', 8], ['local', 9], ['ssh', 8], ['ssh', 9], ['peer-device', 8], ['peer-device', 9]] as const)(
+    'reacquires the reader window when the %s catalog changes after a turn (total: %d)', async (location, count) => {
+      if (location === 'peer-device') activateSurface('peer-device');
+      const withCatalog = (total: number, revision: string): Session => ({
+        ...sessionWithCatalog(total, revision),
+        ...(location === 'ssh' ? { remoteConnectionId: 'ssh-target', remoteSshHost: 'build-host' } : {}),
+      });
+      fixture.session = withCatalog(8, 'before-turn');
+      const window = presentation(3, 7);
+      fixture.cached = window;
+      stageBookmark(window);
+      await render();
+      expectPresented();
+      // Authoritative completion changes the catalog and invalidates the
+      // store's cached windows, without changing Session id or ready status.
+      fixture.cached = null;
+      const first = deferred<SessionHistoryPresentation | null>();
+      fixture.restore.mockReturnValueOnce(first.promise);
+      fixture.session = withCatalog(count, 'after-turn');
+      await render();
+      expect(fixture.restore).toHaveBeenCalledWith('session-a', window.range, expect.any(Function));
+      const firstOwner = fixture.restore.mock.calls[0][2] as () => boolean;
+      const second = deferred<SessionHistoryPresentation | null>();
+      fixture.restore.mockReturnValueOnce(second.promise);
+      fixture.session = withCatalog(count, 'corrected-turn');
+      await render();
+      expect(fixture.restore).toHaveBeenCalledTimes(2);
+      expect(firstOwner()).toBe(false);
+      await act(async () => first.reject(new Error('Retired history read failed')));
+      await act(async () => second.resolve(window));
+      expectPresented();
+      expect(messages().dataset.presentationMode).toBe('history-window');
+      expect(host.querySelector('[data-turn-id="turn-3"]')).not.toBeNull();
+    },
+  );
+
+  it('drops a retained live-tail history projection when its catalog changes', async () => {
+    fixture.session = sessionWithCatalog(8, 'before-turn');
+    fixture.cached = presentation(0, 8);
+    stageBookmark(fixture.cached);
+    await render();
+    await act(async () => { fixture.listProps!.onRequestJumpToLatest!(); });
+    expect(host.querySelector('[data-turn-id="turn-0"]')).not.toBeNull();
+    fixture.cached = null;
+    fixture.session = sessionWithCatalog(8, 'after-turn');
+    await render();
+    expectPresented();
+    expect(messages().dataset.presentationMode).toBe('tail');
+    expect(host.querySelector('[data-turn-id="turn-0"]')).toBeNull();
+    expect(fixture.restore).not.toHaveBeenCalled();
+  });
+
+  it.each([false, true])('retries an orphaned reader after a failed read (explicit sync: %s)', async synchronizeHistory => {
+    const window = presentation(3, 7);
+    fixture.cached = window;
+    stageBookmark(window);
+    await render();
+    fixture.cached = null;
+    fixture.restore.mockImplementation(async () => {
+      fixture.cached = window;
+      return window;
+    });
+    fixture.restore.mockRejectedValueOnce(new Error('History read timed out'));
+    fixture.loadTurnWindow.mockResolvedValue({ status: 'ready', isCurrent: true });
+    fixture.extend.mockImplementation(() => fixture.cached ? presentation(0, 7) : null);
+    let failed: unknown;
+    await act(async () => {
+      failed = await fixture.listProps!.onHistoryWindowBoundaryIntent!('before', { synchronizeHistory });
+    });
+    expect(failed).toBe('not-ready');
+    expect(fixture.loadTurnWindow).not.toHaveBeenCalled();
+    expect(host.querySelector('[data-turn-id="turn-3"]')).not.toBeNull();
+    let outcome: unknown;
+    await act(async () => {
+      outcome = await fixture.listProps!.onHistoryWindowBoundaryIntent!('before', { synchronizeHistory });
+    });
+    expect(outcome).toBe('applied');
+    expect(fixture.restore).toHaveBeenCalledWith('session-a', window.range, expect.any(Function));
+    expect(fixture.restore).toHaveBeenCalledTimes(2);
+    expect(fixture.synchronize).toHaveBeenCalledTimes(synchronizeHistory ? 2 : 0);
+    expect(host.querySelector('[data-turn-id="turn-0"]')).not.toBeNull();
+  });
+
+  it('rejects a late orphan recovery after changing devices', async () => {
+    const window = presentation(3, 7);
+    fixture.cached = window;
+    stageBookmark(window);
+    await render();
+    fixture.cached = null;
+    const read = deferred<SessionHistoryPresentation | null>();
+    fixture.restore.mockReturnValueOnce(read.promise);
+    let outcome!: Promise<unknown>;
+    await act(async () => {
+      outcome = fixture.listProps!.onHistoryWindowBoundaryIntent!('before', { synchronizeHistory: true });
+    });
+    expect(fixture.restore).toHaveBeenCalledTimes(1);
+    activateSurface('peer-device');
+    await act(async () => { read.resolve(window); await outcome; });
+    expect(await outcome).toBe('cancelled');
+    expect(fixture.loadTurnWindow).not.toHaveBeenCalled();
+    expect(fixture.extend).not.toHaveBeenCalled();
+  });
+
   it('cancels a pending cold range when a ready history rewrite removes it', async () => {
     fixture.cached = null;
     stageBookmark();

src/web-ui/src/flow_chat/components/modern/ModernFlowChatContainer.tsx

@@ -777,7 +777,8 @@ export const ModernFlowChatContainer: React.FC<ModernFlowChatContainerProps> = (
   useLayoutEffect(() => {
     const sessionId = activeSession?.sessionId;
     const previousSessionId = activeSessionIdRef.current;
-    if (previousSessionId && previousSessionId !== sessionId) {
+    if (previousSessionId && (previousSessionId !== sessionId
+      || historyPresentationRef.current?.sessionId === previousSessionId)) {
       rememberSessionViewportState(previousSessionId, {
         historyPresentation: historyPresentationRef.current?.sessionId === previousSessionId
           ? historyPresentationRef.current
@@ -851,6 +852,28 @@ export const ModernFlowChatContainer: React.FC<ModernFlowChatContainerProps> = (
     }
   }, [activeSession?.sessionId, activeSession?.historyState, isRememberedHistoryRangeInvalid, invalidateHistoryPresentationOwner, rememberSessionViewportState, updateViewportIntent, restoreRevision, historyRestoreRetry, surfaceScope]);
 
+  const readerCatalogRef = useRef<{ sessionId: string | undefined; revision: string | undefined } | null>(null);
+  useEffect(() => {
+    const sessionId = activeSession?.sessionId;
+    const revision = activeSession?.turnCatalog?.revision;
+    const previous = readerCatalogRef.current;
+    readerCatalogRef.current = { sessionId, revision };
+    if (!sessionId || previous?.sessionId !== sessionId || previous.revision === revision) return;
+    const intent = viewportIntentRef.current;
+    if (intent?.kind !== 'turn' || intent.sessionId !== sessionId || intent.source !== 'history-range') {
+      if (historyPresentationRef.current?.sessionId === sessionId) {
+        switchToLiveTailForSession(sessionId, { discardRecentHistory: true });
+      }
+      return;
+    }
+    // The store invalidates cached ranges when the authoritative catalog
+    // changes. Retire the view's matching owner too, then reload its bookmark
+    // through the normal history path. Keeping only the rendered copy leaves
+    // every later extend/synchronize pointed at a window the store no longer owns.
+    invalidateHistoryPresentationOwner();
+    setHistoryRestoreRetry(retry => retry + 1);
+  }, [activeSession?.sessionId, activeSession?.turnCatalog?.revision, invalidateHistoryPresentationOwner, switchToLiveTailForSession]);
+
   useEffect(() => {
     const retainedSessionId = continuousProjectionSessionId;
     if (!retainedSessionId) {
@@ -1991,6 +2014,27 @@ export const ModernFlowChatContainer: React.FC<ModernFlowChatContainerProps> = (
           return 'not-ready';
         }
       }
+      // A catalog refresh or cache eviction can retire the store's window
+      // while the reader still displays its copy. Reacquire that exact range
+      // before extending it; repeatedly fetching the adjacent page cannot
+      // restore a missing active window, even after an explicit source sync.
+      const activeRange = flowChatStore.getSessionHistoryViewState(sessionId)?.activeRange;
+      if (presentation && (!activeRange
+        || activeRange.startOrdinal !== presentation.range.startOrdinal
+        || activeRange.endOrdinalExclusive !== presentation.range.endOrdinalExclusive)) {
+        setHistoryBoundaryState(previous => ({ ...previous, [direction]: 'loading' }));
+        try {
+          const restored = await flowChatStore.restoreSessionHistoryWindow(sessionId, presentation.range, isCurrent);
+          if (!isCurrent()) return 'cancelled';
+          if (!restored) throw new Error('The reader history range is no longer available');
+        } catch (error) {
+          if (!isCurrent() || isSurfaceChangedError(error)) return 'cancelled';
+          recordHistoryPagingEvent(sessionId, 'outcome_not_ready', { direction, reason: 'reader-range-restore-failed' });
+          log.warn('Failed to reacquire reader history range', { sessionId, direction, error });
+          setHistoryBoundaryState(previous => ({ ...previous, [direction]: 'error' }));
+          return 'not-ready';
+        }
+      }
       const session = flowChatStore.getState().sessions.get(sessionId);
       if (scope.surfaceId !== 'local' && !presentation) {
         if (direction === 'after' || session?.isPartial === false) {
@@ -2197,6 +2241,10 @@ export const ModernFlowChatContainer: React.FC<ModernFlowChatContainerProps> = (
             direction,
             reason: `load-${result.status}`,
             targetOrdinal,
+            surfaceId: scope.surfaceId,
+            remoteWorkspace: Boolean(session?.remoteConnectionId || session?.remoteSshHost),
+            requestedCatalogRevision: session?.turnCatalog?.revision ?? null,
+            currentCatalogRevision: flowChatStore.getSessionHistoryViewState(sessionId)?.catalog?.revision ?? null,
           });
           setHistoryBoundaryState(previous => ({ ...previous, [direction]: 'error' }));
           return 'not-ready';
@@ -2247,13 +2295,21 @@ export const ModernFlowChatContainer: React.FC<ModernFlowChatContainerProps> = (
           ? flowChatStore.extendSessionHistoryWindow(sessionId, direction)
           : flowChatStore.activateSessionHistoryWindowFromTail(sessionId, targetOrdinal);
         if (!nextPresentation) {
+          const currentHistoryView = flowChatStore.getSessionHistoryViewState(sessionId);
           if (viewportPreparationStarted) {
             options?.cancelViewportPresentationCommit?.();
           }
           recordHistoryPagingEvent(sessionId, 'outcome_not_ready', {
             direction,
             reason: 'no-next-presentation',
             targetOrdinal,
+            surfaceId: scope.surfaceId,
+            remoteWorkspace: Boolean(session?.remoteConnectionId || session?.remoteSshHost),
+            currentCatalogRevision: currentHistoryView?.catalog?.revision ?? null,
+            renderedStartOrdinal: renderedRange.startOrdinal,
+            renderedEndOrdinalExclusive: renderedRange.endOrdinalExclusive,
+            activeStartOrdinal: currentHistoryView?.activeRange?.startOrdinal ?? null,
+            activeEndOrdinalExclusive: currentHistoryView?.activeRange?.endOrdinalExclusive ?? null,
           });
           setHistoryBoundaryState(previous => ({ ...previous, [direction]: 'error' }));
           return 'not-ready';

src/web-ui/src/flow_chat/components/modern/UserMessageItem.test.tsx

@@ -150,6 +150,7 @@ const flowChatStoreMock = vi.hoisted(() => ({
     activeSessionId: null,
   })),
   loadSessionHistory: vi.fn(async () => undefined),
+  subscribeSelector: vi.fn(() => () => {}),
 }));
 
 vi.mock('../../store/FlowChatStore', () => ({

src/web-ui/src/flow_chat/services/flow-chat-manager/EventHandlerModule.test.ts

@@ -123,6 +123,37 @@ describe('external BTW turn projection', () => {
   });
 });
 
+describe('late dialog start events', () => {
+  beforeEach(() => { resetFlowChatStore(); stateMachineManager.clear(); });
+  afterEach(() => { resetFlowChatStore(); stateMachineManager.clear(); });
+
+  it.each(['completed', 'cancelled', 'error'] as const)(
+    'does not restart a turn already projected as %s', status => {
+      createSessionWithTurn({ ...createFinishingTurn(), status });
+      stateMachineManager.getOrCreate('session-1');
+      __test_only__.handleDialogTurnStarted(createFlowChatContext(), {
+        sessionId: 'session-1', turnId: 'turn-1', turnIndex: 0, userInput: 'Delayed start',
+      });
+      expect(stateMachineManager.getCurrentState('session-1')).toBe(SessionExecutionState.IDLE);
+      expect(FlowChatStore.getInstance().getState().sessions.get('session-1')?.dialogTurns[0].status).toBe(status);
+    },
+  );
+
+  it('does not let an older admitted turn take ownership from the current turn', async () => {
+    createSessionWithTurn({ ...createFinishingTurn(), status: 'processing', backendTurnIndex: 1 });
+    FlowChatStore.getInstance().addDialogTurn('session-1', {
+      ...createFinishingTurn(), id: 'turn-2', status: 'processing', backendTurnIndex: 2,
+    });
+    await stateMachineManager.transition('session-1', SessionExecutionEvent.START, {
+      taskId: 'session-1', dialogTurnId: 'turn-2',
+    });
+    __test_only__.handleDialogTurnStarted(createFlowChatContext(), {
+      sessionId: 'session-1', turnId: 'turn-1', turnIndex: 1, userInput: 'Delayed start',
+    });
+    expect(stateMachineManager.get('session-1')?.getContext().currentDialogTurnId).toBe('turn-2');
+  });
+});
+
 describe('Claw bootstrap cancellation', () => {
   beforeEach(() => {
     resetFlowChatStore();

src/web-ui/src/flow_chat/services/flow-chat-manager/EventHandlerModule.ts

@@ -1896,12 +1896,25 @@ function handleDialogTurnStarted(context: FlowChatContext, event: any): void {
     return;
   }
 
+  // Canonical history and direct control events travel independently. A delayed
+  // start must not resurrect a settled turn or supersede a newer admitted turn.
+  // Optimistic queued turns have no backend index and do not block admission.
+  const store = FlowChatStore.getInstance();
+  const knownTurns = store.getState().sessions.get(sessionId)?.dialogTurns ?? [];
+  const knownTurn = knownTurns.find(turn => turn.id === turnId);
+  const admittedIndex = typeof turnIndex === 'number'
+    ? turnIndex : knownTurn?.backendTurnIndex ?? knownTurn?.storageTurnIndex;
+  if (knownTurn && ['completed', 'cancelled', 'error'].includes(knownTurn.status)
+    || admittedIndex !== undefined && knownTurns.some(turn =>
+      (turn.backendTurnIndex ?? turn.storageTurnIndex ?? -1) > admittedIndex)) {
+    log.debug('Dropped stale DialogTurnStarted', { sessionId, turnId, admittedIndex });
+    return;
+  }
+
   clearPromptSuggestion(sessionId);
   finalizePendingTurnCompletionNow(context, sessionId);
   clearPendingTurnCompletion(context, sessionId, turnId);
 
-  const store = FlowChatStore.getInstance();
-
   // Clean up temp image analysis turn if one exists for this session
   const tempTurnId = pendingImageAnalysisTurns.get(sessionId);
   const hadTempTurn = !!tempTurnId;