diff --git a/src/main/ipc/runtime.ts b/src/main/ipc/runtime.ts index 8a704838815..742743f3119 100644 --- a/src/main/ipc/runtime.ts +++ b/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) + event.sender.send(channel, response) } }, { diff --git a/src/main/runtime/rpc/dispatcher-stream-objects.test.ts b/src/main/runtime/rpc/dispatcher-stream-objects.test.ts new file mode 100644 index 00000000000..23bc7f59642 --- /dev/null +++ b/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"}}' + ]) + }) +}) diff --git a/src/main/runtime/rpc/dispatcher-streaming-feature-emitter.ts b/src/main/runtime/rpc/dispatcher-streaming-feature-emitter.ts index a12b71b42c6..754ff3f3df3 100644 --- a/src/main/runtime/rpc/dispatcher-streaming-feature-emitter.ts +++ b/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() 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 } } diff --git a/src/main/runtime/rpc/dispatcher.ts b/src/main/runtime/rpc/dispatcher.ts index 94d7a153744..f1e615887f2 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -107,6 +107,19 @@ export class RpcDispatcher { rawRequest: RpcRequest, reply: (response: string) => void, options?: RpcDispatchStreamingOptions + ): Promise { + 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 { 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)) } } diff --git a/tests/e2e/ai-vault-account-stream-desktop-test-fixture.ts b/tests/e2e/ai-vault-account-stream-desktop-test-fixture.ts index 416c3781c15..72a78bb26c5 100644 --- a/tests/e2e/ai-vault-account-stream-desktop-test-fixture.ts +++ b/tests/e2e/ai-vault-account-stream-desktop-test-fixture.ts @@ -92,8 +92,8 @@ export function accountStreamDesktopFixture( let holdDispatch = false const pending = new Set>() const slotsById = new Map() - 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) => { + const cloned = structuredClone(response) for (const listener of desktopIpc.listeners.get(channel) ?? []) { - listener({}, response) + listener({}, cloned) } } }