feat(native-chat): add session transport with runtime adapter (U2)

Co-authored-by: Orca <help@stably.ai>

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<void> {
+  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<typeof vi.fn>; 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)
+  })
+})

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<NativeChatApi, 'readSession' | 'subscribe'>
+
+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<NativeChatReadSessionResult>(
+          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
+}

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
   }
 }