diff --git a/src/renderer/src/components/native-chat/native-chat-session-transport.test.ts b/src/renderer/src/components/native-chat/native-chat-session-transport.test.ts new file mode 100644 index 00000000000..4a2a5e54e66 --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-session-transport.test.ts @@ -0,0 +1,201 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { NativeChatMessage } from '../../../../shared/native-chat-types' +import { + clearRuntimeCompatibilityCacheForTests, + markRuntimeEnvironmentCompatible, + RuntimeRpcCallError +} from '@/runtime/runtime-rpc-client' +import { RUNTIME_COMPAT_BLOCK_CODE } from '@/runtime/runtime-protocol-compat' +import { + getNativeChatSessionTransport, + toRuntimeNativeChatErrorMessage +} from './native-chat-session-transport' + +const nativeChatReadSession = vi.fn() +const nativeChatSubscribe = vi.fn() +const runtimeEnvironmentsCall = vi.fn() +const runtimeEnvironmentsSubscribe = vi.fn() + +const ENV = 'env-1' + +function message(id: string): NativeChatMessage { + return { id, role: 'assistant', blocks: [{ type: 'text', text: id }], timestamp: 1, source: 'transcript' } +} + +function okEnvelope(result: unknown): { ok: true; result: unknown } { + return { ok: true, result } +} + +/** Flush pending microtasks and a macrotask turn — enough for the compat-gate + * await chain inside callRuntimeRpc to reach window.api.runtimeEnvironments.call. */ +function flush(): Promise { + return new Promise((resolve) => setTimeout(resolve)) +} + +beforeEach(() => { + vi.clearAllMocks() + clearRuntimeCompatibilityCacheForTests() + delete (globalThis as { __ORCA_WEB_CLIENT__?: boolean }).__ORCA_WEB_CLIENT__ + vi.stubGlobal('window', { + location: { pathname: '/index.html' }, + api: { + nativeChat: { readSession: nativeChatReadSession, subscribe: nativeChatSubscribe }, + runtimeEnvironments: { + call: runtimeEnvironmentsCall, + subscribe: runtimeEnvironmentsSubscribe + } + } + }) +}) + +describe('getNativeChatSessionTransport — selection', () => { + it('returns the local adapter for a null owner and forwards readSession args', async () => { + nativeChatReadSession.mockResolvedValue({ messages: [] }) + const transport = getNativeChatSessionTransport(null) + + await transport.readSession('claude', 'sess-1', 40, '/t/path') + + expect(nativeChatReadSession).toHaveBeenCalledWith('claude', 'sess-1', 40, '/t/path') + expect(runtimeEnvironmentsCall).not.toHaveBeenCalled() + }) + + it('returns the runtime adapter for an owner on the desktop client', async () => { + markRuntimeEnvironmentCompatible(ENV) + runtimeEnvironmentsCall.mockResolvedValue(okEnvelope({ messages: [] })) + const transport = getNativeChatSessionTransport(ENV) + + await transport.readSession('claude', 'sess-1', 40, '/t/path') + + expect(runtimeEnvironmentsCall).toHaveBeenCalledWith({ + selector: ENV, + method: 'nativeChat.readSession', + params: { agent: 'claude', sessionId: 'sess-1', limit: 40, transcriptPath: '/t/path' }, + timeoutMs: 15_000 + }) + expect(nativeChatReadSession).not.toHaveBeenCalled() + }) + + it('returns the local adapter on the web client even with an owner (R3 guard)', async () => { + ;(window as unknown as { __ORCA_WEB_CLIENT__?: boolean }).__ORCA_WEB_CLIENT__ = true + nativeChatReadSession.mockResolvedValue({ messages: [] }) + const transport = getNativeChatSessionTransport(ENV) + + await transport.readSession('claude', 'sess-1', 40, undefined) + + expect(nativeChatReadSession).toHaveBeenCalledOnce() + expect(runtimeEnvironmentsCall).not.toHaveBeenCalled() + }) +}) + +describe('runtime subscribe', () => { + function stubSubscribe(): { unsubscribe: ReturnType; deliver: (frame: unknown, ok?: boolean) => void } { + const unsubscribe = vi.fn() + let onResponse: (r: { ok: boolean; result?: unknown }) => void = () => {} + runtimeEnvironmentsSubscribe.mockImplementation((_args, callbacks) => { + onResponse = callbacks.onResponse + return Promise.resolve({ unsubscribe, sendBinary: vi.fn() }) + }) + return { + unsubscribe, + deliver: (frame, ok = true) => onResponse(ok ? { ok: true, result: frame } : { ok: false }) + } + } + + it('forwards only appended frames with an array of messages', async () => { + markRuntimeEnvironmentCompatible(ENV) + const { deliver } = stubSubscribe() + const onAppended = vi.fn() + const transport = getNativeChatSessionTransport(ENV) + + transport.subscribe({ subscriptionId: 's-1', agent: 'claude', sessionId: 'sess-1' }, onAppended) + await Promise.resolve() + + deliver({ type: 'appended', messages: [message('m-1')] }) + deliver({ type: 'ready' }) // non-appended, ignored + deliver({ type: 'appended', messages: undefined }) // no array, ignored + deliver({ type: 'appended', messages: [message('m-2')] }, false) // !ok, ignored + + expect(onAppended).toHaveBeenCalledTimes(1) + expect(onAppended).toHaveBeenCalledWith([message('m-1')]) + }) + + it('sync unsubscribe tears down the handle AND reaps the runtime watcher (KTD-6)', async () => { + markRuntimeEnvironmentCompatible(ENV) + const { unsubscribe } = stubSubscribe() + runtimeEnvironmentsCall.mockResolvedValue(okEnvelope({ unsubscribed: true })) + const transport = getNativeChatSessionTransport(ENV) + + const stop = transport.subscribe( + { subscriptionId: 's-1', agent: 'claude', sessionId: 'sess-1' }, + vi.fn() + ) + await flush() + stop() + await flush() + + expect(unsubscribe).toHaveBeenCalledOnce() + expect(runtimeEnvironmentsCall).toHaveBeenCalledWith( + expect.objectContaining({ + method: 'nativeChat.unsubscribe', + params: { subscriptionId: 's-1' } + }) + ) + }) + + it('closes a stream that resolves after unsubscribe (cancel race)', async () => { + markRuntimeEnvironmentCompatible(ENV) + const unsubscribe = vi.fn() + let resolveHandle: (h: unknown) => void = () => {} + runtimeEnvironmentsSubscribe.mockImplementation( + () => new Promise((resolve) => (resolveHandle = resolve)) + ) + runtimeEnvironmentsCall.mockResolvedValue(okEnvelope({ unsubscribed: true })) + const onAppended = vi.fn() + const transport = getNativeChatSessionTransport(ENV) + + const stop = transport.subscribe( + { subscriptionId: 's-1', agent: 'claude', sessionId: 'sess-1' }, + onAppended + ) + stop() // teardown before the subscribe promise resolves + resolveHandle({ unsubscribe, sendBinary: vi.fn() }) + await Promise.resolve() + await Promise.resolve() + + expect(unsubscribe).toHaveBeenCalledOnce() + expect(onAppended).not.toHaveBeenCalled() + }) +}) + +describe('runtime readSession error mapping', () => { + it('maps a method_not_found rejection to the too-old copy (R4)', async () => { + markRuntimeEnvironmentCompatible(ENV) + runtimeEnvironmentsCall.mockResolvedValue({ + ok: false, + error: { code: 'method_not_found', message: 'no such method' } + }) + const transport = getNativeChatSessionTransport(ENV) + + const result = await transport.readSession('claude', 'sess-1', 40, undefined) + + expect(result).toEqual({ error: expect.stringContaining('too old') }) + }) + + it('maps error classes precisely (KTD-4, not catch-all)', () => { + const tooOld = expect.stringContaining('too old') + const generic = "Couldn't read agent chat from the remote runtime." + + expect( + toRuntimeNativeChatErrorMessage( + new RuntimeRpcCallError({ ok: false, error: { code: 'method_not_found', message: 'x' } }) + ) + ).toEqual(tooOld) + + const compatBlock = Object.assign(new Error('runtime too old'), { + code: RUNTIME_COMPAT_BLOCK_CODE + }) + expect(toRuntimeNativeChatErrorMessage(compatBlock)).toEqual(tooOld) + + expect(toRuntimeNativeChatErrorMessage(new Error('request timed out'))).toBe(generic) + }) +}) diff --git a/src/renderer/src/components/native-chat/native-chat-session-transport.ts b/src/renderer/src/components/native-chat/native-chat-session-transport.ts new file mode 100644 index 00000000000..41053b1e4ac --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-session-transport.ts @@ -0,0 +1,135 @@ +import type { + NativeChatApi, + NativeChatAppendedMessages, + NativeChatReadSessionResult +} from '../../../../preload/api-types' +import { isWebClientLocation } from '@/lib/web-client-location' +import { + callRuntimeRpc, + RuntimeRpcCallError, + type RuntimeClientTarget +} from '@/runtime/runtime-rpc-client' +import { isRuntimeCompatBlockError } from '@/runtime/runtime-protocol-compat' + +/** The read/subscribe surface the live-session hook needs, decoupled from where + * the transcript actually lives. Same shape as `window.api.nativeChat`, so the + * hook and everything downstream (merge, assembler, pagination) are unchanged. */ +export type NativeChatSessionTransport = Pick + +const RUNTIME_TOO_OLD = + 'This remote runtime is too old to show agent chat history. Update the remote runtime to view it.' + +/** Map a runtime read failure to the message the read-error state renders. A + * version block (old runtime lacking the method, or the protocol-compat gate) + * gets the explicit "update the remote runtime" copy (R4); anything else — a + * timeout or transport error — gets a generic message, so a transient failure is + * never mislabeled as a version problem (KTD-4, not catch-all). */ +export function toRuntimeNativeChatErrorMessage(err: unknown): string { + if (err instanceof RuntimeRpcCallError && err.code === 'method_not_found') { + return RUNTIME_TOO_OLD + } + if (isRuntimeCompatBlockError(err)) { + return RUNTIME_TOO_OLD + } + return "Couldn't read agent chat from the remote runtime." +} + +/** Delegates straight to the local Electron IPC bridge. On the web client + * `window.api.nativeChat` already bridges to the paired runtime, so web keeps + * using this adapter (R3). Preserves whatever `subscribe` returns (sync fn on + * desktop, promise on the web bridge) — the hook's teardown handles both (R6). */ +const localNativeChatTransport: NativeChatSessionTransport = { + readSession: (agent, sessionId, limit, transcriptPath) => + window.api.nativeChat.readSession(agent, sessionId, limit, transcriptPath), + subscribe: (args, onAppended) => window.api.nativeChat.subscribe(args, onAppended) +} + +function createRuntimeNativeChatTransport(environmentId: string): NativeChatSessionTransport { + const target: RuntimeClientTarget = { kind: 'environment', environmentId } + + return { + readSession: async (agent, sessionId, limit, transcriptPath) => { + try { + return await callRuntimeRpc( + target, + 'nativeChat.readSession', + { agent, sessionId, limit, transcriptPath }, + { timeoutMs: 15_000 } + ) + } catch (err) { + return { error: toRuntimeNativeChatErrorMessage(err) } + } + }, + subscribe: (args, onAppended) => { + const { subscriptionId, agent, sessionId, transcriptPath } = args + let cancelled = false + let handleUnsubscribe: (() => void) | null = null + + void window.api.runtimeEnvironments + .subscribe( + { + selector: environmentId, + method: 'nativeChat.subscribe', + params: { subscriptionId, agent, sessionId, transcriptPath }, + timeoutMs: 15_000 + }, + { + onResponse: (response) => { + if (cancelled || response.ok === false) { + return + } + const frame = response.result as { + type?: string + messages?: NativeChatAppendedMessages + } + if (frame?.type === 'appended' && Array.isArray(frame.messages)) { + onAppended(frame.messages) + } + } + } + ) + .then((handle) => { + handleUnsubscribe = handle.unsubscribe + // The stream resolved after teardown already ran — close it now so the + // late-arriving handle doesn't leak (same race as runtime-file-client). + if (cancelled) { + handle.unsubscribe() + handleUnsubscribe = null + } + }) + .catch(() => { + // A failed subscribe surfaces through the parallel readSession's error + // mapping; the live tail simply never starts, nothing to tear down. + }) + + // Sync unsubscribe does two distinct jobs (KTD-6): (a) tear down the + // renderer dispatcher + runtime stream, and (b) reap the runtime-side + // fs-watcher, keyed by the SAME subscriptionId the subscribe sent (KTD-5). + return () => { + cancelled = true + handleUnsubscribe?.() + handleUnsubscribe = null + void callRuntimeRpc<{ unsubscribed: boolean }>( + target, + 'nativeChat.unsubscribe', + { subscriptionId }, + { timeoutMs: 15_000 } + ).catch(() => { + // Best-effort: the watcher also reaps when the connection closes. + }) + } + } + } +} + +/** Select the read/subscribe transport for a pane. Route to the remote runtime + * only for a `runtime:`-owned pane on a non-web client (KTD-2); web and + * local/`ssh:`-owned panes keep the local adapter. */ +export function getNativeChatSessionTransport( + runtimeEnvironmentId: string | null +): NativeChatSessionTransport { + if (runtimeEnvironmentId && !isWebClientLocation()) { + return createRuntimeNativeChatTransport(runtimeEnvironmentId) + } + return localNativeChatTransport +} diff --git a/src/renderer/src/runtime/runtime-protocol-compat.ts b/src/renderer/src/runtime/runtime-protocol-compat.ts index 3459c01b834..8a1e1ac382a 100644 --- a/src/renderer/src/runtime/runtime-protocol-compat.ts +++ b/src/renderer/src/runtime/runtime-protocol-compat.ts @@ -5,6 +5,17 @@ import { } from '../../../shared/protocol-version' import type { RuntimeStatus } from '../../../shared/runtime-types' +/** Marker on the compat-gate block error. Tagged as a `.code` on a plain Error + * (not a subclass) so the gate keeps throwing an `Error` — its other consumer, + * the runtime-environment switch flow, reads only `.message` and is unaffected. */ +export const RUNTIME_COMPAT_BLOCK_CODE = 'runtime_compat_block' + +/** True when `error` is the protocol-compat block thrown by + * `assertRuntimeStatusCompatible` (vs a transient transport/timeout error). */ +export function isRuntimeCompatBlockError(error: unknown): boolean { + return error instanceof Error && (error as { code?: string }).code === RUNTIME_COMPAT_BLOCK_CODE +} + export function assertRuntimeStatusCompatible(status: RuntimeStatus): void { const verdict = evaluateRuntimeCompat({ clientProtocolVersion: RUNTIME_PROTOCOL_VERSION, @@ -14,6 +25,10 @@ export function assertRuntimeStatusCompatible(status: RuntimeStatus): void { status.minCompatibleRuntimeClientVersion ?? status.minCompatibleMobileVersion }) if (verdict.kind === 'blocked') { - throw new Error(describeRuntimeCompatBlock(verdict)) + // Preserve the descriptive message; add a `.code` marker so callers can + // distinguish a version block from a transport failure. + const error = new Error(describeRuntimeCompatBlock(verdict)) + ;(error as { code?: string }).code = RUNTIME_COMPAT_BLOCK_CODE + throw error } }