fix(terminal): drop stale terminal_output chunks after a reconnect race

__tests__/unit/hooks/useTerminalStream.seq.test.tsx

@@ -0,0 +1,128 @@
+import { renderHook, act } from '@testing-library/react-native'
+import { useTerminalStream } from '@/hooks/useTerminalStream'
+import { createWrapper } from '@/test-utils'
+
+// ── Controllable wsManager fake ──────────────────────────────────────────────
+// Same shape as the userMessages test's fake, extended with `seq`.
+type WsMsg = {
+  type: string
+  sessionId?: string
+  data?: string
+  lines?: string[]
+  seq?: number
+}
+type Handler = (msg: WsMsg) => void
+type StatusListener = (serverId: string, s: string) => void
+
+jest.mock('@/services/ws-client', () => {
+  const handlers = new Map<string, Set<Handler>>()
+  const statusListeners = new Set<StatusListener>()
+  const send = jest.fn()
+  const fakeClient = {
+    send,
+    status: () => 'connected',
+    on: (type: string, h: Handler) => {
+      if (!handlers.has(type)) handlers.set(type, new Set())
+      handlers.get(type)!.add(h)
+      return () => handlers.get(type)?.delete(h)
+    },
+  }
+  return {
+    wsManager: {
+      getClient: () => fakeClient,
+      forceReconnect: jest.fn(),
+      status: () => 'connected',
+      onAnyStatusChange: (l: StatusListener) => {
+        statusListeners.add(l)
+        return () => statusListeners.delete(l)
+      },
+    },
+    __wsTest: {
+      send,
+      emit: (msg: WsMsg) => {
+        handlers.get(msg.type)?.forEach((h) => h(msg))
+        handlers.get('*')?.forEach((h) => h(msg))
+      },
+      emitStatus: (sid: string, s: string) => statusListeners.forEach((l) => l(sid, s)),
+      reset: () => {
+        handlers.clear()
+        statusListeners.clear()
+        send.mockClear()
+      },
+    },
+  }
+})
+
+jest.mock('@/services/api-client', () => ({
+  createApiForServer: () => ({ get: jest.fn().mockResolvedValue({ output: '' }) }),
+  NotFoundError: class NotFoundError extends Error {},
+}))
+
+const { __wsTest } = jest.requireMock('@/services/ws-client') as {
+  __wsTest: {
+    send: jest.Mock
+    emit: (msg: WsMsg) => void
+    emitStatus: (serverId: string, s: string) => void
+    reset: () => void
+  }
+}
+
+async function renderStream() {
+  return await renderHook(() => useTerminalStream('srv-1', 'sess-1'), { wrapper: createWrapper() })
+}
+
+beforeEach(() => {
+  jest.useFakeTimers()
+  __wsTest.reset()
+})
+
+afterEach(() => {
+  jest.useRealTimers()
+})
+
+describe('useTerminalStream – seq guard', () => {
+  it('accepts terminal_output chunks with increasing seq', async () => {
+    const { result } = await renderStream()
+
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'one\n', seq: 1 }))
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'two\n', seq: 2 }))
+
+    expect(result.current.lines.join('\n')).toContain('one')
+    expect(result.current.lines.join('\n')).toContain('two')
+  })
+
+  it('drops a stale chunk whose seq is not greater than the last accepted seq', async () => {
+    const { result } = await renderStream()
+
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'fresh\n', seq: 5 }))
+    // A late frame from a superseded connection, arriving after a newer seq.
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'STALE\n', seq: 3 }))
+
+    expect(result.current.lines.join('\n')).not.toContain('STALE')
+    expect(result.current.lines.join('\n')).toContain('fresh')
+  })
+
+  it('baselines the seq guard from terminal_replay before accepting further chunks', async () => {
+    const { result } = await renderStream()
+
+    await act(() =>
+      __wsTest.emit({ type: 'terminal_replay', sessionId: 'sess-1', lines: ['replayed'], seq: 10 })
+    )
+    // Stale relative to the replay baseline — must be dropped.
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'STALE\n', seq: 7 }))
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'fresh\n', seq: 11 }))
+
+    expect(result.current.lines.join('\n')).not.toContain('STALE')
+    expect(result.current.lines.join('\n')).toContain('fresh')
+  })
+
+  it('never rejects chunks from a streamer that omits seq (backward compat)', async () => {
+    const { result } = await renderStream()
+
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'one\n' }))
+    await act(() => __wsTest.emit({ type: 'terminal_output', sessionId: 'sess-1', data: 'two\n' }))
+
+    expect(result.current.lines.join('\n')).toContain('one')
+    expect(result.current.lines.join('\n')).toContain('two')
+  })
+})

