refactor(rpc): one dispatcher (#27486)

Fold the streaming dispatcher, the unary invocation helper and two
single-importer helpers into dispatcher.ts so request/response and
streaming-unary share one unary path. No behaviour change.

Co-authored-by: m4air <m4air@Mac.localdomain>

config/scripts/pr-code-change-scope-cross-version-send-path.test.mjs

@@ -10,7 +10,6 @@ describe('cross-version wire routing for the send path', () => {
     'src/shared/structured-agent-session-send-mutation.ts',
     'src/main/runtime/rpc/core.ts',
     'src/main/runtime/rpc/errors.ts',
-    'src/main/runtime/rpc/rpc-streaming-dispatcher.ts',
     'src/main/runtime/rpc/orchestration-contract-fence.ts',
     'src/main/runtime/rpc/orchestration-session-caller.ts',
     'src/main/runtime/rpc/orchestration-legacy-compatibility.ts',

config/scripts/pr-code-change-scope.mjs

@@ -206,7 +206,6 @@ const CROSS_VERSION_WIRE_PREFIXES = [
   // Run on every request the suites dispatch, whatever its method.
   'src/main/runtime/rpc/core.ts',
   'src/main/runtime/rpc/errors.ts',
-  'src/main/runtime/rpc/rpc-streaming-dispatcher.ts',
   'src/main/runtime/rpc/orchestration-contract-fence.ts',
   'src/main/runtime/rpc/orchestration-session-caller.ts',
   'src/main/runtime/rpc/orchestration-legacy-compatibility.ts',

src/main/runtime/rpc/dispatcher-caller-fingerprint.ts

@@ -1,9 +0,0 @@
-import { isDurableMutation } from '../../../shared/orchestration-rpc-contract'
-import type { RpcRequest } from './core'
-
-export function needsLocalCallerFingerprint(request: RpcRequest, params: unknown): boolean {
-  return (
-    request.method.startsWith('orchestration.federation') ||
-    (!!request.orchestrationRequestId && isDurableMutation(request.method, params))
-  )
-}

src/main/runtime/rpc/dispatcher-client-browser-routing.ts

@@ -1,12 +0,0 @@
-import type { OrcaRuntimeService } from '../orca-runtime'
-
-export function routeDispatcherClientHostedBrowserRpc(
-  runtime: OrcaRuntimeService,
-  method: string,
-  params: unknown
-) {
-  const candidate = runtime as OrcaRuntimeService & {
-    routeClientHostedBrowserRpc?: OrcaRuntimeService['routeClientHostedBrowserRpc']
-  }
-  return candidate.routeClientHostedBrowserRpc?.(method, params) ?? { handled: false as const }
-}

src/main/runtime/rpc/dispatcher-unary-method-invocation.ts

@@ -1,91 +0,0 @@
-import type { OrcaRuntimeService } from '../orca-runtime'
-import type { RpcContext, RpcMethod, RpcRequest } from './core'
-import { routeDispatcherClientHostedBrowserRpc } from './dispatcher-client-browser-routing'
-import { needsLocalCallerFingerprint } from './dispatcher-caller-fingerprint'
-import type { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility'
-import type {
-  DurableMutationInvocation,
-  OrchestrationMutationExecutor
-} from './orchestration-mutation-executor'
-import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction'
-
-type DispatcherUnaryMethodInvocation = {
-  runtime: OrcaRuntimeService
-  request: RpcRequest
-  method: RpcMethod
-  params: unknown
-  context: RpcContext
-  orchestrationMutations: OrchestrationMutationExecutor
-  legacyOrchestration: OrchestrationLegacyCompatibility
-}
-
-export async function invokeDispatcherUnaryMethod({
-  runtime,
-  request,
-  method,
-  params,
-  context,
-  orchestrationMutations,
-  legacyOrchestration
-}: DispatcherUnaryMethodInvocation): Promise<unknown> {
-  const clientHostedBrowser = await routeDispatcherClientHostedBrowserRpc(
-    runtime,
-    request.method,
-    params
-  )
-  if (clientHostedBrowser.handled) {
-    recordRuntimeFeatureInteraction(
-      runtime,
-      request.method,
-      clientHostedBrowser.result,
-      undefined,
-      request.params
-    )
-    return clientHostedBrowser.result
-  }
-
-  const compatibility = await legacyOrchestration.tryHandle(request, params, context.signal)
-  if (compatibility.handled) {
-    return compatibility.result
-  }
-  const effectiveParams = compatibility.params ?? params
-  const legacyCoordinator = legacyOrchestration.createCoordinatorInvocation(
-    request,
-    compatibility.legacyCoordinatorAuthority
-  )
-  // Why: a legacy coordinator's fingerprint outranks the transport's, so skip minting a local one.
-  const fallbackCallerFingerprint =
-    context.authenticatedCallerFingerprint ??
-    (!legacyCoordinator && needsLocalCallerFingerprint(request, effectiveParams)
-      ? orchestrationMutations.getLocalAuthenticatedCallerFingerprint()
-      : undefined)
-  const callerFingerprint =
-    legacyCoordinator?.mutationCallerFingerprint ?? fallbackCallerFingerprint
-  const invoke = (mutation?: DurableMutationInvocation) => {
-    const legacyCoordinatorRunId = legacyCoordinator?.revalidate()
-    return method.handler(effectiveParams, {
-      ...context,
-      authenticatedCallerFingerprint: mutation?.identity.callerFingerprint ?? callerFingerprint,
-      recordMutationReceipt: mutation?.recordReceipt,
-      markWorkerDoneMutationEffectFree: mutation?.markWorkerDoneEffectFree,
-      markMutationEffectPossible: mutation?.markEffectPossible,
-      orchestrationMutation: mutation?.identity,
-      replayedMutationReceipt: mutation?.replayedReceipt,
-      legacyCoordinatorRunId,
-      legacyCoordinatorAuthority: legacyCoordinator?.authority,
-      revalidateLegacyCoordinator: legacyCoordinator?.revalidate,
-      orchestrationCompatibilityCallerAuthority:
-        compatibility.orchestrationCompatibilityCallerAuthority,
-      orchestrationCompatibilityEvidence: request.orchestrationCompatibilityEvidence
-    })
-  }
-  const result = await orchestrationMutations.run(
-    request,
-    effectiveParams,
-    invoke,
-    callerFingerprint,
-    context.orchestrationCaller?.orcaSessionId
-  )
-  recordRuntimeFeatureInteraction(runtime, request.method, result, undefined, request.params)
-  return result
-}

src/main/runtime/rpc/dispatcher.ts

@@ -14,8 +14,11 @@ import { successResponse } from './errors'
 import { ALL_RPC_METHODS } from './methods'
 import { emulatorProbe, emulatorProbeError } from '../../emulator/emulator-probe'
 import type { OrcaRuntimeService } from '../orca-runtime'
+import { isDurableMutation } from '../../../shared/orchestration-rpc-contract'
+import type { OrchestrationSessionCaller } from '../orchestration/orchestration-caller-identity'
 import {
   getOrchestrationMutationExecutor,
+  type DurableMutationInvocation,
   type OrchestrationMutationExecutor
 } from './orchestration-mutation-executor'
 import { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility'
@@ -24,8 +27,8 @@ import {
   type RpcDispatchStreamingOptions
 } from './dispatcher-stream-options'
 import { mapDispatcherError } from './dispatcher-error-response'
-import { RpcStreamingDispatcher } from './rpc-streaming-dispatcher'
-import { invokeDispatcherUnaryMethod } from './dispatcher-unary-method-invocation'
+import { createDispatcherStreamingFeatureEmitter } from './dispatcher-streaming-feature-emitter'
+import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction'
 import type { RpcCallerScope } from './rpc-caller-scope'
 import { admitRpcRequest } from './rpc-request-admission-prologue'
 
@@ -36,12 +39,18 @@ export type DispatcherOptions = {
   callerScope?: RpcCallerScope
 }
 
+type AdmittedUnaryCall = {
+  request: RpcRequest
+  method: RpcMethod
+  params: unknown
+  caller: OrchestrationSessionCaller | undefined
+}
+
 export class RpcDispatcher {
   private readonly runtime: OrcaRuntimeService
   private readonly registry: RpcRegistry
   private readonly orchestrationMutations: OrchestrationMutationExecutor
   private readonly legacyOrchestration: OrchestrationLegacyCompatibility
-  private readonly streamingDispatcher: RpcStreamingDispatcher
   private readonly pinnedCallerScope: RpcCallerScope | undefined
 
   constructor({ runtime, methods = ALL_RPC_METHODS, callerScope }: DispatcherOptions) {
@@ -50,27 +59,14 @@ export class RpcDispatcher {
     this.registry = buildRegistry(methods)
     this.orchestrationMutations = getOrchestrationMutationExecutor(runtime)
     this.legacyOrchestration = new OrchestrationLegacyCompatibility(runtime)
-    this.streamingDispatcher = new RpcStreamingDispatcher({
-      runtime,
-      registry: this.registry,
-      orchestrationMutations: this.orchestrationMutations,
-      legacyOrchestration: this.legacyOrchestration,
-      meta: () => this.meta(),
-      pinnedCallerScope: callerScope
-    })
   }
 
   async dispatch(request: RpcRequest, options?: RpcDispatchStreamingOptions): Promise<RpcResponse> {
     const meta = this.meta()
-    const pendingAdmission = admitRpcRequest(
-      {
-        runtime: this.runtime,
-        registry: this.registry,
-        meta,
-        pinnedCallerScope: this.pinnedCallerScope
-      },
+    const pendingAdmission = this.admit(
       request,
       options,
+      meta,
       (method: RpcMethod | RpcStreamingMethod): method is RpcMethod => !isStreamingMethod(method)
     )
     const admission =
@@ -79,27 +75,15 @@ export class RpcDispatcher {
       return admission.rejected
     }
 
-    if (request.method.startsWith('emulator.')) {
+    // Why: only the request/response transport probes emulator calls; streaming-unary never did.
+    const probeEmulator = request.method.startsWith('emulator.')
+    if (probeEmulator) {
       emulatorProbe(`rpc ${request.method}`, request.params)
     }
     try {
-      const result = await invokeDispatcherUnaryMethod({
-        runtime: this.runtime,
-        request: admission.request,
-        method: admission.method,
-        params: admission.params,
-        context: rpcContextFromTransport(
-          this.runtime,
-          admission.request,
-          options,
-          admission.caller
-        ),
-        orchestrationMutations: this.orchestrationMutations,
-        legacyOrchestration: this.legacyOrchestration
-      })
-      return successResponse(request.id, meta, result)
+      return successResponse(request.id, meta, await this.invokeUnary(admission, options))
     } catch (error) {
-      if (request.method.startsWith('emulator.')) {
+      if (probeEmulator) {
         emulatorProbeError(`rpc ${request.method}`, error, {
           params: request.params
         })
@@ -108,12 +92,148 @@ export class RpcDispatcher {
     }
   }
 
+  // Why: streaming dispatch sends multiple responses through the reply callback instead of a Promise.
   async dispatchStreaming(
-    request: RpcRequest,
+    rawRequest: RpcRequest,
     reply: (response: string) => void,
     options?: RpcDispatchStreamingOptions
   ): Promise<void> {
-    return this.streamingDispatcher.dispatch(request, reply, options)
+    const meta = this.meta()
+    const pendingAdmission = this.admit(
+      rawRequest,
+      options,
+      meta,
+      (_method: RpcMethod | RpcStreamingMethod): _method is RpcMethod | RpcStreamingMethod => true
+    )
+    const admission =
+      pendingAdmission instanceof Promise ? await pendingAdmission : pendingAdmission
+    if ('rejected' in admission) {
+      reply(JSON.stringify(admission.rejected))
+      return
+    }
+    const { method, request, params, caller } = admission
+
+    if (!isStreamingMethod(method)) {
+      try {
+        const result = await this.invokeUnary({ method, request, params, caller }, options)
+        reply(JSON.stringify(successResponse(request.id, meta, result)))
+      } catch (error) {
+        reply(JSON.stringify(mapDispatcherError(request, meta, error)))
+      }
+      return
+    }
+
+    const { emit, recordedFeatureInteractions } = createDispatcherStreamingFeatureEmitter(
+      this.runtime,
+      request,
+      meta,
+      reply
+    )
+    try {
+      const result = await method.handler(
+        params,
+        rpcContextFromTransport(this.runtime, request, options, caller),
+        emit
+      )
+      recordRuntimeFeatureInteraction(
+        this.runtime,
+        request.method,
+        result,
+        recordedFeatureInteractions,
+        request.params
+      )
+    } catch (error) {
+      reply(JSON.stringify(mapDispatcherError(request, meta, error)))
+    }
+  }
+
+  private admit<M extends RpcMethod | RpcStreamingMethod>(
+    request: RpcRequest,
+    options: RpcDispatchStreamingOptions | undefined,
+    meta: RpcEnvelopeMeta,
+    isTransportSupported: (method: RpcMethod | RpcStreamingMethod) => method is M
+  ) {
+    return admitRpcRequest(
+      {
+        runtime: this.runtime,
+        registry: this.registry,
+        meta,
+        pinnedCallerScope: this.pinnedCallerScope
+      },
+      request,
+      options,
+      isTransportSupported
+    )
+  }
+
+  private async invokeUnary(
+    { request, method, params, caller }: AdmittedUnaryCall,
+    options: RpcDispatchStreamingOptions | undefined
+  ): Promise<unknown> {
+    const { runtime, orchestrationMutations, legacyOrchestration } = this
+    const context = rpcContextFromTransport(runtime, request, options, caller)
+    // Why: test fakes of the runtime omit the client-hosted browser route.
+    const clientHostedBrowser = await (runtime.routeClientHostedBrowserRpc?.(
+      request.method,
+      params
+    ) ?? { handled: false as const })
+    if (clientHostedBrowser.handled) {
+      recordRuntimeFeatureInteraction(
+        runtime,
+        request.method,
+        clientHostedBrowser.result,
+        undefined,
+        request.params
+      )
+      return clientHostedBrowser.result
+    }
+
+    const compatibility = await legacyOrchestration.tryHandle(request, params, context.signal)
+    if (compatibility.handled) {
+      return compatibility.result
+    }
+    const effectiveParams = compatibility.params ?? params
+    const legacyCoordinator = legacyOrchestration.createCoordinatorInvocation(
+      request,
+      compatibility.legacyCoordinatorAuthority
+    )
+    // Why: a legacy coordinator's fingerprint outranks the transport's, so skip minting a local one.
+    const fallbackCallerFingerprint =
+      context.authenticatedCallerFingerprint ??
+      (!legacyCoordinator &&
+      (request.method.startsWith('orchestration.federation') ||
+        (!!request.orchestrationRequestId && isDurableMutation(request.method, effectiveParams)))
+        ? orchestrationMutations.getLocalAuthenticatedCallerFingerprint()
+        : undefined)
+    const callerFingerprint =
+      legacyCoordinator?.mutationCallerFingerprint ?? fallbackCallerFingerprint
+    const invoke = (mutation?: DurableMutationInvocation) => {
+      const legacyCoordinatorRunId = legacyCoordinator?.revalidate()
+      return method.handler(effectiveParams, {
+        ...context,
+        authenticatedCallerFingerprint: mutation?.identity.callerFingerprint ?? callerFingerprint,
+        recordMutationReceipt: mutation?.recordReceipt,
+        markWorkerDoneMutationEffectFree: mutation?.markWorkerDoneEffectFree,
+        markMutationEffectPossible: mutation?.markEffectPossible,
+        orchestrationMutation: mutation?.identity,
+        replayedMutationReceipt: mutation?.replayedReceipt,
+        legacyCoordinatorRunId,
+        legacyCoordinatorAuthority: legacyCoordinator?.authority,
+        revalidateLegacyCoordinator: legacyCoordinator?.revalidate,
+        orchestrationCompatibilityCallerAuthority:
+          compatibility.orchestrationCompatibilityCallerAuthority,
+        orchestrationCompatibilityEvidence: request.orchestrationCompatibilityEvidence
+      })
+    }
+    const result = await orchestrationMutations.run(
+      request,
+      effectiveParams,
+      invoke,
+      callerFingerprint,
+      context.orchestrationCaller?.orcaSessionId
+    )
+    recordRuntimeFeatureInteraction(runtime, request.method, result, undefined, request.params)
+    return result
   }
 
   private meta(): RpcEnvelopeMeta {

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

@@ -1,107 +0,0 @@
-import {
-  isStreamingMethod,
-  type RpcEnvelopeMeta,
-  type RpcMethod,
-  type RpcRegistry,
-  type RpcRequest,
-  type RpcStreamingMethod
-} from './core'
-
-import { successResponse } from './errors'
-import type { OrcaRuntimeService } from '../orca-runtime'
-import type { OrchestrationMutationExecutor } from './orchestration-mutation-executor'
-import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction'
-import type { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility'
-import {
-  rpcContextFromTransport,
-  type RpcDispatchStreamingOptions
-} from './dispatcher-stream-options'
-import { mapDispatcherError } from './dispatcher-error-response'
-import { createDispatcherStreamingFeatureEmitter } from './dispatcher-streaming-feature-emitter'
-import { invokeDispatcherUnaryMethod } from './dispatcher-unary-method-invocation'
-import type { RpcCallerScope } from './rpc-caller-scope'
-import { admitRpcRequest } from './rpc-request-admission-prologue'
-
-export type RpcStreamingDispatcherDependencies = {
-  runtime: OrcaRuntimeService
-  registry: RpcRegistry
-  orchestrationMutations: OrchestrationMutationExecutor
-  legacyOrchestration: OrchestrationLegacyCompatibility
-  meta: () => RpcEnvelopeMeta
-  pinnedCallerScope?: RpcCallerScope
-}
-
-export class RpcStreamingDispatcher {
-  constructor(private readonly dependencies: RpcStreamingDispatcherDependencies) {}
-
-  // Why: streaming dispatch sends multiple responses through the reply callback instead of a Promise.
-  async dispatch(
-    rawRequest: RpcRequest,
-    reply: (response: string) => void,
-    options?: RpcDispatchStreamingOptions
-  ): Promise<void> {
-    const { runtime, registry, orchestrationMutations, legacyOrchestration, meta } =
-      this.dependencies
-    const envelopeMeta = meta()
-    const pendingAdmission = admitRpcRequest(
-      {
-        runtime,
-        registry,
-        meta: envelopeMeta,
-        pinnedCallerScope: this.dependencies.pinnedCallerScope
-      },
-      rawRequest,
-      options,
-      (_method: RpcMethod | RpcStreamingMethod): _method is RpcMethod | RpcStreamingMethod => true
-    )
-    const admission =
-      pendingAdmission instanceof Promise ? await pendingAdmission : pendingAdmission
-    if ('rejected' in admission) {
-      reply(JSON.stringify(admission.rejected))
-      return
-    }
-    const { method, request, params, caller } = admission
-
-    if (!isStreamingMethod(method)) {
-      try {
-        const result = await invokeDispatcherUnaryMethod({
-          runtime,
-          request,
-          method,
-          params,
-          context: rpcContextFromTransport(runtime, request, options, caller),
-          orchestrationMutations,
-          legacyOrchestration
-        })
-        reply(JSON.stringify(successResponse(request.id, envelopeMeta, result)))
-      } catch (error) {
-        reply(JSON.stringify(mapDispatcherError(request, envelopeMeta, error)))
-      }
-      return
-    }
-
-    const { emit, recordedFeatureInteractions } = createDispatcherStreamingFeatureEmitter(
-      runtime,
-      request,
-      envelopeMeta,
-      reply
-    )
-
-    try {
-      const result = await method.handler(
-        params,
-        rpcContextFromTransport(runtime, request, options, caller),
-        emit
-      )
-      recordRuntimeFeatureInteraction(
-        runtime,
-        request.method,
-        result,
-        recordedFeatureInteractions,
-        request.params
-      )
-    } catch (error) {
-      reply(JSON.stringify(mapDispatcherError(request, envelopeMeta, error)))
-    }
-  }
-}