diff --git a/__tests__/unit/hooks/useTerminalStream.seq.test.tsx b/__tests__/unit/hooks/useTerminalStream.seq.test.tsx new file mode 100644 index 00000000..6499c0c0 --- /dev/null +++ b/__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>() + const statusListeners = new Set() + 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') + }) +}) diff --git a/hooks/useTerminalStream.ts b/hooks/useTerminalStream.ts index 940ce9a6..5e3439ee 100644 --- a/hooks/useTerminalStream.ts +++ b/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) diff --git a/services/ws-client.ts b/services/ws-client.ts index 2049e279..56f4acf7 100644 --- a/services/ws-client.ts +++ b/services/ws-client.ts @@ -13,7 +13,11 @@ 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 } @@ -21,7 +25,16 @@ export type WSMessage = // client can positively identify user-owned output instead of parsing the // `❯ ` 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 }