hooks/useTerminalStream.ts

@@ -56,6 +56,14 @@ export function useTerminalStream(
   const replayReceivedRef = useRef(false)
   // Track whether history has been fed (from replay or HTTP) to avoid double-feeding
   const historyFedRef = useRef(false)
+  // Last terminal_output seq accepted for the current session/connection
+  // generation. A stale frame from a superseded WS connection can otherwise
+  // land after a reconnect reset; any terminal_output whose seq isn't
+  // greater than this is dropped instead of being fed to the VT. Reset to 0
+  // alongside the other per-session refs; baselined from terminal_replay's
+  // seq (undefined when the streamer predates this field, or via the HTTP
+  // fallback, which carries no seq — the guard then never rejects).
+  const lastSeqRef = useRef(0)
 
   // HTTP fallback query — disabled by default, enabled only when WS replay times out
   const [httpFallbackEnabled, setHttpFallbackEnabled] = useState(false)
@@ -88,9 +96,10 @@ export function useTerminalStream(
     meta: { persist: false },
   })
 
-  function feedHistory(raw: string) {
+  function feedHistory(raw: string, baselineSeq = 0) {
     if (historyFedRef.current) return
     historyFedRef.current = true
+    lastSeqRef.current = baselineSeq
     vtRef.current!.reset()
     vtRef.current!.setProvider(provider)
     setLines([])
@@ -124,6 +133,7 @@ export function useTerminalStream(
     vtRef.current!.reset()
     replayReceivedRef.current = false
     historyFedRef.current = false
+    lastSeqRef.current = 0
     queueMicrotask(() => {
       setLines([])
       setParseConfidence('high')
@@ -189,7 +199,7 @@ export function useTerminalStream(
           fallbackTimer = null
         }
         if (msg.userMessages) addUserMessages(msg.userMessages.map((m) => m.text))
-        feedHistory(msg.lines.join('\n'))
+        feedHistory(msg.lines.join('\n'), msg.seq ?? 0)
       })
 
       // Start fallback timer — if no terminal_replay within 2s, fall back to HTTP
@@ -207,6 +217,12 @@ export function useTerminalStream(
       if (!client) return
       unsubOutput = client.on('terminal_output', (msg) => {
         if (msg.type !== 'terminal_output' || msg.sessionId !== sessionId) return
+        // A stale frame from a superseded connection can arrive after a
+        // reconnect reset baselined lastSeqRef higher — drop it rather than
+        // feeding it out of order. Streamers that omit seq (older versions)
+        // never trip this: seq stays undefined and the check is skipped.
+        if (msg.seq != null && msg.seq <= lastSeqRef.current) return
+        if (msg.seq != null) lastSeqRef.current = msg.seq
 
         setIsStreaming(true)
         vtRef.current!.feed(msg.data)

services/ws-client.ts

@@ -13,15 +13,28 @@ import { clientLog } from '@/lib/clientLog'
 
 export type WSMessage =
   | { type: 'session_update'; session: Session }
-  | { type: 'terminal_output'; sessionId: string; data: string }
+  // `seq` is a per-session monotonically increasing chunk counter from the
+  // streamer (starts at 1). Additive; old streamers omit it. Lets the client
+  // detect a stale chunk delivered after a reconnect race instead of
+  // trusting raw WS arrival order.
+  | { type: 'terminal_output'; sessionId: string; data: string; seq?: number }
   | { type: 'session_list'; sessions: Session[] }
   | { type: 'notification'; event: NotificationEvent }
   | { type: 'plan_ready'; sessionId: string; plan: string }
   // Ground-truth user message: the streamer wrote this text to the PTY, so the
   // client can positively identify user-owned output instead of parsing the
   // `❯ <text>` transcript line heuristically. Additive; old streamers omit it.
   | { type: 'user_message'; sessionId: string; text: string; ts: number }
-  | { type: 'terminal_replay'; sessionId: string; lines: string[]; userMessages?: { text: string; ts: number }[] }
+  // `seq` is the streamer's last-emitted terminal_output seq at replay time,
+  // letting the client baseline before trusting subsequent chunks. Additive;
+  // old streamers omit it.
+  | {
+      type: 'terminal_replay'
+      sessionId: string
+      lines: string[]
+      userMessages?: { text: string; ts: number }[]
+      seq?: number
+    }
   | { type: 'session_ready'; session: Session }
   | { type: 'cache_ready' }
   | { type: 'scan_progress'; scanned: number; total: number }