Avoid JSON round-trips for desktop chat updates (#38936)

* perf(runtime): pass local stream responses directly over IPC

* test(runtime): explain the partial dispatcher runtime double

* test(runtime): observe object streams in the desktop account fixture

src/main/ipc/runtime.ts

@@ -118,7 +118,7 @@ export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void {
       senderSubscriptions.set(args.subscriptionId, controller)
       const channel = `runtime:subscription:${args.subscriptionId}`
       // Streaming outlives dispatch setup; unsubscribe or sender retirement owns cancellation.
-      void new RpcDispatcher({ runtime, methods: ALL_RPC_METHODS }).dispatchStreaming(
+      void new RpcDispatcher({ runtime, methods: ALL_RPC_METHODS }).dispatchStreamingObjects(
         {
           id: args.subscriptionId,
           authToken: 'desktop-ipc',
@@ -127,7 +127,7 @@ export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void {
         },
         (response) => {
           if (!controller.signal.aborted && !event.sender.isDestroyed()) {
-            event.sender.send(channel, JSON.parse(response) as RuntimeRpcResponse<unknown>)
+            event.sender.send(channel, response)
           }
         },
         {

src/main/runtime/rpc/dispatcher-stream-objects.test.ts

@@ -0,0 +1,100 @@
+import './unused-default-rpc-methods.test-fixture'
+import { describe, expect, it } from 'vitest'
+import type { OrcaRuntimeService } from '../orca-runtime'
+import { defineMethod, defineStreamingMethod, type RpcResponse } from './core'
+import { RpcDispatcher } from './dispatcher'
+
+// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: These test methods only read the runtime ID; optional runtime hooks are absent.
+const runtime = { getRuntimeId: () => 'test-runtime' } as OrcaRuntimeService
+
+describe('object and JSON stream transports', () => {
+  it('preserves ordered stream results and errors across structured cloning and JSON', async () => {
+    const dispatcher = new RpcDispatcher({
+      runtime,
+      methods: [
+        defineStreamingMethod({
+          name: 'test.stream',
+          permission: 'workspace',
+          params: null,
+          handler: async (_params, _ctx, emit) => {
+            emit({ type: 'text', text: 'hello\n"😀' })
+            emit({ type: 'end' })
+            throw new Error('boom')
+          }
+        })
+      ]
+    })
+    const request = { id: 'sub-1', authToken: 'token', method: 'test.stream' }
+    const objects: RpcResponse[] = []
+    const json: string[] = []
+    await dispatcher.dispatchStreamingObjects(request, (response) =>
+      objects.push(structuredClone(response))
+    )
+    await dispatcher.dispatchStreaming(request, (response) => json.push(response))
+
+    expect(objects).toEqual([
+      {
+        id: 'sub-1',
+        ok: true,
+        result: { type: 'text', text: 'hello\n"😀' },
+        _meta: { runtimeId: 'test-runtime' },
+        streaming: true
+      },
+      {
+        id: 'sub-1',
+        ok: true,
+        result: { type: 'end' },
+        _meta: { runtimeId: 'test-runtime' },
+        streaming: true
+      },
+      {
+        id: 'sub-1',
+        ok: false,
+        error: { code: 'runtime_error', message: 'boom' },
+        _meta: { runtimeId: 'test-runtime' }
+      }
+    ])
+    expect(json).toEqual([
+      '{"id":"sub-1","ok":true,"result":{"type":"text","text":"hello\\n\\"😀"},"_meta":{"runtimeId":"test-runtime"},"streaming":true}',
+      '{"id":"sub-1","ok":true,"result":{"type":"end"},"_meta":{"runtimeId":"test-runtime"},"streaming":true}',
+      '{"id":"sub-1","ok":false,"error":{"code":"runtime_error","message":"boom"},"_meta":{"runtimeId":"test-runtime"}}'
+    ])
+  })
+
+  it('delivers unary fallback and admission failures over either transport', async () => {
+    const dispatcher = new RpcDispatcher({
+      runtime,
+      methods: [
+        defineMethod({
+          name: 'test.unary',
+          permission: 'workspace',
+          params: null,
+          handler: async () => ({ value: 7 })
+        })
+      ]
+    })
+    const objects: RpcResponse[] = []
+    const json: string[] = []
+    for (const method of ['test.unary', 'test.missing']) {
+      const request = { id: 'req-1', authToken: 'token', method }
+      await dispatcher.dispatchStreamingObjects(request, (response) =>
+        objects.push(structuredClone(response))
+      )
+      await dispatcher.dispatchStreaming(request, (response) => json.push(response))
+    }
+
+    expect(objects).toEqual([
+      { id: 'req-1', ok: true, result: { value: 7 }, _meta: { runtimeId: 'test-runtime' } },
+      {
+        id: 'req-1',
+        ok: false,
+        error: { code: 'method_not_found', message: 'Unknown method: test.missing' },
+        _meta: { runtimeId: 'test-runtime' }
+      }
+    ])
+    expect(json).toEqual([
+      '{"id":"req-1","ok":true,"result":{"value":7},"_meta":{"runtimeId":"test-runtime"}}',
+      '{"id":"req-1","ok":false,"error":{"code":"method_not_found","message":"Unknown method: test.missing"},"_meta":{"runtimeId":"test-runtime"}}'
+    ])
+  })
+})

src/main/runtime/rpc/dispatcher-streaming-feature-emitter.ts

@@ -1,6 +1,6 @@
 import type { FeatureInteractionId } from '../../../shared/feature-interactions'
 import type { OrcaRuntimeService } from '../orca-runtime'
-import type { RpcEnvelopeMeta, RpcRequest } from './core'
+import type { RpcEnvelopeMeta, RpcRequest, RpcResponse } from './core'
 import { successResponse } from './errors'
 import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction'
 
@@ -8,7 +8,7 @@ export function createDispatcherStreamingFeatureEmitter(
   runtime: OrcaRuntimeService,
   request: RpcRequest,
   meta: RpcEnvelopeMeta,
-  reply: (response: string) => void
+  reply: (response: RpcResponse) => void
 ) {
   const recordedFeatureInteractions = new Set<FeatureInteractionId>()
   const emit = (result: unknown): void => {
@@ -21,7 +21,7 @@ export function createDispatcherStreamingFeatureEmitter(
     )
     const response = successResponse(request.id, meta, result)
     response.streaming = true
-    reply(JSON.stringify(response))
+    reply(response)
   }
   return { emit, recordedFeatureInteractions }
 }

src/main/runtime/rpc/dispatcher.ts

@@ -107,6 +107,19 @@ export class RpcDispatcher {
     rawRequest: RpcRequest,
     reply: (response: string) => void,
     options?: RpcDispatchStreamingOptions
+  ): Promise<void> {
+    return this.dispatchStreamingObjects(
+      rawRequest,
+      (response) => reply(JSON.stringify(response)),
+      options
+    )
+  }
+
+  // Transport callbacks must clone or encode synchronously before runtime-owned values change.
+  async dispatchStreamingObjects(
+    rawRequest: RpcRequest,
+    reply: (response: RpcResponse) => void,
+    options?: RpcDispatchStreamingOptions
   ): Promise<void> {
     const meta = this.meta()
     const pendingAdmission = this.admit(
@@ -118,7 +131,7 @@ export class RpcDispatcher {
     const admission =
       pendingAdmission instanceof Promise ? await pendingAdmission : pendingAdmission
     if ('rejected' in admission) {
-      reply(JSON.stringify(admission.rejected))
+      reply(admission.rejected)
       return
     }
     const { method, request, params, caller } = admission
@@ -130,7 +143,7 @@ export class RpcDispatcher {
       try {
         await featureWait
       } catch (error) {
-        reply(JSON.stringify(mapDispatcherError(request, meta, error)))
+        reply(mapDispatcherError(request, meta, error))
         return
       }
     }
@@ -138,9 +151,9 @@ export class RpcDispatcher {
     if (!isStreamingMethod(method)) {
       try {
         const result = await this.invokeUnary({ method, request, params, caller }, options)
-        reply(JSON.stringify(successResponse(request.id, meta, result)))
+        reply(successResponse(request.id, meta, result))
       } catch (error) {
-        reply(JSON.stringify(mapDispatcherError(request, meta, error)))
+        reply(mapDispatcherError(request, meta, error))
       }
       return
     }
@@ -165,7 +178,7 @@ export class RpcDispatcher {
         request.params
       )
     } catch (error) {
-      reply(JSON.stringify(mapDispatcherError(request, meta, error)))
+      reply(mapDispatcherError(request, meta, error))
     }
   }
 

tests/e2e/ai-vault-account-stream-desktop-test-fixture.ts

@@ -92,8 +92,8 @@ export function accountStreamDesktopFixture(
   let holdDispatch = false
   const pending = new Set<Promise<void>>()
   const slotsById = new Map<string, (typeof slots)[number]>()
-  const dispatch = RpcDispatcher.prototype.dispatchStreaming
-  vi.spyOn(RpcDispatcher.prototype, 'dispatchStreaming').mockImplementation(function (
+  const dispatch = RpcDispatcher.prototype.dispatchStreamingObjects
+  vi.spyOn(RpcDispatcher.prototype, 'dispatchStreamingObjects').mockImplementation(function (
     this: RpcDispatcher,
     request,
     reply,
@@ -112,8 +112,8 @@ export function accountStreamDesktopFixture(
       dispatch.call(
         this,
         request,
-        (wire) => {
-          const response = RuntimeRpcEnvelopeSchema.parse(JSON.parse(wire))
+        (frame) => {
+          const response = RuntimeRpcEnvelopeSchema.parse(frame)
           if ('ok' in response && response.ok) {
             frames.push(response.result)
             const frame = response.result
@@ -126,7 +126,7 @@ export function accountStreamDesktopFixture(
               slot.serverId = frame.subscriptionId
             }
           }
-          reply(wire)
+          reply(frame)
         },
         options
       )
@@ -143,8 +143,9 @@ export function accountStreamDesktopFixture(
     once: vi.fn(),
     isDestroyed: () => false,
     send: (channel: string, response: RuntimeRpcResponse<unknown>) => {
+      const cloned = structuredClone(response)
       for (const listener of desktopIpc.listeners.get(channel) ?? []) {
-        listener({}, response)
+        listener({}, cloned)
       }
     }
   }