diff --git a/mobile/src/session/use-mobile-native-chat-session.test.ts b/mobile/src/session/use-mobile-native-chat-session.test.ts index dfe5f285937..2ff97a3370c 100644 --- a/mobile/src/session/use-mobile-native-chat-session.test.ts +++ b/mobile/src/session/use-mobile-native-chat-session.test.ts @@ -109,7 +109,10 @@ describe('useMobileNativeChatSession', () => { await Promise.resolve() }) - expect(state?.messages).toEqual([]) + // The retained window keeps rendering while the source is gone; what must + // never land is the page that resolved after it disappeared. + expect(state?.messages.map((entry) => entry.id)).not.toContain('stale-page') + expect(state?.messages).toHaveLength(40) expect(state?.status).toBe('idle') expect(state?.loadingEarlier).toBe(false) }) @@ -557,6 +560,30 @@ describe('useMobileNativeChatSession transcriptLoading', () => { }) }) + it('keeps the retained transcript rendered when the stream reports an error', async () => { + // A transient read failure must not blank a conversation the user is looking + // at; the last settled list for this identity stays until a read supersedes it. + let emitFrame: (frame: unknown) => void = () => {} + const client = { + subscribe: vi.fn((_method: string, _params: unknown, onData: (frame: unknown) => void) => { + emitFrame = onData + onData({ type: 'snapshot', messages: [message('a-1')], hasMore: false }) + return () => {} + }) + } as unknown as RpcClient + await mountAt(client, 'session-a') + expect(renders.at(-1)).toMatchObject({ status: 'ready', ids: ['a-1'] }) + + renders.length = 0 + await act(async () => emitFrame({ type: 'error', message: 'stream broke' })) + + expect(renders.at(-1)).toMatchObject({ + status: 'error', + transcriptLoading: false, + ids: ['a-1'] + }) + }) + it('never holds a cached list across a host/workspace source change', async () => { const firstClient = { subscribe: vi.fn((_method: string, _params: unknown, onData: (frame: unknown) => void) => { diff --git a/mobile/src/session/use-mobile-native-chat-session.ts b/mobile/src/session/use-mobile-native-chat-session.ts index 656847079f9..a79b5648dd7 100644 --- a/mobile/src/session/use-mobile-native-chat-session.ts +++ b/mobile/src/session/use-mobile-native-chat-session.ts @@ -273,11 +273,12 @@ export function useMobileNativeChatSession(args: { })() }, [client, agent, sessionId, transcriptPath, hasMore, setList]) + // Held for any unsettled read, not just an in-flight one: a stream error or a + // dropped client would otherwise trade the conversation for an error card. const visibleMessages = transcriptRetentionRef.current.visible({ identity, messages, - settled: settledReady, - loading: status === 'loading' + settled: settledReady }) return { diff --git a/src/renderer/src/components/native-chat/NativeChatView.tsx b/src/renderer/src/components/native-chat/NativeChatView.tsx index 2fb458166ec..4da40f7a1fc 100644 --- a/src/renderer/src/components/native-chat/NativeChatView.tsx +++ b/src/renderer/src/components/native-chat/NativeChatView.tsx @@ -2,7 +2,6 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import { useShallow } from 'zustand/react/shallow' import { useAppStore } from '../../store' import { useNativeChatLaunchDraftSignal } from './use-native-chat-launch-draft-adoption' -import type { NativeChatSession } from '../../../../shared/native-chat-types' import { useNativeChatRetainedSession } from './use-native-chat-retained-session' import { selectNativeChatViewState } from './native-chat-view-state' import { NativeChatMessageList } from './NativeChatMessageList' @@ -47,18 +46,18 @@ import { emptyNativeChatContextMenuActions, useNativeChatContextMenu } from './use-native-chat-context-menu' -import type { NativeChatContextMenuActions } from './use-native-chat-context-menu' import { resolveNativeChatFileLinkContext } from './native-chat-file-link' import { selectNativeChatRuntimeEnvironmentId } from './native-chat-runtime-owner' import { useNativeChatPasteBridge } from './use-native-chat-paste-bridge' import { useNativeChatFileLinkClick } from './use-native-chat-file-link-click' -import type { NativeChatViewProps } from './native-chat-view-types' +import type { NativeChatResolvedViewProps, NativeChatViewProps } from './native-chat-view-types' export type { NativeChatViewProps } from './native-chat-view-types' /** Resolves an agent terminal into its native conversation and composer UI. */ export default function NativeChatView({ terminalTabId, + isVisible, paneKey: preferredPaneKey, targetPtyId = null, launchAgent, @@ -94,6 +93,7 @@ export default function NativeChatView({ agent={resolution.agent} sessionId={resolution.sessionId} transcriptPath={resolution.transcriptPath} + isVisible={isVisible} targetPtyId={targetPtyId} terminalTabId={terminalTabId} onSwitchToTerminal={onSwitchToTerminal} @@ -110,22 +110,13 @@ function NativeChatResolvedView({ agent, sessionId, transcriptPath, + isVisible, targetPtyId, terminalTabId, onSwitchToTerminal, readTerminalScreen, contextMenuActions -}: { - paneKey: string - agent: NativeChatSession['agent'] - sessionId: string | null - transcriptPath: string | null - targetPtyId: string | null - terminalTabId: string - onSwitchToTerminal?: () => void - readTerminalScreen?: () => string | null - contextMenuActions?: Omit -}): React.JSX.Element { +}: NativeChatResolvedViewProps): React.JSX.Element { // Primitive owner selection (no useShallow): routes the pane's read/subscribe to // the remote runtime host for a runtime-owned pane; null keeps the local path. const runtimeEnvironmentId = useAppStore((s) => @@ -136,7 +127,8 @@ function NativeChatResolvedView({ agent, sessionId, transcriptPath, - runtimeEnvironmentId + runtimeEnvironmentId, + enabled: isVisible }) const launchPrompt = useAppStore((s) => s.nativeChatLaunchPromptByTabId[terminalTabId] ?? null) const clearNativeChatLaunchPrompt = useAppStore((s) => s.clearNativeChatLaunchPrompt) diff --git a/src/renderer/src/components/native-chat/native-chat-read-retry-timer.ts b/src/renderer/src/components/native-chat/native-chat-read-retry-timer.ts new file mode 100644 index 00000000000..58f5f742924 --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-read-retry-timer.ts @@ -0,0 +1,31 @@ +const RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000] +const FIXED_RETRY_DELAY_MS = 10_000 + +function retryDelayMs(attempt: number): number { + return RETRY_DELAYS_MS[attempt] ?? FIXED_RETRY_DELAY_MS +} + +export type NativeChatReadRetryTimer = { + schedule: (attempt: number, retry: () => void) => void + cancel: () => void +} + +export function createNativeChatReadRetryTimer(): NativeChatReadRetryTimer { + let timer: ReturnType | null = null + const cancel = (): void => { + if (timer !== null) { + clearTimeout(timer) + timer = null + } + } + return { + schedule(attempt, retry): void { + cancel() + timer = setTimeout(() => { + timer = null + retry() + }, retryDelayMs(attempt)) + }, + cancel + } +} diff --git a/src/renderer/src/components/native-chat/native-chat-stream-teardown.ts b/src/renderer/src/components/native-chat/native-chat-stream-teardown.ts new file mode 100644 index 00000000000..d640c14801d --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-stream-teardown.ts @@ -0,0 +1,24 @@ +import type { NativeChatSessionTransport } from './native-chat-session-transport' + +/** Normalizes desktop's synchronous teardown and the paired web bridge's deferred teardown. */ +export function openNativeChatTranscriptStream( + transport: NativeChatSessionTransport, + args: Parameters[0], + onFrame: Parameters[1] +): () => void { + const teardown = transport.subscribe(args, onFrame) as unknown + if (typeof teardown === 'function') { + return teardown as () => void + } + if (!teardown || typeof (teardown as { then?: unknown }).then !== 'function') { + return () => undefined + } + const deferredTeardown = (teardown as Promise).catch(() => undefined) + return () => { + void deferredTeardown.then((resolvedTeardown) => { + if (typeof resolvedTeardown === 'function') { + ;(resolvedTeardown as () => void)() + } + }) + } +} diff --git a/src/renderer/src/components/native-chat/native-chat-view-types.ts b/src/renderer/src/components/native-chat/native-chat-view-types.ts index 2dce759075d..ca374bab11c 100644 --- a/src/renderer/src/components/native-chat/native-chat-view-types.ts +++ b/src/renderer/src/components/native-chat/native-chat-view-types.ts @@ -1,9 +1,12 @@ import type { TuiAgent } from '../../../../shared/types' +import type { NativeChatSession } from '../../../../shared/native-chat-types' import type { NativeChatContextMenuActions } from './use-native-chat-context-menu' export type NativeChatViewProps = { /** The terminal tab hosting the agent. paneKey is `${tabId}:${leafId}`. */ terminalTabId: string + /** Whether the hosted terminal surface is currently visible. */ + isVisible: boolean /** Specific split leaf this chat surface replaces. */ paneKey?: string /** PTY bound to `paneKey`, used for composer and interactive-card sends. */ @@ -18,3 +21,16 @@ export type NativeChatViewProps = { readTerminalScreen?: () => string | null contextMenuActions?: Omit } + +export type NativeChatResolvedViewProps = { + paneKey: string + agent: NativeChatSession['agent'] + sessionId: string | null + transcriptPath: string | null + isVisible: boolean + targetPtyId: string | null + terminalTabId: string + onSwitchToTerminal?: () => void + readTerminalScreen?: () => string | null + contextMenuActions?: Omit +} diff --git a/src/renderer/src/components/native-chat/use-native-chat-assembled-messages.ts b/src/renderer/src/components/native-chat/use-native-chat-assembled-messages.ts new file mode 100644 index 00000000000..7ec4beab524 --- /dev/null +++ b/src/renderer/src/components/native-chat/use-native-chat-assembled-messages.ts @@ -0,0 +1,86 @@ +import { useLayoutEffect, useMemo, useRef } from 'react' +import type { AgentType, NativeChatMessage } from '../../../../shared/native-chat-types' +import { + applyAppends, + createIncrementalAssembler, + type IncrementalChatAssembler, + reset as resetAssembler +} from './native-chat-incremental-assembler' +import { prepareNativeChatLiveMessages } from './native-chat-live-message-preparation' + +type AssemblyCache = { + assembler: IncrementalChatAssembler + baseSignature: string + baseMessages: readonly NativeChatMessage[] + transcript: readonly NativeChatMessage[] + assembledMessages: NativeChatMessage[] +} + +function cloneAssembler(assembler: IncrementalChatAssembler): IncrementalChatAssembler { + return { + byId: new Map(assembler.byId), + byTurn: new Map(assembler.byTurn), + messages: assembler.messages + } +} + +function sharesPrefix( + whole: readonly NativeChatMessage[], + prefix: readonly NativeChatMessage[], + length: number +): boolean { + for (let index = 0; index < length; index += 1) { + if (whole[index] !== prefix[index]) { + return false + } + } + return true +} + +/** Keeps transcript assembly off the status-only render axis. */ +export function useNativeChatAssembledMessages(args: { + agent: AgentType + sessionId: string | null + baseMessages: readonly NativeChatMessage[] + appended: NativeChatMessage[] +}): { assembledMessages: NativeChatMessage[]; normalizedMessages: NativeChatMessage[] } { + const committedCacheRef = useRef(null) + const { agent, sessionId, baseMessages, appended } = args + + const assembly = useMemo(() => { + const committed = committedCacheRef.current + const transcript = + appended.length > 0 ? [...baseMessages, ...appended] : (baseMessages as NativeChatMessage[]) + const baseSignature = `${agent}\u0000${sessionId ?? ''}` + const baseChanged = + !committed || + baseSignature !== committed.baseSignature || + baseMessages !== committed.baseMessages + const applied = committed?.transcript ?? [] + const isSuffixExtension = + !baseChanged && + transcript.length >= applied.length && + sharesPrefix(transcript, applied, applied.length) + // A discarded render must not mutate the last committed assembler. + const assembler = baseChanged + ? createIncrementalAssembler() + : cloneAssembler(committed.assembler) + + const assembledMessages = isSuffixExtension + ? transcript.length > applied.length + ? applyAppends(assembler, transcript.slice(applied.length)) + : assembler.messages + : resetAssembler(assembler, transcript) + return { assembler, baseSignature, baseMessages, transcript, assembledMessages } + }, [agent, appended, baseMessages, sessionId]) + + useLayoutEffect(() => { + committedCacheRef.current = assembly + }, [assembly]) + + const normalizedMessages = useMemo( + () => prepareNativeChatLiveMessages(assembly.assembledMessages, agent), + [agent, assembly.assembledMessages] + ) + return { assembledMessages: assembly.assembledMessages, normalizedMessages } +} diff --git a/src/renderer/src/components/native-chat/use-native-chat-live-session-visibility.test.ts b/src/renderer/src/components/native-chat/use-native-chat-live-session-visibility.test.ts new file mode 100644 index 00000000000..f9770d87267 --- /dev/null +++ b/src/renderer/src/components/native-chat/use-native-chat-live-session-visibility.test.ts @@ -0,0 +1,393 @@ +// @vitest-environment happy-dom + +import { act, createElement } from 'react' +import { createRoot, type Root } from 'react-dom/client' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { NativeChatMessage } from '../../../../shared/native-chat-types' +import { useAppStore } from '@/store' +import { NATIVE_CHAT_INITIAL_LIMIT, nextNativeChatLimit } from './native-chat-pagination' + +const { resetTransport, subscriptions, transport } = vi.hoisted(() => { + type Subscription = { + onFrame: (frame: unknown) => void + unsubscribe: ReturnType + } + const subscriptions: Subscription[] = [] + const transport = { + readSession: vi.fn(), + subscribe: vi.fn() + } + const resetTransport = (): void => { + subscriptions.splice(0) + transport.readSession.mockReset().mockImplementation(() => new Promise(() => {})) + transport.subscribe.mockReset().mockImplementation((_args, onFrame) => { + const subscription = { onFrame, unsubscribe: vi.fn() } + subscriptions.push(subscription) + return subscription.unsubscribe + }) + } + return { resetTransport, subscriptions, transport } +}) + +vi.mock('./native-chat-session-transport', () => ({ + getNativeChatSessionTransport: () => transport +})) + +import type { + NativeChatLiveSession, + UseNativeChatLiveSessionArgs +} from './use-native-chat-live-session' +import { useNativeChatRetainedSession } from './use-native-chat-retained-session' + +const BASE_ARGS: UseNativeChatLiveSessionArgs = { + paneKey: 'tab-1:leaf-1', + agent: 'claude', + sessionId: 'session-1', + transcriptPath: '/remote/session-1.jsonl', + runtimeEnvironmentId: 'environment-1', + enabled: true +} + +function assistant(id: string, timestamp: number): NativeChatMessage { + return { + id, + role: 'assistant', + blocks: [{ type: 'text', text: id }], + timestamp, + source: 'transcript' + } +} + +function deferred(): { + promise: Promise + resolve: (value: T) => void + reject: (reason: unknown) => void +} { + let resolve = (_value: T): void => {} + let reject = (_reason: unknown): void => {} + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve + reject = promiseReject + }) + return { promise, reject, resolve } +} + +describe('useNativeChatLiveSession visibility', () => { + let root: Root + let latest: NativeChatLiveSession | null = null + const renders = vi.fn() + + function Probe(props: UseNativeChatLiveSessionArgs): null { + renders() + latest = useNativeChatRetainedSession(props) + return null + } + + async function render(props: UseNativeChatLiveSessionArgs): Promise { + await act(async () => { + root.render(createElement(Probe, props)) + await Promise.resolve() + await Promise.resolve() + }) + } + + async function emit(index: number, frame: unknown): Promise { + await act(async () => { + subscriptions[index]?.onFrame(frame) + await Promise.resolve() + }) + } + + beforeEach(() => { + globalThis.IS_REACT_ACT_ENVIRONMENT = true + root = createRoot(document.createElement('div')) + latest = null + renders.mockClear() + resetTransport() + useAppStore.setState({ agentStatusByPaneKey: {} }) + }) + + afterEach(() => { + act(() => root.unmount()) + vi.useRealTimers() + }) + + it('does no transcript IO when initially hidden', async () => { + await render({ ...BASE_ARGS, enabled: false }) + + expect(transport.readSession).not.toHaveBeenCalled() + expect(transport.subscribe).not.toHaveBeenCalled() + }) + + it('unsubscribes on hide, retains committed messages, and rejects hidden work', async () => { + useAppStore.setState({ + agentStatusByPaneKey: { + [BASE_ARGS.paneKey]: { state: 'working', stateStartedAt: 100 } + } + } as never) + await render(BASE_ARGS) + await emit(0, { + type: 'snapshot', + messages: [assistant('committed', 1)], + hasMore: true, + lifecycle: { state: 'working', turnId: 'turn-1', timestamp: 100 } + }) + const staleLoadEarlier = latest?.loadEarlier + + await render({ ...BASE_ARGS, enabled: false }) + + expect(subscriptions[0]?.unsubscribe).toHaveBeenCalledOnce() + expect(latest?.messages.map((message) => message.id)).toEqual(['committed']) + expect(latest?.readPhase).toBe('ready') + expect(latest?.status).toBe('working') + + const readsAfterHide = transport.readSession.mock.calls.length + staleLoadEarlier?.() + latest?.loadEarlier() + expect(transport.readSession).toHaveBeenCalledTimes(readsAfterHide) + + const rendersAfterHide = renders.mock.calls.length + await act(async () => { + for (let index = 0; index < 1_000; index += 1) { + subscriptions[0]?.onFrame({ + type: 'appended', + messages: [assistant(`stale-hidden-${index}`, index + 2)] + }) + } + await Promise.resolve() + }) + expect(renders).toHaveBeenCalledTimes(rendersAfterHide) + expect(latest?.messages.map((message) => message.id)).toEqual(['committed']) + }) + + it('cancels a not-found retry when hidden', async () => { + vi.useFakeTimers() + transport.readSession.mockResolvedValue({ error: 'not found', notFound: true }) + + await render(BASE_ARGS) + expect(transport.readSession).toHaveBeenCalledOnce() + + await render({ ...BASE_ARGS, enabled: false }) + await act(async () => vi.advanceTimersByTimeAsync(60_000)) + + expect(transport.readSession).toHaveBeenCalledOnce() + expect(subscriptions[0]?.unsubscribe).toHaveBeenCalledOnce() + }) + + it('reopens one fresh stream whose snapshot wins every stale generation', async () => { + const oldSeed = deferred<{ messages: NativeChatMessage[] }>() + const freshSeed = deferred<{ messages: NativeChatMessage[] }>() + transport.readSession + .mockImplementationOnce(() => oldSeed.promise) + .mockImplementationOnce(() => freshSeed.promise) + + await render(BASE_ARGS) + await emit(0, { + type: 'snapshot', + messages: [assistant('before-hide', 1)], + hasMore: false + }) + await render({ ...BASE_ARGS, enabled: false }) + await render(BASE_ARGS) + + expect(transport.readSession).toHaveBeenCalledTimes(2) + expect(transport.subscribe).toHaveBeenCalledTimes(2) + expect(subscriptions[0]?.unsubscribe).toHaveBeenCalledOnce() + expect(subscriptions[1]?.unsubscribe).not.toHaveBeenCalled() + expect(latest?.readPhase).toBe('loading') + expect(latest?.messages.map((message) => message.id)).toEqual(['before-hide']) + + await emit(1, { + type: 'snapshot', + messages: [assistant('fresh', 3)], + hasMore: false + }) + await emit(0, { + type: 'replacement', + messages: [assistant('stale-stream', 4)], + hasMore: false + }) + await act(async () => { + oldSeed.resolve({ messages: [assistant('stale-old-seed', 5)] }) + freshSeed.resolve({ messages: [assistant('stale-fresh-seed', 6)] }) + await Promise.resolve() + await Promise.resolve() + }) + + expect(latest?.messages.map((message) => message.id)).toEqual(['fresh']) + }) + + it('closes a deferred paired-client stream after a rapid hide and reveal', async () => { + const deferredTeardown = deferred<() => void>() + const oldUnsubscribe = vi.fn() + transport.subscribe.mockImplementationOnce((_args, onFrame) => { + subscriptions.push({ onFrame, unsubscribe: oldUnsubscribe }) + return deferredTeardown.promise + }) + + await render(BASE_ARGS) + await render({ ...BASE_ARGS, enabled: false }) + await render(BASE_ARGS) + + expect(transport.subscribe).toHaveBeenCalledTimes(2) + expect(subscriptions).toHaveLength(2) + expect(subscriptions[1]?.unsubscribe).not.toHaveBeenCalled() + + await act(async () => { + deferredTeardown.resolve(oldUnsubscribe) + await Promise.resolve() + await Promise.resolve() + }) + + expect(oldUnsubscribe).toHaveBeenCalledOnce() + expect(subscriptions[1]?.unsubscribe).not.toHaveBeenCalled() + }) + + it('contains a deferred paired-client stream-handle rejection', async () => { + const deferredTeardown = deferred<() => void>() + transport.subscribe.mockImplementationOnce((_args, onFrame) => { + subscriptions.push({ onFrame, unsubscribe: vi.fn() }) + return deferredTeardown.promise + }) + + await render(BASE_ARGS) + await act(async () => { + deferredTeardown.reject(new Error('stream setup failed')) + await Promise.resolve() + await Promise.resolve() + }) + await render({ ...BASE_ARGS, enabled: false }) + + expect(transport.subscribe).toHaveBeenCalledOnce() + }) + + it('drops retained messages when the hidden transcript identity changes', async () => { + await render(BASE_ARGS) + await emit(0, { + type: 'snapshot', + messages: [assistant('session-a', 1)], + hasMore: false + }) + await render({ ...BASE_ARGS, enabled: false }) + const hiddenSessionB = { + ...BASE_ARGS, + enabled: false, + sessionId: 'session-2', + transcriptPath: '/remote/session-2.jsonl', + runtimeEnvironmentId: 'environment-2' + } + + await render(hiddenSessionB) + await render(hiddenSessionB) + + expect(transport.readSession).toHaveBeenCalledOnce() + expect(transport.subscribe).toHaveBeenCalledOnce() + expect(latest?.readPhase).toBe('loading') + expect(latest?.messages).toEqual([]) + + await render({ ...hiddenSessionB, enabled: true }) + await emit(1, { + type: 'snapshot', + messages: [assistant('session-b', 2)], + hasMore: false + }) + + expect(transport.readSession).toHaveBeenCalledTimes(2) + expect(transport.subscribe).toHaveBeenCalledTimes(2) + expect(latest?.messages.map((message) => message.id)).toEqual(['session-b']) + }) + + it('reveals with the paged window and restarts it only on a source change', async () => { + const pagedLimit = nextNativeChatLimit(NATIVE_CHAT_INITIAL_LIMIT) + transport.readSession.mockResolvedValue({ messages: [assistant('read', 0)] }) + const initialMessages = Array.from({ length: NATIVE_CHAT_INITIAL_LIMIT }, (_unused, index) => + assistant(`old-${index}`, index) + ) + + await render(BASE_ARGS) + await emit(0, { type: 'snapshot', messages: initialMessages, hasMore: true }) + await act(async () => { + latest?.loadEarlier() + await Promise.resolve() + await Promise.resolve() + }) + expect(transport.readSession.mock.calls[1]?.[2]).toBe(pagedLimit) + + await render({ ...BASE_ARGS, enabled: false }) + await render(BASE_ARGS) + + expect(transport.readSession.mock.calls[2]?.[2]).toBe(pagedLimit) + expect(transport.subscribe.mock.calls[1]?.[0]?.limit).toBe(pagedLimit) + + await render({ ...BASE_ARGS, sessionId: 'session-2', transcriptPath: '/remote/session-2.jsonl' }) + + expect(transport.readSession.mock.calls[3]?.[2]).toBe(NATIVE_CHAT_INITIAL_LIMIT) + }) + + it('lets the pending read repair a stream error frame', async () => { + const seed = deferred<{ messages: NativeChatMessage[] }>() + transport.readSession.mockImplementationOnce(() => seed.promise) + + await render(BASE_ARGS) + await emit(0, { type: 'snapshot', messages: [], hasMore: false, error: 'stream failed' }) + expect(latest?.status).toBe('error') + + await act(async () => { + seed.resolve({ messages: [assistant('seeded', 1)] }) + await Promise.resolve() + await Promise.resolve() + }) + + expect(latest?.readPhase).toBe('ready') + expect(latest?.messages.map((message) => message.id)).toEqual(['seeded']) + }) + + it('keeps the retained transcript when the reveal stream errors', async () => { + await render(BASE_ARGS) + await emit(0, { + type: 'snapshot', + messages: [assistant('committed', 1)], + hasMore: false + }) + await render({ ...BASE_ARGS, enabled: false }) + await render(BASE_ARGS) + await emit(1, { type: 'snapshot', messages: [], hasMore: false, error: 'stream failed' }) + + expect(latest?.messages.map((message) => message.id)).toEqual(['committed']) + expect(latest?.status).not.toBe('error') + }) + + it('fences pagination started before a rapid hide and reveal', async () => { + const initialSeed = deferred<{ messages: NativeChatMessage[] }>() + const oldPage = deferred<{ messages: NativeChatMessage[] }>() + const revealSeed = deferred<{ messages: NativeChatMessage[] }>() + transport.readSession + .mockImplementationOnce(() => initialSeed.promise) + .mockImplementationOnce(() => oldPage.promise) + .mockImplementationOnce(() => revealSeed.promise) + const initialMessages = Array.from({ length: NATIVE_CHAT_INITIAL_LIMIT }, (_unused, index) => + assistant(`old-${index}`, index) + ) + + await render(BASE_ARGS) + await emit(0, { type: 'snapshot', messages: initialMessages, hasMore: true }) + await act(async () => latest?.loadEarlier()) + expect(transport.readSession).toHaveBeenCalledTimes(2) + + await render({ ...BASE_ARGS, enabled: false }) + await render(BASE_ARGS) + await emit(1, { + type: 'snapshot', + messages: [assistant('fresh-after-reveal', 1_000)], + hasMore: false + }) + await act(async () => { + oldPage.resolve({ messages: [assistant('stale-page', -1)] }) + await Promise.resolve() + await Promise.resolve() + }) + + expect(transport.readSession).toHaveBeenCalledTimes(3) + expect(latest?.messages.map((message) => message.id)).toEqual(['fresh-after-reveal']) + }) +}) diff --git a/src/renderer/src/components/native-chat/use-native-chat-live-session.ts b/src/renderer/src/components/native-chat/use-native-chat-live-session.ts index a50f8683426..6044da30673 100644 --- a/src/renderer/src/components/native-chat/use-native-chat-live-session.ts +++ b/src/renderer/src/components/native-chat/use-native-chat-live-session.ts @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react' import { NATIVE_CHAT_SOURCE_PRIORITY, type AgentType, @@ -10,13 +10,7 @@ import { createNativeChatMerger, replaceList } from '../../../../shared/native-chat-merge' -import { - applyAppends, - createIncrementalAssembler, - reset as resetAssembler -} from './native-chat-incremental-assembler' import { mergeNativeChatLiveSession } from './native-chat-live-status' -import { prepareNativeChatLiveMessages } from './native-chat-live-message-preparation' import { hasMoreNativeChatHistory, NATIVE_CHAT_INITIAL_LIMIT, @@ -25,6 +19,9 @@ import { import { getNativeChatSessionTransport } from './native-chat-session-transport' import { useNativeChatTranscriptLifecycle } from './use-native-chat-transcript-lifecycle' import { useNativeChatHookStatus } from './use-native-chat-hook-status' +import { useNativeChatAssembledMessages } from './use-native-chat-assembled-messages' +import { createNativeChatReadRetryTimer } from './native-chat-read-retry-timer' +import { openNativeChatTranscriptStream } from './native-chat-stream-teardown' export type UseNativeChatLiveSessionArgs = { /** Composite `${tabId}:${leafId}` key — selects the live hook entry. */ @@ -36,6 +33,8 @@ export type UseNativeChatLiveSessionArgs = { transcriptPath?: string | null /** Runtime owner (Model B): non-null routes read/subscribe to the remote host; null keeps the local IPC path. */ runtimeEnvironmentId?: string | null + /** False suspends transcript IO while retaining the last committed session. */ + enabled?: boolean } /** A live session plus the older-history pagination controls the view needs. */ @@ -55,20 +54,6 @@ export type NativeChatLiveSession = NativeChatSession & { // Stable empty-base reference so a non-ready read doesn't churn the base axis. const EMPTY_MESSAGES: readonly NativeChatMessage[] = [] -/** True when `whole`'s first `len` entries are referentially identical to `prefix` (a tail-extension), so the assembler can splice just the suffix. */ -function sharesPrefix( - whole: readonly NativeChatMessage[], - prefix: readonly NativeChatMessage[], - len: number -): boolean { - for (let i = 0; i < len; i += 1) { - if (whole[i] !== prefix[i]) { - return false - } - } - return true -} - let subscriptionCounter = 0 function nextSubscriptionId(): string { @@ -76,15 +61,9 @@ function nextSubscriptionId(): string { return `native-chat-${subscriptionCounter}-${Date.now()}` } -// Why: a new session's transcript can take minutes to appear on disk (#8401); a `notFound` miss retries with backoff until the window below elapses. -const NOTFOUND_RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000] -const NOTFOUND_RETRY_FIXED_DELAY_MS = 10_000 +// Why: a new session's transcript can take minutes to appear on disk (#8401). const NOTFOUND_RETRY_WINDOW_MS = 60_000 -function notFoundRetryDelayMs(attempt: number): number { - return NOTFOUND_RETRY_DELAYS_MS[attempt] ?? NOTFOUND_RETRY_FIXED_DELAY_MS -} - export type ReadState = | { phase: 'loading' } | { phase: 'ready'; messages: NativeChatMessage[] } @@ -101,13 +80,13 @@ export type ReadState = * Transport: per-owner (getNativeChatSessionTransport) — a runtime-owned pane * (Model B) reads/tails the remote host; local/ssh panes keep the local IPC path. * - * Teardown: subscription closes on unmount and on owner/agent/sessionId change so - * a swap or owner-flip never leaks a watcher. + * Teardown: subscription closes when hidden, unmounted, or rebound so no source + * generation leaks a watcher. */ export function useNativeChatLiveSession( args: UseNativeChatLiveSessionArgs ): NativeChatLiveSession { - const { paneKey, agent, sessionId, transcriptPath, runtimeEnvironmentId } = args + const { paneKey, agent, sessionId, transcriptPath, runtimeEnvironmentId, enabled = true } = args // Stable per owner id so a re-render without an owner flip keeps the same transport and doesn't re-subscribe. const transport = useMemo( () => getNativeChatSessionTransport(runtimeEnvironmentId ?? null), @@ -127,23 +106,43 @@ export function useNativeChatLiveSession( const [hookState, hookStateStartedAt, hookHasWorkingSubagents] = useNativeChatHookStatus(paneKey) + const latestEnabled = useRef(enabled) + // Fence late frames before passive cleanup closes the previous stream. + useLayoutEffect(() => { + latestEnabled.current = enabled + }, [enabled]) const latestSessionId = useRef(sessionId) latestSessionId.current = sessionId // Tracks the current transport so a load-earlier resolve from a prior host is discarded after an owner flip (session id can stay the same). const latestTransport = useRef(transport) latestTransport.current = transport const transcriptEpochRef = useRef(0) - - // Incremental assembler: suffix-extensions take the fast append path, anything else resets so the cache can't drift from a full rebuild (#17). - const assemblerRef = useRef(createIncrementalAssembler()) - const appliedTranscriptRef = useRef([]) - const baseSigRef = useRef(null) - const baseMessagesRef = useRef(EMPTY_MESSAGES) + const sourceKey = JSON.stringify([ + paneKey, + runtimeEnvironmentId ?? null, + agent, + sessionId, + transcriptPath ?? null + ]) + const retainedSourceKeyRef = useRef(sourceKey) useEffect(() => { // Why: agent/path/owner rebinds can keep the same session; every source generation must invalidate pagination captured before it. transcriptEpochRef.current += 1 setLoadingEarlier(false) + const sourceChanged = retainedSourceKeyRef.current !== sourceKey + retainedSourceKeyRef.current = sourceKey + if (!enabled) { + if (sourceChanged) { + limitRef.current = NATIVE_CHAT_INITIAL_LIMIT + transcriptLifecycleControl.reset() + setRead({ phase: 'loading' }) + replaceList(appendMergerRef.current, []) + setAppended([]) + setHasMore(false) + } + return () => undefined + } transcriptLifecycleControl.reset() if (!sessionId) { // No session id yet: surface live hook state on an empty transcript; backfills once the id arrives. @@ -151,17 +150,20 @@ export function useNativeChatLiveSession( replaceList(appendMergerRef.current, []) setAppended([]) setHasMore(false) - return + return () => undefined } let cancelled = false // Set by the first authoritative frame so the readSession seed below can't clobber a live snapshot. let frameArrived = false - let retryTimer: ReturnType | null = null + const retryTimer = createNativeChatReadRetryTimer() const retryStartedAt = Date.now() // Re-bound as a const: TS drops the `!sessionId` narrowing inside the hoisted nested function. const activeSessionId = sessionId - limitRef.current = NATIVE_CHAT_INITIAL_LIMIT + // Why: a reveal re-reads the same source, so keep the window the user paged in; only a new source starts over. + if (sourceChanged) { + limitRef.current = NATIVE_CHAT_INITIAL_LIMIT + } setRead({ phase: 'loading' }) replaceList(appendMergerRef.current, []) setAppended([]) @@ -169,22 +171,19 @@ export function useNativeChatLiveSession( // Independent initial seed in case subscribe never delivers a snapshot; applied only until an authoritative frame lands so a live snapshot wins. function loadSession(attempt: number): void { - if (frameArrived) { + if (!latestEnabled.current || frameArrived) { return } void transport .readSession(agent, activeSessionId, limitRef.current, transcriptPath ?? undefined) .then((result) => { - if (cancelled || frameArrived) { + if (cancelled || !latestEnabled.current || frameArrived) { return } if (result && 'error' in result) { // A not-yet-flushed transcript: stay in 'loading' and retry with backoff instead of a permanent error (#8401). if (result.notFound && Date.now() - retryStartedAt < NOTFOUND_RETRY_WINDOW_MS) { - retryTimer = setTimeout(() => { - retryTimer = null - loadSession(attempt + 1) - }, notFoundRetryDelayMs(attempt)) + retryTimer.schedule(attempt, () => loadSession(attempt + 1)) return } setRead({ phase: 'error', error: result.error }) @@ -196,7 +195,7 @@ export function useNativeChatLiveSession( setHasMore(hasMoreNativeChatHistory(messages.length, limitRef.current)) }) .catch((err: unknown) => { - if (!cancelled && !frameArrived) { + if (!cancelled && latestEnabled.current && !frameArrived) { setRead({ phase: 'error', error: err instanceof Error ? err.message : String(err) }) } }) @@ -205,7 +204,8 @@ export function useNativeChatLiveSession( loadSession(0) const subscriptionId = nextSubscriptionId() - const unsubscribe = transport.subscribe( + const closeStream = openNativeChatTranscriptStream( + transport, { subscriptionId, agent, @@ -214,53 +214,48 @@ export function useNativeChatLiveSession( limit: limitRef.current }, (frame) => { - if (!cancelled) { - if (frame.type === 'snapshot' || frame.type === 'replacement') { - // Why: snapshots and inode replacements are authoritative generations; older pagination must not repaint them. - frameArrived = true - transcriptEpochRef.current += 1 - setLoadingEarlier(false) - if ('error' in frame && frame.error) { - setRead({ phase: 'error', error: frame.error }) - return - } - transcriptLifecycleControl.replace(frame.lifecycle) - replaceList(appendMergerRef.current, frame.messages) - setAppended([]) - setRead({ phase: 'ready', messages: appendMergerRef.current.list }) - setHasMore(frame.hasMore) + if (cancelled || !latestEnabled.current) { + return + } + if (frame.type === 'snapshot' || frame.type === 'replacement') { + // Why: snapshots and inode replacements are authoritative generations; older pagination must not repaint them. + transcriptEpochRef.current += 1 + setLoadingEarlier(false) + if ('error' in frame && frame.error) { + // Why: an error frame carries no transcript, so it must not consume the seed — a healthy read still has to repair the pane. + setRead({ phase: 'error', error: frame.error }) return } - transcriptLifecycleControl.append(frame.lifecycle) - // Merge by id then bound to the window; the base read + assembler re-dedup mean trimming the append tail can't drop a covered turn (#6). - setAppended(applyAppend(appendMergerRef.current, frame.messages, limitRef.current)) + frameArrived = true + transcriptLifecycleControl.replace(frame.lifecycle) + replaceList(appendMergerRef.current, frame.messages) + setAppended([]) + setRead({ phase: 'ready', messages: appendMergerRef.current.list }) + setHasMore(frame.hasMore) + return } + transcriptLifecycleControl.append(frame.lifecycle) + // Merge by id then bound to the window; the base read + assembler re-dedup mean trimming the append tail can't drop a covered turn (#6). + setAppended(applyAppend(appendMergerRef.current, frame.messages, limitRef.current)) } ) return () => { cancelled = true - if (retryTimer) { - clearTimeout(retryTimer) - retryTimer = null - } - // Web RPC bridge returns a Promise (not the desktop sync unsubscribe fn); calling it as a function crashed the view, so resolve first. - const teardown = unsubscribe as unknown - if (typeof teardown === 'function') { - ;(teardown as () => void)() - } else if (teardown && typeof (teardown as { then?: unknown }).then === 'function') { - void (teardown as Promise).then((fn) => { - if (typeof fn === 'function') { - ;(fn as () => void)() - } - }) - } + retryTimer.cancel() + closeStream() } // `transport` identity changes on an owner flip, re-running this effect to re-subscribe against the new host. - }, [agent, sessionId, transcriptPath, transport, transcriptLifecycleControl]) + }, [agent, enabled, sessionId, sourceKey, transcriptPath, transport, transcriptLifecycleControl]) const loadEarlier = useCallback(() => { - if (!sessionId || loadingEarlier || !hasMore || read.phase !== 'ready') { + if ( + !latestEnabled.current || + !sessionId || + loadingEarlier || + !hasMore || + read.phase !== 'ready' + ) { return } const nextLimit = nextNativeChatLimit(limitRef.current) @@ -272,6 +267,7 @@ export function useNativeChatLiveSession( .then((result) => { // Ignore a stale resolve from a swapped session or flipped owner — either would paint the wrong host's history. if ( + !latestEnabled.current || latestSessionId.current !== sessionId || latestTransport.current !== transport || transcriptEpochRef.current !== requestEpoch @@ -292,7 +288,7 @@ export function useNativeChatLiveSession( }) .finally(() => { // Clear the loading flag on the current epoch even when the result is discarded, so a stale resolve can't wedge it true. - if (transcriptEpochRef.current === requestEpoch) { + if (latestEnabled.current && transcriptEpochRef.current === requestEpoch) { setLoadingEarlier(false) } }) @@ -309,39 +305,12 @@ export function useNativeChatLiveSession( // Computed outside the status memo so hookState churn (status-only) never re-runs the assembler. const baseMessages = read.phase === 'ready' ? read.messages : EMPTY_MESSAGES - const assembledMessages = useMemo(() => { - const transcript = - appended.length > 0 ? [...baseMessages, ...appended] : (baseMessages as NativeChatMessage[]) - // Base-axis signature: any change forces a full assembler reset so a missed trigger can't leave the cache stale. - const baseSig = `${agent}\u0000${sessionId ?? ''}` - const baseChanged = baseSig !== baseSigRef.current || baseMessages !== baseMessagesRef.current - const applied = appliedTranscriptRef.current - const isSuffixExtension = - !baseChanged && - transcript.length >= applied.length && - sharesPrefix(transcript, applied, applied.length) - - let out: NativeChatMessage[] - if (isSuffixExtension && transcript.length > applied.length) { - out = applyAppends(assemblerRef.current, transcript.slice(applied.length)) - } else if (isSuffixExtension) { - out = assemblerRef.current.messages - } else { - out = resetAssembler(assemblerRef.current, transcript) - } - baseSigRef.current = baseSig - baseMessagesRef.current = baseMessages - appliedTranscriptRef.current = transcript - return out - // baseMessages/appended are the only message-set inputs; sessionId/agent gate the reset. hookState intentionally excluded. - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [baseMessages, appended, sessionId, agent]) - - // Keep presentation transforms off the status-only render axis. - const normalizedMessages = useMemo( - () => prepareNativeChatLiveMessages(assembledMessages, agent), - [assembledMessages, agent] - ) + const { assembledMessages, normalizedMessages } = useNativeChatAssembledMessages({ + agent, + sessionId, + baseMessages, + appended + }) return useMemo(() => { const session = mergeNativeChatLiveSession({ diff --git a/src/renderer/src/components/native-chat/use-native-chat-retained-session.test.ts b/src/renderer/src/components/native-chat/use-native-chat-retained-session.test.ts index 17b5801b05a..bee8efe98ed 100644 --- a/src/renderer/src/components/native-chat/use-native-chat-retained-session.test.ts +++ b/src/renderer/src/components/native-chat/use-native-chat-retained-session.test.ts @@ -88,6 +88,28 @@ describe('useNativeChatRetainedSession', () => { expect(latest?.messages).toEqual([]) }) + it('keeps retained messages instead of a full-pane read error', async () => { + liveSession.mockReturnValue(session('ready', [message('settled')])) + await render(ARGS) + + liveSession.mockReturnValue({ ...session('error', []), status: 'error', error: 'read failed' }) + await render(ARGS) + + expect(latest?.messages.map((entry) => entry.id)).toEqual(['settled']) + expect(latest?.readPhase).toBe('error') + expect(latest?.status).toBe('ready') + expect(latest?.error).toBeUndefined() + }) + + it('surfaces a read error when nothing was retained', async () => { + liveSession.mockReturnValue({ ...session('error', []), status: 'error', error: 'read failed' }) + await render(ARGS) + + expect(latest?.messages).toEqual([]) + expect(latest?.status).toBe('error') + expect(latest?.error).toBe('read failed') + }) + it('does not overwrite retention with the session-less view', async () => { liveSession.mockReturnValue(session('ready', [message('settled')])) await render(ARGS) diff --git a/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts b/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts index e10ccecf5aa..6d176f2ac55 100644 --- a/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts +++ b/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts @@ -38,16 +38,17 @@ export function useNativeChatRetainedSession( const messages = retentionRef.current.visible({ identity, messages: session.messages, - settled: readPhase === 'ready', - loading: readPhase === 'loading' + settled: readPhase === 'ready' }) if (messages === session.messages && readPhase === session.readPhase) { return session } - return { - ...session, - messages, - readPhase, - ...(sessionMatchesIdentity ? {} : { status: 'loading' as const, error: undefined }) + if (!sessionMatchesIdentity) { + return { ...session, messages, readPhase, status: 'loading', error: undefined } } + // Retained history beats the full-pane error: a reveal-time read/stream failure is usually transient. + if (session.status === 'error' && messages.length > 0) { + return { ...session, messages, readPhase, status: 'ready', error: undefined } + } + return { ...session, messages, readPhase } } diff --git a/src/renderer/src/components/terminal-pane/TerminalPane.tsx b/src/renderer/src/components/terminal-pane/TerminalPane.tsx index da6070241f9..4498dbe6bef 100644 --- a/src/renderer/src/components/terminal-pane/TerminalPane.tsx +++ b/src/renderer/src/components/terminal-pane/TerminalPane.tsx @@ -3033,6 +3033,7 @@ function TerminalPane(
{ - it('holds only the latest settled transcript for the same identity while loading', () => { + it('holds only the latest settled transcript for the same identity while unsettled', () => { const retention = createNativeChatTranscriptRetention() const first = [message('first')] const second = [message('second')] retention.capture('source-a', first) - expect( - retention.visible({ identity: 'source-a', messages: [], settled: false, loading: true }) - ).toBe(first) - expect( - retention.visible({ identity: 'source-b', messages: [], settled: false, loading: true }) - ).toEqual([]) + expect(retention.visible({ identity: 'source-a', messages: [], settled: false })).toBe(first) + expect(retention.visible({ identity: 'source-b', messages: [], settled: false })).toEqual([]) retention.capture('source-b', second) - expect( - retention.visible({ identity: 'source-a', messages: [], settled: false, loading: true }) - ).toEqual([]) - expect( - retention.visible({ identity: 'source-b', messages: [], settled: false, loading: true }) - ).toBe(second) + expect(retention.visible({ identity: 'source-a', messages: [], settled: false })).toEqual([]) + expect(retention.visible({ identity: 'source-b', messages: [], settled: false })).toBe(second) }) - it('never substitutes retained history for a settled or non-loading read', () => { + it('never substitutes retained history for a settled read', () => { const retention = createNativeChatTranscriptRetention() const retained = [message('retained')] const fresh = [message('fresh')] retention.capture('source', retained) - expect( - retention.visible({ identity: 'source', messages: fresh, settled: true, loading: false }) - ).toBe(fresh) - expect( - retention.visible({ identity: 'source', messages: [], settled: false, loading: false }) - ).toEqual([]) + expect(retention.visible({ identity: 'source', messages: fresh, settled: true })).toBe(fresh) }) it('encodes identity components without delimiter collisions', () => { diff --git a/src/shared/native-chat-transcript-retention.ts b/src/shared/native-chat-transcript-retention.ts index a2f891bb763..860bea95778 100644 --- a/src/shared/native-chat-transcript-retention.ts +++ b/src/shared/native-chat-transcript-retention.ts @@ -12,7 +12,6 @@ export type NativeChatTranscriptRetention = { identity: string messages: NativeChatMessage[] settled: boolean - loading: boolean }) => NativeChatMessage[] } @@ -24,13 +23,12 @@ export function createNativeChatTranscriptRetention(): NativeChatTranscriptReten capture(identity, messages) { captured = { identity, messages } }, - visible({ identity, messages, settled, loading }) { + visible({ identity, messages, settled }) { if (settled) { return messages } - return loading && captured?.identity === identity - ? captured.messages - : EMPTY_NATIVE_CHAT_TRANSCRIPT + // An unsettled read includes a failed one: retained history beats blanking the pane for a transient error. + return captured?.identity === identity ? captured.messages : EMPTY_NATIVE_CHAT_TRANSCRIPT } } }