diff --git a/config/scripts/pr-code-change-scope.mjs b/config/scripts/pr-code-change-scope.mjs index 3a0dd150970..9e386542264 100644 --- a/config/scripts/pr-code-change-scope.mjs +++ b/config/scripts/pr-code-change-scope.mjs @@ -216,7 +216,7 @@ const CROSS_VERSION_WIRE_PREFIXES = [ 'src/shared/orchestration-rpc-contract.ts', 'src/main/runtime/rpc/methods/agent-launch', 'src/main/runtime/rpc/methods/ai-vault.ts', - 'src/main/runtime/rpc/methods/browser-tab-create-schema', + 'src/shared/rpc-contract/browser-tab-create-params', 'src/main/runtime/rpc/methods/session-tabs.ts', 'src/main/runtime/rpc/methods/structured-agent-session', 'src/main/runtime/rpc/methods/terminal', diff --git a/config/scripts/pr-code-change-scope.test.mjs b/config/scripts/pr-code-change-scope.test.mjs index 8463074fd24..71f442b8ce5 100644 --- a/config/scripts/pr-code-change-scope.test.mjs +++ b/config/scripts/pr-code-change-scope.test.mjs @@ -378,7 +378,7 @@ describe('per-job path classification', () => { 'src/main/runtime/agent-session-record-store.ts', 'src/main/runtime/rpc/dispatcher.ts', 'src/main/runtime/rpc/methods/ai-vault.ts', - 'src/main/runtime/rpc/methods/browser-tab-create-schema.ts', + 'src/shared/rpc-contract/browser-tab-create-params.ts', 'src/main/runtime/rpc/methods/session-tabs.ts', 'src/main/runtime/rpc/methods/structured-agent-session.ts', 'src/main/runtime/rpc/methods/structured-agent-session-gate.ts', diff --git a/config/scripts/vitest-node-runtime-files.mjs b/config/scripts/vitest-node-runtime-files.mjs index 1233564b961..a5ea5a72ef3 100644 --- a/config/scripts/vitest-node-runtime-files.mjs +++ b/config/scripts/vitest-node-runtime-files.mjs @@ -73,6 +73,7 @@ export const NODE_RUNTIME_INCLUDE = [ 'src/main/runtime/agent-session-conversation-name-store.test.ts', 'src/main/runtime/structured-session-mail-redrive-wiring.test.ts', 'src/main/runtime/rpc/ws-transport.test.ts', + 'src/main/runtime/rpc/dispatcher-transport-parity.test.ts', // Bun resolves 'ws' to its built-in client, which bypasses the patched package. 'src/main/runtime/rpc/ws-empty-payload-frame-write.test.ts', 'src/shared/remote-runtime-client.test.ts', diff --git a/mobile/src/session/session-read-reply-schema.ts b/mobile/src/session/session-read-reply-schema.ts index 14d163360a0..20f3e25ceac 100644 --- a/mobile/src/session/session-read-reply-schema.ts +++ b/mobile/src/session/session-read-reply-schema.ts @@ -6,7 +6,7 @@ import type { WorktreeDisplayNameSource } from './worktree-display-name' // workspace's connection through, the session tab snapshot, native chat's workspace paths and // older-history page, the quick-command list, the whole `worktree.show` record and a markdown // tab's document. Checked against the handlers in src/main/runtime/rpc/methods/ — repo.ts:29, -// files.ts:27-56, session-tabs.ts:24, client-ui.ts:29-42, mobile-markdown-tab-methods.ts:6-20 — +// files.ts:27-56, session-tabs.ts:24, client-ui.ts:29-42, session-tab-markdown-methods.ts:6-20 — // and the shared result types they return verbatim. /** diff --git a/mobile/src/tasks/mobile-agent-launch-architecture.test.ts b/mobile/src/tasks/mobile-agent-launch-architecture.test.ts index ff595a652bb..eed7f808faf 100644 --- a/mobile/src/tasks/mobile-agent-launch-architecture.test.ts +++ b/mobile/src/tasks/mobile-agent-launch-architecture.test.ts @@ -8,7 +8,7 @@ import { OrcaRuntimeRpcServer } from '../../../src/main/runtime/runtime-rpc' import { DeviceRegistry } from '../../../src/main/runtime/device-registry' import type { AuthenticatedMobileSocket } from '../../../src/main/runtime/rpc/mobile-socket-wiring' import { RpcDispatcher } from '../../../src/main/runtime/rpc/dispatcher' -import { AgentLaunchFields } from '../../../src/main/runtime/rpc/methods/agent-launch-schemas' +import { AgentLaunchFields } from '../../../src/shared/rpc-contract/agent-launch-params' import { runtimeStub, setAgentLaunchRecordStore diff --git a/src/main/runtime/rpc/core.ts b/src/main/runtime/rpc/core.ts index 82df98666bc..88b6b5220a7 100644 --- a/src/main/runtime/rpc/core.ts +++ b/src/main/runtime/rpc/core.ts @@ -125,8 +125,6 @@ export type RpcContext = { ) => () => void } -export type RpcHandler = (params: TParams, ctx: RpcContext) => TResult - // Why: a schema-less method takes no params, so its handler must not be able to read the first argument. type RpcParsedParams = TSchema extends ZodType ? TSchema['_output'] @@ -138,92 +136,60 @@ export type RpcTypedMethod, TResult> + readonly handler: (params: RpcParsedParams, ctx: RpcContext) => TResult } export function defineMethod( spec: RpcTypedMethod ): RpcTypedMethod { - return { - name: spec.name, - permission: spec.permission, - params: spec.params, - handler: spec.handler - } + return spec } -export type RpcStreamingHandler = ( - params: TParams, - ctx: RpcContext, - emit: (result: unknown) => void -) => Promise - // Why: emitted values stay `unknown` — the emit callback is an input, so there is no return position to infer them from. export type RpcTypedStreamingMethod = { readonly name: TName readonly permission: RpcMethodPermission readonly params: TSchema readonly stream: true - readonly handler: RpcStreamingHandler> + readonly handler: ( + params: RpcParsedParams, + ctx: RpcContext, + emit: (result: unknown) => void + ) => Promise } export function defineStreamingMethod( spec: Omit, 'stream'> ): RpcTypedStreamingMethod { - return { - name: spec.name, - permission: spec.permission, - params: spec.params, - stream: true, - handler: spec.handler - } + return { ...spec, stream: true } +} + +type RpcMethodWithParams

= { + readonly name: string + readonly permission: RpcMethodPermission + readonly params: ZodType | null + readonly handler: (params: P, ctx: RpcContext) => unknown +} + +type RpcStreamingMethodWithParams

= { + readonly name: string + readonly permission: RpcMethodPermission + readonly params: ZodType | null + readonly stream: true + readonly handler: (params: P, ctx: RpcContext, emit: (result: unknown) => void) => Promise } // Why `never` params: it makes the declaration a supertype of every parsed-params handler, so typed methods // travel to the registry boundary — and only there get erased — without a cast in each methods module. -export type RpcMethodDeclaration = { - readonly name: string - readonly permission: RpcMethodPermission - readonly params: ZodType | null - readonly handler: (params: never, ctx: RpcContext) => unknown -} - -export type RpcStreamingMethodDeclaration = { - readonly name: string - readonly permission: RpcMethodPermission - readonly params: ZodType | null - readonly stream: true - readonly handler: ( - params: never, - ctx: RpcContext, - emit: (result: unknown) => void - ) => Promise -} - +export type RpcMethodDeclaration = RpcMethodWithParams +type RpcStreamingMethodDeclaration = RpcStreamingMethodWithParams export type RpcAnyMethodDeclaration = RpcMethodDeclaration | RpcStreamingMethodDeclaration // Why: RpcMethod is the registry's erased view; the dispatcher parses params itself and hands handlers `unknown`. -export type RpcMethod = { - readonly name: string - readonly permission: RpcMethodPermission - readonly params: ZodType | null - readonly handler: (params: unknown, ctx: RpcContext) => unknown -} - +export type RpcMethod = RpcMethodWithParams // Why: the `stream` flag lets the dispatcher route these to the emit-based path instead of the one-shot Promise path. -export type RpcStreamingMethod = { - readonly name: string - readonly permission: RpcMethodPermission - readonly params: ZodType | null - readonly stream: true - readonly handler: ( - params: unknown, - ctx: RpcContext, - emit: (result: unknown) => void - ) => Promise -} - -export type RpcAnyMethod = RpcMethod | RpcStreamingMethod +export type RpcStreamingMethod = RpcStreamingMethodWithParams +type RpcAnyMethod = RpcMethod | RpcStreamingMethod // Why the overloads: erasure drops the parsed-params type, not the one-shot/streaming split the dispatcher routes on. export function eraseRpcMethods(methods: readonly RpcMethodDeclaration[]): readonly RpcMethod[] diff --git a/src/main/runtime/rpc/dispatcher-stream-options.ts b/src/main/runtime/rpc/dispatcher-stream-options.ts index af8a87a8edb..6587825ac82 100644 --- a/src/main/runtime/rpc/dispatcher-stream-options.ts +++ b/src/main/runtime/rpc/dispatcher-stream-options.ts @@ -1,29 +1,56 @@ -import type { RuntimeCapability } from '../../../shared/protocol-version' -import type { TerminalStreamFrame } from '../../../shared/terminal-stream-protocol' -import type { PairingRpcContext } from './core' -import type { RpcCallerIdentity } from './rpc-caller-identity' +import type { OrcaRuntimeService } from '../orca-runtime' +import type { OrchestrationSessionCaller } from '../orchestration/orchestration-caller-identity' +import { isRegistrationFencedUnsubscribe, type RpcContext, type RpcRequest } from './core' +import { resolveRpcCallerIdentity } from './rpc-caller-identity' import type { RpcCallerScope } from './rpc-caller-scope' -export type RpcDispatchStreamingOptions = { - authenticatedCallerFingerprint?: string - connectionId?: string - signal?: AbortSignal - clientId?: string - pairedDeviceId?: string - /** Set by a transport that knows its caller but carries no paired device (the desktop's IPC). */ - caller?: RpcCallerIdentity +export type RpcDispatchStreamingOptions = Pick< + RpcContext, + | 'authenticatedCallerFingerprint' + | 'connectionId' + | 'signal' + | 'clientId' + | 'pairedDeviceId' + | 'caller' + | 'clientKind' + | 'clientCapabilities' + | 'updateClientCapabilities' + | 'pairing' + | 'sendBinary' + | 'registerBinaryStreamHandler' + | 'registerBinaryMessageHandler' +> & { /** What the transport proved the caller may do; absent means this host's owner. */ callerScope?: RpcCallerScope - clientKind?: 'mobile' | 'runtime' - clientCapabilities?: readonly RuntimeCapability[] - updateClientCapabilities?: (capabilities: readonly RuntimeCapability[]) => void - pairing?: PairingRpcContext - sendBinary?: (bytes: Uint8Array) => boolean | void - registerBinaryStreamHandler?: ( - streamId: number, - handler: (frame: TerminalStreamFrame) => void - ) => () => void - registerBinaryMessageHandler?: ( - handler: (bytes: Uint8Array) => void - ) => () => void +} + +export function rpcContextFromTransport( + runtime: OrcaRuntimeService, + request: RpcRequest, + options: RpcDispatchStreamingOptions | undefined, + orchestrationCaller: OrchestrationSessionCaller | undefined +): RpcContext { + return { + runtime, + signal: options?.signal, + requestId: request.id, + connectionId: options?.connectionId, + // Session tabs always need this fence. COMPAT(terminal request-addressed unsubscribe): terminal only for phones without `requestId`. + // Capture before middleware yields to a replacement subscribe on the same connection. + subscriptionRegistrationVersion: isRegistrationFencedUnsubscribe(request.method) + ? runtime.getSubscriptionRegistrationVersion() + : undefined, + clientId: options?.clientId, + pairedDeviceId: options?.pairedDeviceId, + caller: resolveRpcCallerIdentity(options), + clientKind: options?.clientKind, + clientCapabilities: options?.clientCapabilities, + updateClientCapabilities: options?.updateClientCapabilities, + authenticatedCallerFingerprint: options?.authenticatedCallerFingerprint, + pairing: options?.pairing, + sendBinary: options?.sendBinary, + registerBinaryStreamHandler: options?.registerBinaryStreamHandler, + registerBinaryMessageHandler: options?.registerBinaryMessageHandler, + orchestrationCaller + } } diff --git a/src/main/runtime/rpc/dispatcher-transport-parity.test.ts b/src/main/runtime/rpc/dispatcher-transport-parity.test.ts new file mode 100644 index 00000000000..371ebb8587d --- /dev/null +++ b/src/main/runtime/rpc/dispatcher-transport-parity.test.ts @@ -0,0 +1,197 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { z } from 'zod' +import { ORCHESTRATION_CONTRACT_VERSION } from '../../../shared/protocol-version' +import { OrcaRuntimeService } from '../orca-runtime' +import { OrchestrationDb } from '../orchestration/db' +import { + defineMethod, + defineStreamingMethod, + type RpcAnyMethodDeclaration, + type RpcRequest, + type RpcResponse +} from './core' +import { RpcDispatcher } from './dispatcher' +import { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility' +import { + attachMutationReplayNudge, + stripMutationReplayNudge +} from './orchestration-mutation-receipt' + +type Transport = 'dispatch' | 'streaming' +const TRANSPORTS: readonly Transport[] = ['dispatch', 'streaming'] +const AnyParams = z.record(z.string(), z.unknown()) +const databases: OrchestrationDb[] = [] + +afterEach(() => { + for (const db of databases.splice(0)) { + db.close() + } + vi.restoreAllMocks() +}) + +function createDispatcher(methods: readonly RpcAnyMethodDeclaration[]): RpcDispatcher { + const db = new OrchestrationDb(':memory:') + databases.push(db) + const runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + vi.spyOn(runtime, 'getTerminalPromptRequestBinding').mockReturnValue({ + ptyId: 'pty-prompt', + processIncarnation: 'incarnation-1', + generation: 1 + }) + vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue('window-1:leaf-prompt') + return new RpcDispatcher({ runtime, methods }) +} + +async function call( + dispatcher: RpcDispatcher, + transport: Transport, + request: RpcRequest, + options?: Parameters[1] +): Promise { + if (transport === 'dispatch') { + return dispatcher.dispatch(request, options) + } + const replies: string[] = [] + await dispatcher.dispatchStreaming(request, (reply) => replies.push(reply), options) + return JSON.parse(replies[0]!) +} + +function mutationRequest(method: string, params: unknown, requestId: string): RpcRequest { + return { + id: `rpc-${requestId}`, + authToken: 'token', + method, + orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, + orchestrationRequestId: requestId, + params + } +} + +describe('RpcDispatcher unary transport parity', () => { + it.each(TRANSPORTS)('keeps an ambiguous prompt receipt over %s', async (transport) => { + const handler = vi.fn((_params: unknown, ctx: { markMutationEffectPossible?: () => void }) => { + ctx.markMutationEffectPossible?.() + throw new Error('terminal_not_writable') + }) + const dispatcher = createDispatcher([ + defineMethod({ name: 'terminal.send', permission: 'workspace', params: AnyParams, handler }) + ]) + const request = mutationRequest( + 'terminal.send', + { + terminal: 'term-prompt', + text: 'go', + enter: true, + agentPrompt: true, + client: { id: 'orca-cli', type: 'desktop' } + }, + 'post-write' + ) + + await expect(call(dispatcher, transport, request)).resolves.toMatchObject({ ok: false }) + await expect(call(dispatcher, transport, request)).resolves.toMatchObject({ + ok: false, + error: { code: 'operation_unknown' } + }) + expect(handler).toHaveBeenCalledOnce() + }) + + it('returns the same de-nudged replay receipt over both transports', async () => { + const replays = await Promise.all( + TRANSPORTS.map(async (transport) => { + const dispatcher = createDispatcher([ + defineMethod({ + name: 'orchestration.send', + permission: 'workspace', + params: AnyParams, + handler: (_params, { replayedMutationReceipt }) => + replayedMutationReceipt + ? stripMutationReplayNudge(replayedMutationReceipt) + : attachMutationReplayNudge( + { messageId: 'msg_1' }, + { kind: 'messages', targets: [{ to: 'term_peer', type: 'status' }] } + ) + }) + ]) + const request = mutationRequest('orchestration.send', { to: 'term_peer' }, 'send-1') + await call(dispatcher, transport, request) + return call(dispatcher, transport, request) + }) + ) + + const results = replays.map((reply) => (reply.ok ? reply.result : reply.error)) + expect(results[0]).toEqual({ + messageId: 'msg_1', + mutation: { requestId: 'send-1', replayed: true } + }) + expect(results[1]).toEqual(results[0]) + }) + + it.each(TRANSPORTS)( + 'passes a legacy coordinator fingerprint over an authenticated %s caller', + async (transport) => { + vi.spyOn(OrchestrationLegacyCompatibility.prototype, 'tryHandle').mockResolvedValue({ + handled: false + }) + vi.spyOn( + OrchestrationLegacyCompatibility.prototype, + 'createCoordinatorInvocation' + ).mockReturnValue({ + authority: { + runId: 'run_1', + principalId: null, + terminalHandle: 'term_coord', + paneKey: 'tab:leaf', + consumerGeneration: 0 + }, + mutationCallerFingerprint: 'legacy-fingerprint', + revalidate: () => 'run_1' + }) + const seen: (string | undefined)[] = [] + const dispatcher = createDispatcher([ + defineMethod({ + name: 'orchestration.taskList', + permission: 'workspace', + params: null, + handler: (_params, { authenticatedCallerFingerprint }) => { + seen.push(authenticatedCallerFingerprint) + return {} + } + }) + ]) + + await call( + dispatcher, + transport, + { id: 'rpc-1', authToken: 'token', method: 'orchestration.taskList' }, + { authenticatedCallerFingerprint: 'transport-fingerprint' } + ) + expect(seen).toEqual(['legacy-fingerprint']) + } + ) + + it('refuses a streaming method over one-shot dispatch before binding an SSH bridge caller', async () => { + const dispatcher = new RpcDispatcher({ + runtime: new OrcaRuntimeService(), + methods: [ + defineStreamingMethod({ + name: 'terminal.read', + permission: 'workspace', + params: z.object({ terminal: z.string() }), + handler: async () => {} + }) + ], + callerScope: { kind: 'ssh-bridge', targetId: 'ssh-1', remoteCliControl: false } + }) + + await expect( + dispatcher.dispatch({ + id: 'rpc-1', + authToken: 'token', + method: 'terminal.read', + params: { terminal: 'term_elsewhere' } + }) + ).resolves.toMatchObject({ ok: false, error: { code: 'method_not_supported' } }) + }) +}) diff --git a/src/main/runtime/rpc/dispatcher-unary-method-invocation.ts b/src/main/runtime/rpc/dispatcher-unary-method-invocation.ts index d8bf3e25b53..7b26823026c 100644 --- a/src/main/runtime/rpc/dispatcher-unary-method-invocation.ts +++ b/src/main/runtime/rpc/dispatcher-unary-method-invocation.ts @@ -53,18 +53,19 @@ export async function invokeDispatcherUnaryMethod({ request, compatibility.legacyCoordinatorAuthority ) - const authenticatedCallerFingerprint = + // Why: a legacy coordinator's fingerprint outranks the transport's, so skip minting a local one. + const fallbackCallerFingerprint = context.authenticatedCallerFingerprint ?? - legacyCoordinator?.mutationCallerFingerprint ?? - (needsLocalCallerFingerprint(request, effectiveParams) + (!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 ?? authenticatedCallerFingerprint, + authenticatedCallerFingerprint: mutation?.identity.callerFingerprint ?? callerFingerprint, recordMutationReceipt: mutation?.recordReceipt, markWorkerDoneMutationEffectFree: mutation?.markWorkerDoneEffectFree, markMutationEffectPossible: mutation?.markEffectPossible, @@ -82,7 +83,7 @@ export async function invokeDispatcherUnaryMethod({ request, effectiveParams, invoke, - legacyCoordinator?.mutationCallerFingerprint ?? authenticatedCallerFingerprint, + callerFingerprint, context.orchestrationCaller?.orcaSessionId ) recordRuntimeFeatureInteraction(runtime, request.method, result, undefined, request.params) diff --git a/src/main/runtime/rpc/dispatcher.ts b/src/main/runtime/rpc/dispatcher.ts index 7920166d119..478c2db953b 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -1,15 +1,16 @@ import { buildRegistry, - isRegistrationFencedUnsubscribe, isStreamingMethod, type RpcAnyMethodDeclaration, type RpcEnvelopeMeta, + type RpcMethod, type RpcRegistry, type RpcRequest, - type RpcResponse + type RpcResponse, + type RpcStreamingMethod } from './core' -import { errorResponse, successResponse } from './errors' +import { successResponse, errorResponse } from './errors' import { ALL_RPC_METHODS } from './methods' import { emulatorProbe, emulatorProbeError } from '../../emulator/emulator-probe' import type { OrcaRuntimeService } from '../orca-runtime' @@ -17,25 +18,16 @@ import { getOrchestrationMutationExecutor, type OrchestrationMutationExecutor } from './orchestration-mutation-executor' -import { orchestrationMigrationFence } from './orchestration-contract-fence' import { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility' -import type { RpcDispatchStreamingOptions } from './dispatcher-stream-options' +import { + rpcContextFromTransport, + type RpcDispatchStreamingOptions +} from './dispatcher-stream-options' import { mapDispatcherError } from './dispatcher-error-response' -import { parseRpcRequestParams } from './dispatcher-request-parsing' import { RpcStreamingDispatcher } from './rpc-streaming-dispatcher' import { invokeDispatcherUnaryMethod } from './dispatcher-unary-method-invocation' -import { resolveRpcCallerIdentity } from './rpc-caller-identity' -import { - bindRpcCallToCallerScope, - denyRpcMethodForCaller, - OWNER_RPC_CALLER_SCOPE, - type RpcCallerScope -} from './rpc-caller-scope' -import { - needsOrchestrationCallerResolution, - resolveOrchestrationSessionCaller, - type ResolvedOrchestrationRequest -} from './orchestration-session-caller' +import type { RpcCallerScope } from './rpc-caller-scope' +import { admitRpcRequest } from './rpc-request-admission-prologue' export type DispatcherOptions = { runtime: OrcaRuntimeService @@ -44,8 +36,6 @@ export type DispatcherOptions = { callerScope?: RpcCallerScope } -type DispatchCallOptions = RpcDispatchStreamingOptions - export class RpcDispatcher { private readonly runtime: OrcaRuntimeService private readonly registry: RpcRegistry @@ -70,59 +60,23 @@ export class RpcDispatcher { }) } - async dispatch(request: RpcRequest, options?: DispatchCallOptions): Promise { + async dispatch(request: RpcRequest, options?: RpcDispatchStreamingOptions): Promise { const meta = this.meta() - const method = this.registry.get(request.method) - const callerScope = this.pinnedCallerScope ?? options?.callerScope ?? OWNER_RPC_CALLER_SCOPE - const denial = denyRpcMethodForCaller(callerScope, request.method, method?.permission) - if (denial) { - return errorResponse(request.id, meta, 'forbidden', denial) - } - if (!method) { - return errorResponse( - request.id, + const pendingAdmission = admitRpcRequest( + { + runtime: this.runtime, + registry: this.registry, meta, - 'method_not_found', - `Unknown method: ${request.method}` - ) - } - - const migrationFence = orchestrationMigrationFence(request, meta) - if (migrationFence) { - return migrationFence - } - - let resolved: ResolvedOrchestrationRequest = { request } - if (needsOrchestrationCallerResolution(request)) { - try { - resolved = await resolveOrchestrationSessionCaller(this.runtime, request, options) - } catch (error) { - return mapDispatcherError(request, meta, error) - } - } - const parsedParams = parseRpcRequestParams(resolved.request, method, meta) - if (parsedParams.error) { - return parsedParams.error - } - - if (isStreamingMethod(method)) { - return errorResponse( - request.id, - meta, - 'method_not_supported', - `Method ${request.method} requires a streaming transport` - ) - } - - const pendingBinding = bindRpcCallToCallerScope( - callerScope, - this.runtime, - request.method, - parsedParams.value + pinnedCallerScope: this.pinnedCallerScope + }, + request, + options, + (method: RpcMethod | RpcStreamingMethod): method is RpcMethod => !isStreamingMethod(method) ) - const binding = pendingBinding ? await pendingBinding : null - if (binding?.kind === 'denied') { - return errorResponse(request.id, meta, 'forbidden', binding.message) + const admission = + pendingAdmission instanceof Promise ? await pendingAdmission : pendingAdmission + if ('rejected' in admission) { + return admission.rejected } if (request.method.startsWith('emulator.')) { @@ -131,37 +85,31 @@ export class RpcDispatcher { try { const result = await invokeDispatcherUnaryMethod({ runtime: this.runtime, - request: resolved.request, - method, - params: parsedParams.value, - context: { - runtime: this.runtime, - signal: options?.signal, - connectionId: options?.connectionId, - // Session tabs always need this fence. COMPAT(terminal request-addressed unsubscribe): terminal only for phones without `requestId`. - subscriptionRegistrationVersion: isRegistrationFencedUnsubscribe(request.method) - ? this.runtime.getSubscriptionRegistrationVersion() - : undefined, - requestId: request.id, - clientId: options?.clientId, - caller: resolveRpcCallerIdentity(options), - clientKind: options?.clientKind, - clientCapabilities: options?.clientCapabilities, - updateClientCapabilities: options?.updateClientCapabilities, - authenticatedCallerFingerprint: options?.authenticatedCallerFingerprint, - orchestrationCaller: resolved.caller - }, + request: admission.request, + method: admission.method, + params: admission.params, + context: rpcContextFromTransport( + this.runtime, + admission.request, + options, + admission.caller + ), orchestrationMutations: this.orchestrationMutations, legacyOrchestration: this.legacyOrchestration }) - const filtered = binding?.filterResult?.(result) ?? { kind: 'allowed', result } + const filtered = admission.binding?.filterResult?.(result) ?? { + kind: 'allowed', + result + } if (filtered.kind === 'denied') { return errorResponse(request.id, meta, 'forbidden', filtered.message) } return successResponse(request.id, meta, filtered.result) } catch (error) { if (request.method.startsWith('emulator.')) { - emulatorProbeError(`rpc ${request.method}`, error, { params: request.params }) + emulatorProbeError(`rpc ${request.method}`, error, { + params: request.params + }) } return mapDispatcherError(request, meta, error) } diff --git a/src/main/runtime/rpc/methods/agent-launch-folder-workspace-creation.ts b/src/main/runtime/rpc/methods/agent-launch-folder-workspace-creation.ts index 560a94c07e3..6a5080f7986 100644 --- a/src/main/runtime/rpc/methods/agent-launch-folder-workspace-creation.ts +++ b/src/main/runtime/rpc/methods/agent-launch-folder-workspace-creation.ts @@ -9,7 +9,7 @@ import type { TuiAgent } from '../../../../shared/tui-agent' import type { AgentLaunchWorkspaceFactory } from '../../../agent-launch/agent-launch-surface-factories' import type { RpcContext } from '../core' import { resolveRpcWorkspaceCreatorProvenance } from '../workspace-creator-context' -import type { AgentLaunchParams } from './agent-launch-schemas' +import type { AgentLaunchParams } from '../../../../shared/rpc-contract/agent-launch-params' type FolderWorkspaceCreateParams = Extract< AgentLaunchParams['target'], diff --git a/src/main/runtime/rpc/methods/agent-launch-intent-resolution.ts b/src/main/runtime/rpc/methods/agent-launch-intent-resolution.ts index 8fad5220e4c..be113c90c28 100644 --- a/src/main/runtime/rpc/methods/agent-launch-intent-resolution.ts +++ b/src/main/runtime/rpc/methods/agent-launch-intent-resolution.ts @@ -6,7 +6,7 @@ import type { AgentLaunchIntent, AgentLaunchTarget } from '../../../../shared/agent-launch-intent' import type { OrcaRuntimeService } from '../../orca-runtime' import { assertOpenCodeModelLaunchPreferencesAbsent } from '../../../opencode/opencode-model-startup-plan' -import type { AgentLaunchParams } from './agent-launch-schemas' +import type { AgentLaunchParams } from '../../../../shared/rpc-contract/agent-launch-params' import type { EarlyAgentLaunchTab } from './agent-launch-tab-publication' /** diff --git a/src/main/runtime/rpc/methods/agent-launch-replay.ts b/src/main/runtime/rpc/methods/agent-launch-replay.ts index a64d0a0cd90..2a15b78fae6 100644 --- a/src/main/runtime/rpc/methods/agent-launch-replay.ts +++ b/src/main/runtime/rpc/methods/agent-launch-replay.ts @@ -30,7 +30,7 @@ import type { import { resolveAgentSessionReplayOutcome } from '../../../native-chat/agent-session-wire/structured-agent-session-replay-outcome' import type { RpcContext } from '../core' import { rpcCallerOperationKey } from '../rpc-caller-identity' -import type { AgentLaunchParams } from './agent-launch-schemas' +import type { AgentLaunchParams } from '../../../../shared/rpc-contract/agent-launch-params' /** * The ledger namespace of whoever the transport says is calling. A transport that could not name its diff --git a/src/main/runtime/rpc/methods/agent-launch-schemas.ts b/src/main/runtime/rpc/methods/agent-launch-schemas.ts deleted file mode 100644 index 15445902d0b..00000000000 --- a/src/main/runtime/rpc/methods/agent-launch-schemas.ts +++ /dev/null @@ -1,6 +0,0 @@ -export { - AgentLaunch, - AgentLaunchFields, - AgentLaunchReplay, - type AgentLaunchParams -} from '../../../../shared/rpc-contract/agent-launch-params' diff --git a/src/main/runtime/rpc/methods/agent-launch-tab-publication.ts b/src/main/runtime/rpc/methods/agent-launch-tab-publication.ts index ba48fada691..26770ed49d9 100644 --- a/src/main/runtime/rpc/methods/agent-launch-tab-publication.ts +++ b/src/main/runtime/rpc/methods/agent-launch-tab-publication.ts @@ -44,7 +44,7 @@ import { readAgentLaunchModeSettings } from '../../../agent-launch/agent-launch-mode' import type { RpcContext } from '../core' -import type { AgentLaunchParams } from './agent-launch-schemas' +import type { AgentLaunchParams } from '../../../../shared/rpc-contract/agent-launch-params' import { agentLaunchOperationCallerKey } from './agent-launch-replay' /** Whether a launch may move the host window at all: a paired device's launch moves only its own diff --git a/src/main/runtime/rpc/methods/agent-launch-worktree-creation.ts b/src/main/runtime/rpc/methods/agent-launch-worktree-creation.ts index e04ff3ceb13..02f1bb86911 100644 --- a/src/main/runtime/rpc/methods/agent-launch-worktree-creation.ts +++ b/src/main/runtime/rpc/methods/agent-launch-worktree-creation.ts @@ -22,7 +22,7 @@ import type { RpcContext } from '../core' import { resolveRpcWorkspaceCreatorProvenance } from '../workspace-creator-context' import { buildManagedWorktreeCreateArgs } from './worktree-create-args' import { toAgentLaunchPreferences } from '../../../../shared/agent-launch-preferences' -import type { AgentLaunchParams } from './agent-launch-schemas' +import type { AgentLaunchParams } from '../../../../shared/rpc-contract/agent-launch-params' import { agentLaunchFolderWorkspaceCreator } from './agent-launch-folder-workspace-creation' import { agentLaunchMovesHostWindow } from './agent-launch-tab-publication' diff --git a/src/main/runtime/rpc/methods/agent-launch.ts b/src/main/runtime/rpc/methods/agent-launch.ts index 442e01a0371..5a37362e916 100644 --- a/src/main/runtime/rpc/methods/agent-launch.ts +++ b/src/main/runtime/rpc/methods/agent-launch.ts @@ -43,7 +43,11 @@ import { settleQuietly, withEarlyTab } from './agent-launch-execution-outcome' -import { AgentLaunch, AgentLaunchReplay, type AgentLaunchParams } from './agent-launch-schemas' +import { + AgentLaunch, + AgentLaunchReplay, + type AgentLaunchParams +} from '../../../../shared/rpc-contract/agent-launch-params' import { agentLaunchSurfaceFactory } from './agent-launch-surfaces' import { agentLaunchFailureCode, diff --git a/src/main/runtime/rpc/methods/agent-skill-sharing-capability-grant.test.ts b/src/main/runtime/rpc/methods/agent-skill-sharing-capability-grant.test.ts index c5e089082bd..80c6ed65826 100644 --- a/src/main/runtime/rpc/methods/agent-skill-sharing-capability-grant.test.ts +++ b/src/main/runtime/rpc/methods/agent-skill-sharing-capability-grant.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import { getDefaultSettings } from '../../../../shared/constants' import { OrcaRuntimeService } from '../../orca-runtime' -import { SettingsUpdate } from './client-settings-schemas' +import { SettingsUpdate } from '../../../../shared/rpc-contract/client-settings-params' vi.mock('electron', () => ({ app: { getPath: () => '/orca-state', isPackaged: true } diff --git a/src/main/runtime/rpc/methods/artifact-sharing-capability-grant.test.ts b/src/main/runtime/rpc/methods/artifact-sharing-capability-grant.test.ts index 1f7ad45a2fb..e10b507d2b6 100644 --- a/src/main/runtime/rpc/methods/artifact-sharing-capability-grant.test.ts +++ b/src/main/runtime/rpc/methods/artifact-sharing-capability-grant.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from 'vitest' import { getDefaultSettings } from '../../../../shared/constants' import { OrcaRuntimeService } from '../../orca-runtime' -import { SettingsUpdate } from './client-settings-schemas' +import { SettingsUpdate } from '../../../../shared/rpc-contract/client-settings-params' function runtimeWithSharing(artifactSharingEnabled: unknown): OrcaRuntimeService { return new OrcaRuntimeService({ diff --git a/src/main/runtime/rpc/methods/automation-schemas.ts b/src/main/runtime/rpc/methods/automation-schemas.ts deleted file mode 100644 index 23962b84ccc..00000000000 --- a/src/main/runtime/rpc/methods/automation-schemas.ts +++ /dev/null @@ -1,10 +0,0 @@ -// Why: the automation method table stays readable only if its field-level validation lives beside it rather than inside it. -export { - AutomationCreate, - AutomationId, - AutomationList, - AutomationRuns, - AutomationUpdate, - Destination, - ExpectedOwner -} from '../../../../shared/rpc-contract/automation-params' diff --git a/src/main/runtime/rpc/methods/automations.ts b/src/main/runtime/rpc/methods/automations.ts index 861eb6e6462..dc4934f27e7 100644 --- a/src/main/runtime/rpc/methods/automations.ts +++ b/src/main/runtime/rpc/methods/automations.ts @@ -7,7 +7,7 @@ import { AutomationList, AutomationRuns, AutomationUpdate -} from './automation-schemas' +} from '../../../../shared/rpc-contract/automation-params' function mutationOwner( id: string, diff --git a/src/main/runtime/rpc/methods/browser-core.ts b/src/main/runtime/rpc/methods/browser-core.ts index dfe1dbf8fd5..f716c1e219d 100644 --- a/src/main/runtime/rpc/methods/browser-core.ts +++ b/src/main/runtime/rpc/methods/browser-core.ts @@ -1,5 +1,5 @@ import { defineMethod } from '../core' -import { BrowserTarget } from '../schemas' +import { BrowserTarget } from '../../../../shared/rpc-contract/rpc-param-primitives' import { Check, Drag, @@ -26,7 +26,10 @@ import { Upload, Wait } from './browser-schemas' -import { BrowserOpenUrlParams, BrowserTabCreateParams } from './browser-tab-create-schema' +import { + BrowserOpenUrlParams, + BrowserTabCreateParams +} from '../../../../shared/rpc-contract/browser-tab-create-params' import { BROWSER_TEXT_METHODS } from './browser-text-rpc-methods' import { BROWSER_PROFILE_METHODS } from './browser-profile-rpc-methods' import { CertificateProceed } from '../../../../shared/rpc-contract/browser-core-params' diff --git a/src/main/runtime/rpc/methods/browser-extras.ts b/src/main/runtime/rpc/methods/browser-extras.ts index 41954070929..d725e7b886c 100644 --- a/src/main/runtime/rpc/methods/browser-extras.ts +++ b/src/main/runtime/rpc/methods/browser-extras.ts @@ -1,6 +1,6 @@ import { defineMethod } from '../core' import { assertRpcClipboardTextWriteWithinLimit } from '../rpc-clipboard-text-validation' -import { BrowserTarget } from '../schemas' +import { BrowserTarget } from '../../../../shared/rpc-contract/rpc-param-primitives' import { ClipboardWrite, CookieDelete, diff --git a/src/main/runtime/rpc/methods/browser-tab-create-schema.test.ts b/src/main/runtime/rpc/methods/browser-tab-create-schema.test.ts index b862825f7c3..fce5793d942 100644 --- a/src/main/runtime/rpc/methods/browser-tab-create-schema.test.ts +++ b/src/main/runtime/rpc/methods/browser-tab-create-schema.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' import { RUNTIME_NAVIGATION_TARGETS } from '../../../../shared/runtime-navigation' -import { BrowserTabCreateParams } from './browser-tab-create-schema' +import { BrowserTabCreateParams } from '../../../../shared/rpc-contract/browser-tab-create-params' describe('browser.tabCreate placement schema', () => { it('keeps placement optional for older clients', () => { diff --git a/src/main/runtime/rpc/methods/browser-tab-create-schema.ts b/src/main/runtime/rpc/methods/browser-tab-create-schema.ts deleted file mode 100644 index ad7c0ee5b75..00000000000 --- a/src/main/runtime/rpc/methods/browser-tab-create-schema.ts +++ /dev/null @@ -1,4 +0,0 @@ -export { - BrowserOpenUrlParams, - BrowserTabCreateParams -} from '../../../../shared/rpc-contract/browser-tab-create-params' diff --git a/src/main/runtime/rpc/methods/client-settings-schemas.ts b/src/main/runtime/rpc/methods/client-settings-schemas.ts deleted file mode 100644 index c8467636d9a..00000000000 --- a/src/main/runtime/rpc/methods/client-settings-schemas.ts +++ /dev/null @@ -1,5 +0,0 @@ -export { - NativeChatSessionOptionsMutation, - PRBotAuthorOverrideUpdate, - SettingsUpdate -} from '../../../../shared/rpc-contract/client-settings-params' diff --git a/src/main/runtime/rpc/methods/client-ui-workspace-filter-fields.ts b/src/main/runtime/rpc/methods/client-ui-workspace-filter-fields.ts deleted file mode 100644 index be6bc445f6e..00000000000 --- a/src/main/runtime/rpc/methods/client-ui-workspace-filter-fields.ts +++ /dev/null @@ -1 +0,0 @@ -export { ClientUiWorkspaceFilterFields } from '../../../../shared/rpc-contract/client-ui-workspace-filter-fields-params' diff --git a/src/main/runtime/rpc/methods/client-ui.ts b/src/main/runtime/rpc/methods/client-ui.ts index a2c1ad20d95..30d83f3ce07 100644 --- a/src/main/runtime/rpc/methods/client-ui.ts +++ b/src/main/runtime/rpc/methods/client-ui.ts @@ -5,12 +5,12 @@ import { NativeChatSessionOptionsMutation, PRBotAuthorOverrideUpdate, SettingsUpdate -} from './client-settings-schemas' +} from '../../../../shared/rpc-contract/client-settings-params' import { FeatureInteractionIdParam, UiUpdate } from './client-ui-schemas' // Type-only side effect: keeps the schema/PersistedUIState parity assertions in // the typecheck graph so drift fails the build instead of a paired client. -import { TerminalQuickCommandsUpdate } from './terminal-quick-command-rpc-schema' +import { TerminalQuickCommandsUpdate } from '../../../../shared/rpc-contract/terminal-quick-command-params' export const CLIENT_UI_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/computer-schemas.ts b/src/main/runtime/rpc/methods/computer-schemas.ts deleted file mode 100644 index f949fa0459c..00000000000 --- a/src/main/runtime/rpc/methods/computer-schemas.ts +++ /dev/null @@ -1,15 +0,0 @@ -export { - Click, - ComputerObserveTarget, - ComputerPermissions, - Drag, - Hotkey, - ListApps, - ListWindows, - PasteText, - PerformSecondaryAction, - PressKey, - Scroll, - SetValue, - TypeText -} from '../../../../shared/rpc-contract/computer-schemas-params' diff --git a/src/main/runtime/rpc/methods/computer.ts b/src/main/runtime/rpc/methods/computer.ts index c580a0d9af8..b9f3402d9d6 100644 --- a/src/main/runtime/rpc/methods/computer.ts +++ b/src/main/runtime/rpc/methods/computer.ts @@ -21,7 +21,7 @@ import { Scroll, SetValue, TypeText -} from './computer-schemas' +} from '../../../../shared/rpc-contract/computer-schemas-params' import { ComputerCapabilitiesParams, ComputerPermissionsStatusParams diff --git a/src/main/runtime/rpc/methods/files-target-schemas.ts b/src/main/runtime/rpc/methods/files-target-schemas.ts deleted file mode 100644 index 945d3c6aa85..00000000000 --- a/src/main/runtime/rpc/methods/files-target-schemas.ts +++ /dev/null @@ -1 +0,0 @@ -export { FileOpen, WorktreeSelector } from '../../../../shared/rpc-contract/files-target-params' diff --git a/src/main/runtime/rpc/methods/files.ts b/src/main/runtime/rpc/methods/files.ts index a9a3d5530d1..7c49771448a 100644 --- a/src/main/runtime/rpc/methods/files.ts +++ b/src/main/runtime/rpc/methods/files.ts @@ -6,7 +6,7 @@ import { remoteFileContentBudget } from './files-remote-content-budget' import { limitQuickOpenSearchReplyBySerializedBytes } from '../../../../shared/quick-open-transport-budget' -import { FileOpen, WorktreeSelector } from './files-target-schemas' +import { FileOpen, WorktreeSelector } from '../../../../shared/rpc-contract/files-target-params' import { FILE_TERMINAL_ARTIFACT_METHODS } from './files-terminal-artifact-methods' import { FilePathsExist, diff --git a/src/main/runtime/rpc/methods/git-admission-tier-schema.ts b/src/main/runtime/rpc/methods/git-admission-tier-schema.ts deleted file mode 100644 index c5366441b4b..00000000000 --- a/src/main/runtime/rpc/methods/git-admission-tier-schema.ts +++ /dev/null @@ -1 +0,0 @@ -export { OptionalGitAdmissionTier } from '../../../../shared/rpc-contract/git-admission-tier-params' diff --git a/src/main/runtime/rpc/methods/git-commit-message-generation-methods.ts b/src/main/runtime/rpc/methods/git-commit-message-generation-methods.ts index 24c28f47f70..9574d1b146f 100644 --- a/src/main/runtime/rpc/methods/git-commit-message-generation-methods.ts +++ b/src/main/runtime/rpc/methods/git-commit-message-generation-methods.ts @@ -6,7 +6,7 @@ import { GitGenerateCommitMessage, GitGeneratePullRequestFields, WorktreeSelector -} from './git-params' +} from '../../../../shared/rpc-contract/git-params' type CommitMessageGenerationOverride = { commitMessageAi?: GlobalSettings['commitMessageAi'] diff --git a/src/main/runtime/rpc/methods/git-diff-methods.ts b/src/main/runtime/rpc/methods/git-diff-methods.ts index 5f83c239772..e14f840faff 100644 --- a/src/main/runtime/rpc/methods/git-diff-methods.ts +++ b/src/main/runtime/rpc/methods/git-diff-methods.ts @@ -1,6 +1,6 @@ import { defineMethod } from '../core' import { remoteRpcContentBudget } from '../../../../shared/remote-rpc-content-budget' -import { GitBranchDiff, GitCommitDiff, GitDiff } from './git-params' +import { GitBranchDiff, GitCommitDiff, GitDiff } from '../../../../shared/rpc-contract/git-params' // Why: clientKind is set only for WebSocket-transported requests, so desktop-local and in-process // callers keep uncapped full-fidelity diffs. diff --git a/src/main/runtime/rpc/methods/git-params.ts b/src/main/runtime/rpc/methods/git-params.ts deleted file mode 100644 index 94a4f7b4b58..00000000000 --- a/src/main/runtime/rpc/methods/git-params.ts +++ /dev/null @@ -1,26 +0,0 @@ -export { - GitBranchCompare, - GitBranchDiff, - GitBulkPaths, - GitBulkStage, - GitCheckIgnored, - GitCheckout, - GitCommit, - GitCommitCompare, - GitCommitDiff, - GitDiff, - GitDiscoverCommitMessageModels, - GitFilePath, - GitForkSync, - GitGenerateCommitMessage, - GitGeneratePullRequestFields, - GitHistory, - GitPush, - GitRebaseFromBase, - GitRemoteCommitUrl, - GitRemoteFileUrl, - GitStatusParams, - GitSubmoduleStatus, - GitTargetedRemote, - WorktreeSelector -} from '../../../../shared/rpc-contract/git-params' diff --git a/src/main/runtime/rpc/methods/git.ts b/src/main/runtime/rpc/methods/git.ts index 89768930133..35c54df7716 100644 --- a/src/main/runtime/rpc/methods/git.ts +++ b/src/main/runtime/rpc/methods/git.ts @@ -20,7 +20,7 @@ import { GitSubmoduleStatus, GitTargetedRemote, WorktreeSelector -} from './git-params' +} from '../../../../shared/rpc-contract/git-params' export const GIT_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/github-issue-update-schema.ts b/src/main/runtime/rpc/methods/github-issue-update-schema.ts deleted file mode 100644 index 6a9f860113b..00000000000 --- a/src/main/runtime/rpc/methods/github-issue-update-schema.ts +++ /dev/null @@ -1 +0,0 @@ -export { IssueUpdate } from '../../../../shared/rpc-contract/github-issue-update-params' diff --git a/src/main/runtime/rpc/methods/github-project-methods.ts b/src/main/runtime/rpc/methods/github-project-methods.ts index 226bcba015d..642391cbdd9 100644 --- a/src/main/runtime/rpc/methods/github-project-methods.ts +++ b/src/main/runtime/rpc/methods/github-project-methods.ts @@ -1,5 +1,5 @@ import { defineMethod } from '../core' -import { SlugRepo } from './github-repo-target-schemas' +import { SlugRepo } from '../../../../shared/rpc-contract/github-repo-target-params' import { ClearProjectItemField, GithubProjectListAccessibleParams, diff --git a/src/main/runtime/rpc/methods/github-repo-target-schemas.ts b/src/main/runtime/rpc/methods/github-repo-target-schemas.ts deleted file mode 100644 index 06d47d47d9b..00000000000 --- a/src/main/runtime/rpc/methods/github-repo-target-schemas.ts +++ /dev/null @@ -1 +0,0 @@ -export { RepoSelector, SlugRepo } from '../../../../shared/rpc-contract/github-repo-target-params' diff --git a/src/main/runtime/rpc/methods/github-repo-work-item-methods.ts b/src/main/runtime/rpc/methods/github-repo-work-item-methods.ts index fabce2b8a0b..fdecb1b483c 100644 --- a/src/main/runtime/rpc/methods/github-repo-work-item-methods.ts +++ b/src/main/runtime/rpc/methods/github-repo-work-item-methods.ts @@ -1,5 +1,5 @@ import { defineMethod } from '../core' -import { RepoSelector } from './github-repo-target-schemas' +import { RepoSelector } from '../../../../shared/rpc-contract/github-repo-target-params' import { IssuesList, RateLimit, diff --git a/src/main/runtime/rpc/methods/linear-issue-attribute-filter-schema.ts b/src/main/runtime/rpc/methods/linear-issue-attribute-filter-schema.ts deleted file mode 100644 index b369a01a15c..00000000000 --- a/src/main/runtime/rpc/methods/linear-issue-attribute-filter-schema.ts +++ /dev/null @@ -1 +0,0 @@ -export { LinearIssueAttributeFilterSchema } from '../../../../shared/rpc-contract/linear-issue-attribute-filter-params' diff --git a/src/main/runtime/rpc/methods/mobile-markdown-tab-methods.ts b/src/main/runtime/rpc/methods/mobile-markdown-tab-methods.ts deleted file mode 100644 index c9eb997f6cd..00000000000 --- a/src/main/runtime/rpc/methods/mobile-markdown-tab-methods.ts +++ /dev/null @@ -1,24 +0,0 @@ -import { defineMethod } from '../core' -import { ActivateTab, SaveMarkdownTab } from './session-tabs-schemas' - -export const MOBILE_MARKDOWN_TAB_METHODS = [ - defineMethod({ - name: 'markdown.readTab', - permission: 'workspace', - params: ActivateTab, - handler: async (params, { runtime }) => - runtime.readMobileMarkdownTab(params.worktree, params.tabId) - }), - defineMethod({ - name: 'markdown.saveTab', - permission: 'workspace', - params: SaveMarkdownTab, - handler: async (params, { runtime }) => - runtime.saveMobileMarkdownTab( - params.worktree, - params.tabId, - params.baseVersion, - params.content - ) - }) -] diff --git a/src/main/runtime/rpc/methods/orchestration/schemas.ts b/src/main/runtime/rpc/methods/orchestration/schemas.ts index eae80849ffc..02f47d3ed1b 100644 --- a/src/main/runtime/rpc/methods/orchestration/schemas.ts +++ b/src/main/runtime/rpc/methods/orchestration/schemas.ts @@ -1,6 +1,10 @@ import { z } from 'zod' import { setImmediate as yieldToEventLoop } from 'node:timers/promises' -import { OptionalString, OptionalBoolean, requiredString } from '../../schemas' +import { + OptionalString, + OptionalBoolean, + requiredString +} from '../../../../../shared/rpc-contract/rpc-param-primitives' import type { TaskStatus } from '../../../orchestration/db' import { isGroupAddress } from '../../../orchestration/groups' import { MESSAGE_TYPES } from '../../../orchestration/types' diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts index 315a8cffdf4..3520ae7e049 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts @@ -23,7 +23,16 @@ import { import { projectWorkerFleet, type WorkerListPageParams } from './worker-list-projection' import { chatAssigneeObserver } from './session-worker-observation' import { exposeWorkerTerminalResource } from './worker-release-completion' -import { WORKER_TERMINAL_LIST_STATES, WorkerListParams } from './worker-release-schemas' +import { + WORKER_TERMINAL_LIST_STATES, + WorkerListParams +} from '../../../../../../shared/rpc-contract/orchestration-worker-release-schemas-params' + +const emptyWorkerListPage = (limit: number) => ({ + workers: [], + counts: {}, + page: { limit, total: 0, hasMore: false, nextCursor: null } +}) export const ORCHESTRATION_WORKER_LIST_METHOD = defineMethod({ name: 'orchestration.workerList', @@ -68,11 +77,7 @@ export const ORCHESTRATION_WORKER_LIST_METHOD = defineMethod({ } const snapshot = db.getWorkerTerminalListingSnapshot(params.run) if (!snapshot) { - return { - workers: [], - counts: {}, - page: { limit, total: 0, hasMore: false, nextCursor: null } - } + return emptyWorkerListPage(limit) } cursor = { version: 2, snapshot, after: legacyKey } } @@ -87,11 +92,7 @@ export const ORCHESTRATION_WORKER_LIST_METHOD = defineMethod({ } const snapshot = cursor?.snapshot ?? db.getWorkerTerminalListingSnapshot(params.run) if (!snapshot) { - return { - workers: [], - counts: {}, - page: { limit, total: 0, hasMore: false, nextCursor: null } - } + return emptyWorkerListPage(limit) } const rows = db.listWorkerTerminalResources({ runId: params.run, diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-release-schemas.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-release-schemas.ts deleted file mode 100644 index 66eaf2263f9..00000000000 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-release-schemas.ts +++ /dev/null @@ -1,6 +0,0 @@ -export { - WORKER_TERMINAL_LIST_STATES, - WorkerDispatchParams, - WorkerListParams, - WorkerRetainParams -} from '../../../../../../shared/rpc-contract/orchestration-worker-release-schemas-params' diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-release.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-release.ts index 976df7ba83d..b0f08dffea3 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-release.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-release.ts @@ -8,7 +8,10 @@ import { completeWorkerTerminalRelease, type WorkerReleaseReceipt } from './worker-release-completion' -import { WorkerDispatchParams, WorkerRetainParams } from './worker-release-schemas' +import { + WorkerDispatchParams, + WorkerRetainParams +} from '../../../../../../shared/rpc-contract/orchestration-worker-release-schemas-params' import { OrchestrationWorkerTerminalUserInputParams } from '../../../../../../shared/rpc-contract/orchestration-worker-release-params' export const ORCHESTRATION_WORKER_RELEASE_METHODS = [ diff --git a/src/main/runtime/rpc/methods/repo-update-schema.ts b/src/main/runtime/rpc/methods/repo-update-schema.ts deleted file mode 100644 index f4e189f84a7..00000000000 --- a/src/main/runtime/rpc/methods/repo-update-schema.ts +++ /dev/null @@ -1,4 +0,0 @@ -export { - RepoSourceControlAiOverrides, - createRepoUpdateSchema -} from '../../../../shared/rpc-contract/repo-update-params' diff --git a/src/main/runtime/rpc/methods/repo.ts b/src/main/runtime/rpc/methods/repo.ts index b3725e17e0e..42563096896 100644 --- a/src/main/runtime/rpc/methods/repo.ts +++ b/src/main/runtime/rpc/methods/repo.ts @@ -5,7 +5,7 @@ import { includesQualifiedSearchRefs, projectRepoSearchRefsForClient } from './repo-search-ref-projection' -import { RepoSelector } from './github-repo-target-schemas' +import { RepoSelector } from '../../../../shared/rpc-contract/github-repo-target-params' import { projectRepoResultVisibilityForClient, projectRepoVisibilityForClient diff --git a/src/main/runtime/rpc/methods/session-tab-close-methods.ts b/src/main/runtime/rpc/methods/session-tab-close-methods.ts index ef2176c0860..b32b8fbdf48 100644 --- a/src/main/runtime/rpc/methods/session-tab-close-methods.ts +++ b/src/main/runtime/rpc/methods/session-tab-close-methods.ts @@ -1,7 +1,10 @@ import { withSpan } from '../../../observability/tracer' import { SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' import { defineMethod } from '../core' -import { CloseLifecycleTab, CloseTab } from './session-tabs-schemas' +import { + CloseLifecycleTab, + CloseTab +} from '../../../../shared/rpc-contract/session-tabs-schemas-params' import { assertProjectedSessionTabVisible } from './session-tab-browser-placement-projection' import { assertAgentSessionTabDestructiveMutationSupported } from './session-tab-agent-status-projection' import { projectSessionTabsForClient } from './session-tabs-inventory' diff --git a/src/main/runtime/rpc/methods/session-tab-markdown-methods.ts b/src/main/runtime/rpc/methods/session-tab-markdown-methods.ts index 58cbaf909f9..f1d7f1ccb38 100644 --- a/src/main/runtime/rpc/methods/session-tab-markdown-methods.ts +++ b/src/main/runtime/rpc/methods/session-tab-markdown-methods.ts @@ -1,5 +1,8 @@ import { defineMethod } from '../core' -import { ActivateTab, SaveMarkdownTab } from './session-tabs-schemas' +import { + ActivateTab, + SaveMarkdownTab +} from '../../../../shared/rpc-contract/session-tabs-schemas-params' export const SESSION_TAB_MARKDOWN_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/session-tab-mutation-methods.ts b/src/main/runtime/rpc/methods/session-tab-mutation-methods.ts index 76b72fb992f..6cdfb2bd0fd 100644 --- a/src/main/runtime/rpc/methods/session-tab-mutation-methods.ts +++ b/src/main/runtime/rpc/methods/session-tab-mutation-methods.ts @@ -6,7 +6,12 @@ import { translateProjectedSessionTabMove } from './session-tab-browser-placement-projection' import { projectSessionTabsForClient } from './session-tabs-inventory' -import { ActivateTab, MoveTab, SetTabProps, UpdatePaneLayout } from './session-tabs-schemas' +import { + ActivateTab, + MoveTab, + SetTabProps, + UpdatePaneLayout +} from '../../../../shared/rpc-contract/session-tabs-schemas-params' export const SESSION_TAB_MUTATION_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/session-tabs-pane-layout-intent.test.ts b/src/main/runtime/rpc/methods/session-tabs-pane-layout-intent.test.ts index 5b6cc92ee57..968e1234d31 100644 --- a/src/main/runtime/rpc/methods/session-tabs-pane-layout-intent.test.ts +++ b/src/main/runtime/rpc/methods/session-tabs-pane-layout-intent.test.ts @@ -3,7 +3,7 @@ import { describe, expect, it, vi } from 'vitest' import type { OrcaRuntimeService } from '../../orca-runtime' import { RpcDispatcher } from '../dispatcher' import { SESSION_TAB_METHODS } from './session-tabs' -import { UpdatePaneLayout } from './session-tabs-schemas' +import { UpdatePaneLayout } from '../../../../shared/rpc-contract/session-tabs-schemas-params' const LAYOUT_PARAMS = { worktree: 'id:wt-1', diff --git a/src/main/runtime/rpc/methods/session-tabs-schemas.test.ts b/src/main/runtime/rpc/methods/session-tabs-schemas.test.ts index c4a97c0400e..5009ff462e8 100644 --- a/src/main/runtime/rpc/methods/session-tabs-schemas.test.ts +++ b/src/main/runtime/rpc/methods/session-tabs-schemas.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from 'vitest' -import { ActivateTab, CloseLifecycleTab, CloseTab, UpdatePaneLayout } from './session-tabs-schemas' +import { + ActivateTab, + CloseLifecycleTab, + CloseTab, + UpdatePaneLayout +} from '../../../../shared/rpc-contract/session-tabs-schemas-params' const WT = 'id:wt' diff --git a/src/main/runtime/rpc/methods/session-tabs-schemas.ts b/src/main/runtime/rpc/methods/session-tabs-schemas.ts deleted file mode 100644 index ecd434f1e0a..00000000000 --- a/src/main/runtime/rpc/methods/session-tabs-schemas.ts +++ /dev/null @@ -1,14 +0,0 @@ -export { - ActivateTab, - CloseLifecycleTab, - CloseTab, - CreateTerminalTab, - MoveTab, - SaveMarkdownTab, - SessionTabsUnsubscribe, - SetTabProps, - TerminalPaneLayoutNodeSchema, - UpdatePaneLayout, - WorktreeTabSelector -} from '../../../../shared/rpc-contract/session-tabs-schemas-params' -export type { TerminalPaneLayoutNodeInput } from '../../../../shared/rpc-contract/session-tabs-schemas-params' diff --git a/src/main/runtime/rpc/methods/session-tabs.ts b/src/main/runtime/rpc/methods/session-tabs.ts index e53349c3e5b..e3be5d2e374 100644 --- a/src/main/runtime/rpc/methods/session-tabs.ts +++ b/src/main/runtime/rpc/methods/session-tabs.ts @@ -5,7 +5,7 @@ import { CreateTerminalTab, SessionTabsUnsubscribe, WorktreeTabSelector -} from './session-tabs-schemas' +} from '../../../../shared/rpc-contract/session-tabs-schemas-params' import { SESSION_TAB_CLOSE_METHODS } from './session-tab-close-methods' import { listSessionTabsInventory, diff --git a/src/main/runtime/rpc/methods/terminal-quick-command-rpc-schema.ts b/src/main/runtime/rpc/methods/terminal-quick-command-rpc-schema.ts deleted file mode 100644 index 9d4511bbd4d..00000000000 --- a/src/main/runtime/rpc/methods/terminal-quick-command-rpc-schema.ts +++ /dev/null @@ -1 +0,0 @@ -export { TerminalQuickCommandsUpdate } from '../../../../shared/rpc-contract/terminal-quick-command-params' diff --git a/src/main/runtime/rpc/methods/terminal/stream-schemas.ts b/src/main/runtime/rpc/methods/terminal/stream-schemas.ts index c0c71ba804e..61dd9224e53 100644 --- a/src/main/runtime/rpc/methods/terminal/stream-schemas.ts +++ b/src/main/runtime/rpc/methods/terminal/stream-schemas.ts @@ -1,12 +1,7 @@ import { z } from 'zod' -import { requiredString } from '../../schemas' -import { TerminalViewport } from './unary-schemas' +import { requiredString } from '../../../../../shared/rpc-contract/rpc-param-primitives' +import { TerminalViewport } from '../../../../../shared/rpc-contract/terminal-unary-params' import { TerminalHandle } from '../../../../../shared/rpc-contract/terminal-stream-params' -export { - TerminalMultiplex, - TerminalResizeForClient, - TerminalSubscribe -} from '../../../../../shared/rpc-contract/terminal-stream-params' export const TerminalMultiplexSubscribeFrame = TerminalHandle.extend({ streamId: z.number().int().min(1), diff --git a/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts b/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts index 7dc4ea6c01b..da0c55e0337 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts @@ -48,7 +48,7 @@ export function resolveMobileFloorClientId( export type TerminalStreamInputOutcome = 'delivered' | 'rejected' | 'failed' -export function isTerminalStreamInputRejection(error: unknown): boolean { +function isTerminalStreamInputRejection(error: unknown): boolean { const message = error instanceof Error ? error.message : String(error) return message.includes('terminal_not_writable') || message.includes('terminal_handle_stale') } @@ -69,24 +69,21 @@ export async function sendTerminalStreamInput( const floorClaim: MobileInputFloorClaimHolder = { current: null } const settlement = args.requireWriteSettlement ? { requireWriteSettlement: true as const } : {} try { - if (!clientId) { - const result = await runtime.sendTerminal(args.terminal, action, { - inputKind: 'driving', - ...settlement - }) - return streamInputOutcome(result) - } const result = await runtime.sendTerminal(args.terminal, action, { inputKind: 'driving', ...settlement, - reserveWrite: (writePtyId) => { - const claim = runtime.beginMobileInputFloor(writePtyId, clientId) - if (!claim) { - throw new Error('mobile_input_floor_unavailable') - } - floorClaim.current = claim - }, - afterWrite: () => commitMobileInputFloorClaim(floorClaim) + ...(clientId + ? { + reserveWrite: (writePtyId: string) => { + const claim = runtime.beginMobileInputFloor(writePtyId, clientId) + if (!claim) { + throw new Error('mobile_input_floor_unavailable') + } + floorClaim.current = claim + }, + afterWrite: () => commitMobileInputFloorClaim(floorClaim) + } + : {}) }) const outcome = streamInputOutcome(result) if (outcome === 'rejected') { diff --git a/src/main/runtime/rpc/methods/terminal/terminal-inspect-process-params.test.ts b/src/main/runtime/rpc/methods/terminal/terminal-inspect-process-params.test.ts index 8377f923ecc..de6e2e0222b 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-inspect-process-params.test.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-inspect-process-params.test.ts @@ -5,7 +5,10 @@ import { describe, expect, it, vi } from 'vitest' import type { ZodType } from 'zod' import { TERMINAL_QUERY_METHODS } from './terminal-query-methods' -import { TerminalHandle, TerminalInspectProcess } from './unary-schemas' +import { + TerminalHandle, + TerminalInspectProcess +} from '../../../../../shared/rpc-contract/terminal-unary-params' import { eraseRpcMethods } from '../../core' /** The method as registered, so a schema swap on the definition cannot pass unseen. */ diff --git a/src/main/runtime/rpc/methods/terminal/terminal-legacy-subscription-types.ts b/src/main/runtime/rpc/methods/terminal/terminal-legacy-subscription-types.ts index 595a65a5eeb..543760e3b65 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-legacy-subscription-types.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-legacy-subscription-types.ts @@ -2,7 +2,7 @@ import type { z } from 'zod' import type { RpcContext } from '../../core' import type { TerminalReplyQuerySequence } from '../../../../../shared/terminal-reply-query-scan' import type { TerminalOutputChunk } from './terminal-stream-types' -import type { TerminalSubscribe } from './stream-schemas' +import type { TerminalSubscribe } from '../../../../../shared/rpc-contract/terminal-stream-params' import type { TerminalOutputBatcher } from './terminal-output-batcher' import type { TerminalStreamOpcode } from '../../../../../shared/terminal-stream-protocol' import type { TerminalSubscriptionRegistration } from './terminal-subscription-registration' diff --git a/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts b/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts index e0775c87f7c..98a2e1a396d 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts @@ -16,8 +16,8 @@ import { TerminalStop, TerminalStopExact, TerminalWait -} from './unary-schemas' -import { TerminalResizeForClient } from './stream-schemas' +} from '../../../../../shared/rpc-contract/terminal-unary-params' +import { TerminalResizeForClient } from '../../../../../shared/rpc-contract/terminal-stream-params' export const TERMINAL_LIFECYCLE_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts index a2a6de449f7..e46b389c1b9 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts @@ -1,13 +1,6 @@ -import type { - TerminalMultiplexCleanupStage, - TerminalMultiplexConnection, - TerminalMultiplexFlowControlStage -} from './terminal-multiplex-connection' +import type { TerminalMultiplexConnection } from './terminal-multiplex-connection' -export function installMultiplexCleanup( - build: TerminalMultiplexFlowControlStage -): asserts build is TerminalMultiplexCleanupStage { - const state = build as TerminalMultiplexConnection +export function installMultiplexCleanup(state: TerminalMultiplexConnection): void { const { runtime, streams, pendingPtyWaitControllers, emit, signal } = state state.detachStream = (streamId: number, endVerdict, releaseRemoteDesktopDriver = true): void => { const stream = streams.get(streamId) diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-connection.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-connection.ts index 094b2b4e94b..a573b0a1153 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-connection.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-connection.ts @@ -102,17 +102,16 @@ export type TerminalMultiplexSubscribeFrame = { handleSubscribeFrame: (payload: Uint8Array) => Promise } -export type TerminalMultiplexFrameDeliveryStage = TerminalMultiplexConnectionBase & - TerminalMultiplexFrameDelivery - -export type TerminalMultiplexFlowControlStage = TerminalMultiplexFrameDeliveryStage & - TerminalMultiplexFlowControl - -export type TerminalMultiplexCleanupStage = TerminalMultiplexFlowControlStage & - TerminalMultiplexCleanup - -export type TerminalMultiplexSlotFramesStage = TerminalMultiplexCleanupStage & - TerminalMultiplexSlotFrames - -export type TerminalMultiplexConnection = TerminalMultiplexSlotFramesStage & +export type TerminalMultiplexConnection = TerminalMultiplexConnectionBase & + TerminalMultiplexFrameDelivery & + TerminalMultiplexFlowControl & + TerminalMultiplexCleanup & + TerminalMultiplexSlotFrames & TerminalMultiplexSubscribeFrame + +export function isMultiplexStreamAttached( + state: Pick, + stream: TerminalMultiplexStream +): boolean { + return !state.closed && state.streams.get(stream.streamId) === stream +} diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-flow-control.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-flow-control.ts index 6d328ecd0f3..d7aa121c2f2 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-flow-control.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-flow-control.ts @@ -9,18 +9,33 @@ import { sendSnapshotFrames, serializeBudgetedRequestedSnapshot } from './terminal-snapshot-publication' -import type { - TerminalMultiplexConnection, - TerminalMultiplexFlowControlStage, - TerminalMultiplexFrameDeliveryStage +import { + isMultiplexStreamAttached, + type TerminalMultiplexConnection } from './terminal-multiplex-connection' -import type { TerminalMultiplexStream } from './terminal-stream-types' -import type { RemoteTerminalSourceRangeReplacementReservation } from '../../../remote-terminal-source-range-consumer' +import type { SerializedSnapshot, TerminalMultiplexStream } from './terminal-stream-types' +import type { + RemoteTerminalSourceRangeReplacementPublication, + RemoteTerminalSourceRangeReplacementReservation +} from '../../../remote-terminal-source-range-consumer' -export function installMultiplexFlowControl( - build: TerminalMultiplexFrameDeliveryStage -): asserts build is TerminalMultiplexFlowControlStage { - const state = build as TerminalMultiplexConnection +const sourceRangeStreamIdentity = (stream: TerminalMultiplexStream) => ({ + ptyId: stream.ptyId, + consumerId: stream.remoteDesktopSubscriptionKey, + streamGeneration: stream.streamGeneration +}) + +function requireRecoverySourceIdentity( + serialized: Pick, 'source' | 'seq'> +): RemoteTerminalSourceRangeReplacementPublication { + const { source, seq } = serialized + if (source === undefined || typeof seq !== 'number') { + throw new Error('Remote terminal recovery snapshot source identity unavailable.') + } + return { source, seq } +} + +export function installMultiplexFlowControl(state: TerminalMultiplexConnection): void { const { runtime, streams } = state // Why: snapshot frames bypass the ACK window, so only a drained link can afford history; // a recovery on every ACK of a sustained flood would otherwise outgrow the client's intake. @@ -29,8 +44,7 @@ export function installMultiplexFlowControl( state.ackTotalInFlightBytes <= TERMINAL_MULTIPLEX_ACK_BATCH_BYTES state.sendAckRecoverySnapshot = async (stream: TerminalMultiplexStream): Promise => { if ( - state.closed || - streams.get(stream.streamId) !== stream || + !isMultiplexStreamAttached(state, stream) || stream.outputPaused || stream.ackRecoverySnapshotInFlight ) { @@ -46,30 +60,19 @@ export function installMultiplexFlowControl( stream.ptyId, carriesHistory ? TERMINAL_MULTIPLEX_RECOVERY_SCROLLBACK_ROWS : 0 ) - if (state.closed || streams.get(stream.streamId) !== stream || stream.outputPaused) { + if (!isMultiplexStreamAttached(state, stream) || stream.outputPaused) { return } if (!serialized) { throw new Error('Remote terminal recovery snapshot unavailable.') } - if ( - stream.ackOutputSourceRanges && - (serialized.source === undefined || typeof serialized.seq !== 'number') - ) { - throw new Error('Remote terminal recovery snapshot source identity unavailable.') - } - if ( - stream.ackOutputSourceRanges && - serialized.source !== undefined && - typeof serialized.seq === 'number' - ) { + const identity = stream.ackOutputSourceRanges + ? requireRecoverySourceIdentity(serialized) + : null + if (identity) { replacement = runtime.reserveRemoteTerminalSourceRangeReplacement( - { - ptyId: stream.ptyId, - consumerId: stream.remoteDesktopSubscriptionKey, - streamGeneration: stream.streamGeneration - }, - serialized.seq, + sourceRangeStreamIdentity(stream), + identity.seq, 'ack-pending-overflow' ) stream.sourceRangeReplacement = replacement @@ -77,8 +80,7 @@ export function installMultiplexFlowControl( const displayMode = runtime.getMobileDisplayMode(stream.ptyId) const publication = sendSnapshotFrames( (opcode, payload) => - !state.closed && - streams.get(stream.streamId) === stream && + isMultiplexStreamAttached(state, stream) && state.sendFrame(stream.streamId, opcode, payload), { kind: 'scrollback', @@ -99,25 +101,21 @@ export function installMultiplexFlowControl( if (!publication.published) { throw new Error('Remote terminal recovery snapshot was not published.') } - if (state.closed || streams.get(stream.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { throw new Error('Remote terminal recovery snapshot stream detached.') } - const localReplacement = replacement - ? typeof serialized.seq === 'number' - ? stream.sourceRangeLedger?.planSourceRangeReplacement(serialized.seq) + // Why `replacement`: reservation is null when no source-range consumer is attached. + const localReplacement = + identity && replacement + ? stream.sourceRangeLedger?.planSourceRangeReplacement(identity.seq) : null - : null if (replacement && !localReplacement) { throw new Error('Remote terminal recovery source ledger replacement unavailable.') } if ( + identity && replacement && - (!serialized.source || - typeof serialized.seq !== 'number' || - !runtime.commitRemoteTerminalSourceRangeReplacement(replacement, { - source: serialized.source, - seq: serialized.seq - })) + !runtime.commitRemoteTerminalSourceRangeReplacement(replacement, identity) ) { throw new Error('Remote terminal recovery snapshot replacement was not accepted.') } @@ -148,7 +146,7 @@ export function installMultiplexFlowControl( } replacement = null } - if (state.closed || streams.get(stream.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return } state.sendStreamError( @@ -253,14 +251,7 @@ export function installMultiplexFlowControl( return } if (result.settled.length > 0) { - runtime.settleRemoteTerminalSourceRanges( - { - ptyId: stream.ptyId, - consumerId: stream.remoteDesktopSubscriptionKey, - streamGeneration: stream.streamGeneration - }, - result.settled - ) + runtime.settleRemoteTerminalSourceRanges(sourceRangeStreamIdentity(stream), result.settled) } state.acknowledgeOutput(stream, result.acknowledgedBytes) } @@ -274,15 +265,10 @@ export function installMultiplexFlowControl( if (!ledger) { return } - const identity = { - ptyId: stream.ptyId, - consumerId: stream.remoteDesktopSubscriptionKey, - streamGeneration: stream.streamGeneration - } const transfer = ledger.beginTransfer() const ranges = transfer.frames.flatMap((frame) => frame.sourceRanges) try { - runtime.cancelRemoteTerminalSourceRanges(identity, ranges, reason) + runtime.cancelRemoteTerminalSourceRanges(sourceRangeStreamIdentity(stream), ranges, reason) } finally { transfer.commit() } diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts index 33ad0d3a664..09682e10c08 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts @@ -6,20 +6,16 @@ import { encodeTerminalStreamText } from '../../../../../shared/terminal-stream-protocol' import { appendAckPendingOutput } from './terminal-stream-replay' -import type { - TerminalMultiplexConnection, - TerminalMultiplexConnectionBase, - TerminalMultiplexFrameDeliveryStage +import { + isMultiplexStreamAttached, + type TerminalMultiplexConnection } from './terminal-multiplex-connection' import type { TerminalOutputFrameChunk } from '../../terminal-output-frame-chunks' import type { TerminalStreamInputOutcome } from './terminal-input-delivery' import type { TerminalMultiplexStream } from './terminal-stream-types' -export function installMultiplexFrameDelivery( - build: TerminalMultiplexConnectionBase -): asserts build is TerminalMultiplexFrameDeliveryStage { - const state = build as TerminalMultiplexConnection - const { sendBinary, emit, streams } = state +export function installMultiplexFrameDelivery(state: TerminalMultiplexConnection): void { + const { sendBinary, emit } = state state.sendFrame = ( streamId: number, opcode: TerminalStreamOpcode, @@ -59,8 +55,7 @@ export function installMultiplexFrameDelivery( outcome: TerminalStreamInputOutcome ): void => { if ( - state.closed || - streams.get(stream.streamId) !== stream || + !isMultiplexStreamAttached(state, stream) || outcome !== 'rejected' || !stream.supportsWriteUnavailable ) { @@ -70,7 +65,7 @@ export function installMultiplexFrameDelivery( } state.sendInputAck = (stream, appliedSeq, kind): void => { // Why: a detached stream's client already resends on its successor, which acks again. - if (state.closed || streams.get(stream.streamId) !== stream || stream.inputSessionId === null) { + if (!isMultiplexStreamAttached(state, stream) || stream.inputSessionId === null) { return } state.sendFrame( @@ -150,7 +145,7 @@ export function installMultiplexFrameDelivery( stream: TerminalMultiplexStream, chunk: TerminalOutputFrameChunk ): void => { - if (state.closed || streams.get(stream.streamId) !== stream || stream.outputPaused) { + if (!isMultiplexStreamAttached(state, stream) || stream.outputPaused) { return } if ( diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts index 34b910b976c..f5c175170a3 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts @@ -4,9 +4,10 @@ import { } from './terminal-snapshot-publication' import { getOutputAfterSnapshotSeq } from './terminal-stream-replay' import { getTerminalInputSequenceLedger } from './terminal-input-sequence-ledger' -import type { - MultiplexSubscribeRequest, - TerminalMultiplexConnection +import { + isMultiplexStreamAttached, + type MultiplexSubscribeRequest, + type TerminalMultiplexConnection } from './terminal-multiplex-connection' import type { TerminalMultiplexStream } from './terminal-stream-types' @@ -21,14 +22,14 @@ export async function publishMultiplexInitialSnapshot( request: MultiplexSubscribeRequest, stream: TerminalMultiplexStream ): Promise { - const { runtime, streams, emit } = state + const { runtime, emit } = state const { ptyId } = stream const isMobile = stream.isMobile const forcedInitialSnapshotTruncated = process.env.ORCA_E2E_FORCE_REMOTE_TERMINAL_INITIAL_SNAPSHOT_TRUNCATED === '1' let read = await runtime.readTerminal(request.terminal) let serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId) - if (state.closed || streams.get(request.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return null } let initialOutputOverflowed = forcedInitialSnapshotTruncated @@ -38,7 +39,7 @@ export async function publishMultiplexInitialSnapshot( stream.pendingOutputOverflowed = false read = await runtime.readTerminal(request.terminal) serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId) - if (state.closed || streams.get(request.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return null } if (stream.pendingOutputOverflowed) { diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-live-stream.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-live-stream.ts index 24a83ce745f..317b121f387 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-live-stream.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-live-stream.ts @@ -1,8 +1,9 @@ import { sendMobileResizeRestream } from './terminal-snapshot-publication' import { updateViewportForClient } from './terminal-viewport-update' -import type { - MultiplexSubscribeRequest, - TerminalMultiplexConnection +import { + isMultiplexStreamAttached, + type MultiplexSubscribeRequest, + type TerminalMultiplexConnection } from './terminal-multiplex-connection' import type { MultiplexPublishedInitialState } from './terminal-multiplex-initial-snapshot' import type { TerminalMultiplexStream } from './terminal-stream-types' @@ -70,14 +71,11 @@ export function activateMultiplexStream( (opcode, payload) => state.sendFrame(request.streamId, opcode, payload), event, () => - !state.closed && - streams.get(request.streamId) === stream && - stream.resizeGeneration === resizeGeneration + isMultiplexStreamAttached(state, stream) && stream.resizeGeneration === resizeGeneration ) .then((restreamed) => { if ( - state.closed || - streams.get(request.streamId) !== stream || + !isMultiplexStreamAttached(state, stream) || stream.resizeGeneration !== resizeGeneration ) { return @@ -89,8 +87,7 @@ export function activateMultiplexStream( // Why: on re-stream failure, still emit the geometry-only Resized frame so the client never misses the resize. .catch(() => { if ( - state.closed || - streams.get(request.streamId) !== stream || + !isMultiplexStreamAttached(state, stream) || stream.resizeGeneration !== resizeGeneration ) { return diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-method.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-method.ts index bfd14a48c09..eb7b1e65434 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-method.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-method.ts @@ -2,8 +2,8 @@ import { defineStreamingMethod } from '../../core' import { TerminalStreamOpcode } from '../../../../../shared/terminal-stream-protocol' import { TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES } from '../../../../../shared/terminal-multiplex-flow-control' import { TerminalSourceRangeRegistry } from '../../terminal-source-range-registry' -import { TerminalMultiplex } from './stream-schemas' -import type { TerminalMultiplexConnectionBase } from './terminal-multiplex-connection' +import { TerminalMultiplex } from '../../../../../shared/rpc-contract/terminal-stream-params' +import type { TerminalMultiplexConnection } from './terminal-multiplex-connection' import type { TerminalMultiplexStream } from './terminal-stream-types' import { installMultiplexFrameDelivery } from './terminal-multiplex-frame-delivery' import { installMultiplexFlowControl } from './terminal-multiplex-flow-control' @@ -29,7 +29,8 @@ export const TERMINAL_MULTIPLEX_METHODS = [ resolveMultiplex = resolve }) // Installers only close over this per-connection state; none of it is module-global. - const state: TerminalMultiplexConnectionBase = { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: installers below fill every method synchronously, and no frame can arrive before the control handler is registered after them. + const state = { runtime, connectionId, sendBinary, @@ -47,7 +48,7 @@ export const TERMINAL_MULTIPLEX_METHODS = [ resolveMultiplex, multiplexClosed, unregisterControlHandler: () => {} - } + } as TerminalMultiplexConnection installMultiplexFrameDelivery(state) installMultiplexFlowControl(state) diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts index 1b1e8762c99..b493cfaa1a5 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts @@ -19,22 +19,42 @@ import { serializeBudgetedRequestedSnapshot } from './terminal-snapshot-publication' import { updateViewportForClient } from './terminal-viewport-update' -import type { - MultiplexSnapshotRequest, - TerminalMultiplexCleanupStage, - TerminalMultiplexConnection, - TerminalMultiplexSlotFramesStage +import { + isMultiplexStreamAttached, + type MultiplexSnapshotRequest, + type TerminalMultiplexConnection } from './terminal-multiplex-connection' -import type { TerminalMultiplexStream } from './terminal-stream-types' +import type { TerminalMultiplexStream, TerminalViewportClient } from './terminal-stream-types' -export function installMultiplexSlotFrames( - build: TerminalMultiplexCleanupStage -): asserts build is TerminalMultiplexSlotFramesStage { - const state = build as TerminalMultiplexConnection +function decodeViewportFrame( + payload: TerminalStreamFrame['payload'] +): { cols: number; rows: number } | null { + const viewport = decodeTerminalStreamJson<{ cols?: unknown; rows?: unknown }>(payload) + return viewport && typeof viewport.cols === 'number' && typeof viewport.rows === 'number' + ? { cols: viewport.cols, rows: viewport.rows } + : null +} + +export function installMultiplexSlotFrames(state: TerminalMultiplexConnection): void { const { runtime, streams } = state const inputSequenceLedger = getTerminalInputSequenceLedger(runtime) + const applyStreamViewport = ( + stream: TerminalMultiplexStream, + client: TerminalViewportClient, + viewport: { cols: number; rows: number } + ) => + updateViewportForClient( + runtime, + stream.ptyId, + stream.remoteDesktopSubscriptionKey, + client, + viewport, + stream.isMobile ? 'mobile' : 'desktop', + 'register', + !stream.supportsDesktopViewportClaims + ) state.handleSlotFrame = (stream: TerminalMultiplexStream, frame: TerminalStreamFrame): void => { - if (state.closed || streams.get(stream.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return } if (frame.opcode === TerminalStreamOpcode.Unsubscribe) { @@ -80,32 +100,21 @@ export function installMultiplexSlotFrames( return } if (frame.opcode === TerminalStreamOpcode.Resize && stream.client) { - const viewport = decodeTerminalStreamJson<{ cols?: unknown; rows?: unknown }>(frame.payload) - if (!viewport || typeof viewport.cols !== 'number' || typeof viewport.rows !== 'number') { + const viewport = decodeViewportFrame(frame.payload) + if (!viewport) { return } - const cols = viewport.cols - const rows = viewport.rows // Why: resize registers stream-scoped geometry so detach can release it; older clients lack explicit claims. if (!stream.isMobile && stream.client?.id) { stream.registeredRemoteDesktopDriver = true if (stream.buffering) { - stream.pendingRemoteDesktopViewport = { cols: viewport.cols, rows: viewport.rows } + stream.pendingRemoteDesktopViewport = viewport return } } stream.desktopClaimTail = stream.desktopClaimTail .then(async (priorClaimed) => { - const result = await updateViewportForClient( - runtime, - stream.ptyId, - stream.remoteDesktopSubscriptionKey, - stream.client!, - { cols, rows }, - stream.isMobile ? 'mobile' : 'desktop', - 'register', - !stream.supportsDesktopViewportClaims - ) + const result = await applyStreamViewport(stream, stream.client!, viewport) return stream.supportsDesktopViewportClaims ? priorClaimed && result.applied : result.applied @@ -114,35 +123,21 @@ export function installMultiplexSlotFrames( return } if (frame.opcode === TerminalStreamOpcode.ClaimViewport && stream.client && !stream.isMobile) { - const viewport = decodeTerminalStreamJson<{ cols?: unknown; rows?: unknown }>(frame.payload) - if (!viewport || typeof viewport.cols !== 'number' || typeof viewport.rows !== 'number') { + const viewport = decodeViewportFrame(frame.payload) + if (!viewport) { return } - const cols = viewport.cols - const rows = viewport.rows stream.registeredRemoteDesktopDriver = true - stream.desktopClaimTail = stream.desktopClaimTail - .then( - () => - runtime.updateRemoteDesktopViewer( - stream.ptyId, - stream.remoteDesktopSubscriptionKey, - stream.client!.id, - cols, - rows, - true - ), - () => - runtime.updateRemoteDesktopViewer( - stream.ptyId, - stream.remoteDesktopSubscriptionKey, - stream.client!.id, - cols, - rows, - true - ) + const claim = () => + runtime.updateRemoteDesktopViewer( + stream.ptyId, + stream.remoteDesktopSubscriptionKey, + stream.client!.id, + viewport.cols, + viewport.rows, + true ) - .catch(() => false) + stream.desktopClaimTail = stream.desktopClaimTail.then(claim, claim).catch(() => false) return } if (frame.opcode === TerminalStreamOpcode.SnapshotRequest) { @@ -156,7 +151,7 @@ export function installMultiplexSlotFrames( stream: TerminalMultiplexStream, request: MultiplexSnapshotRequest ): Promise => { - if (state.closed || streams.get(stream.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return } stream.outputBatcher.flush() @@ -171,7 +166,7 @@ export function installMultiplexSlotFrames( stream.ptyId, scrollbackRows ) - if (state.closed || streams.get(stream.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return } let size = runtime.getTerminalSize(stream.ptyId) @@ -182,7 +177,7 @@ export function installMultiplexSlotFrames( stream.pendingOutputBytes = 0 stream.pendingOutputOverflowed = false serialized = await serializeBudgetedRequestedSnapshot(runtime, stream.ptyId, scrollbackRows) - if (state.closed || streams.get(stream.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return } size = runtime.getTerminalSize(stream.ptyId) @@ -264,16 +259,7 @@ export function installMultiplexSlotFrames( ) { const viewport = stream.pendingRemoteDesktopViewport stream.pendingRemoteDesktopViewport = null - void updateViewportForClient( - runtime, - stream.ptyId, - stream.remoteDesktopSubscriptionKey, - stream.client, - viewport, - 'desktop', - 'register', - !stream.supportsDesktopViewportClaims - ).catch(() => {}) + void applyStreamViewport(stream, stream.client, viewport).catch(() => {}) } } } diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts index 3e1df26e0f6..d9e95c99fec 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts @@ -8,9 +8,10 @@ import { TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES } from '../../../../ import { createTerminalOutputBatcher } from './terminal-output-batcher' import { appendPendingMultiplexOutput } from './terminal-stream-replay' import { updateViewportForClient } from './terminal-viewport-update' -import type { - MultiplexSubscribeRequest, - TerminalMultiplexConnection +import { + isMultiplexStreamAttached, + type MultiplexSubscribeRequest, + type TerminalMultiplexConnection } from './terminal-multiplex-connection' import type { TerminalMultiplexStream } from './terminal-stream-types' @@ -101,7 +102,7 @@ export async function initializeMultiplexStream( onInstalled(stream) const unsubscribeStreamData = runtime.subscribeToTerminalData(ptyId, (data, meta) => { - if (state.closed || streams.get(request.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return } if (stream.outputPaused) { @@ -146,7 +147,7 @@ export async function initializeMultiplexStream( !stream.supportsDesktopViewportClaims ) } - if (state.closed || streams.get(request.streamId) !== stream) { + if (!isMultiplexStreamAttached(state, stream)) { return null } return stream diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-subscribe-frame.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-subscribe-frame.ts index 6a902c59ade..3800fa08a05 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-subscribe-frame.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-subscribe-frame.ts @@ -1,19 +1,13 @@ import { decodeTerminalStreamJson } from '../../../../../shared/terminal-stream-protocol' import { TerminalMultiplexSubscribeFrame } from './stream-schemas' -import type { - TerminalMultiplexConnection, - TerminalMultiplexSlotFramesStage -} from './terminal-multiplex-connection' +import type { TerminalMultiplexConnection } from './terminal-multiplex-connection' import { resolveMultiplexSubscribePty } from './terminal-multiplex-subscribe-resolution' import { initializeMultiplexStream } from './terminal-multiplex-stream-initialization' import { publishMultiplexInitialSnapshot } from './terminal-multiplex-initial-snapshot' import { activateMultiplexStream } from './terminal-multiplex-live-stream' import type { TerminalMultiplexStream } from './terminal-stream-types' -export function installMultiplexSubscribeFrame( - build: TerminalMultiplexSlotFramesStage -): asserts build is TerminalMultiplexConnection { - const state = build as TerminalMultiplexConnection +export function installMultiplexSubscribeFrame(state: TerminalMultiplexConnection): void { state.handleSubscribeFrame = async (payload) => { const raw = decodeTerminalStreamJson(payload) const parsed = TerminalMultiplexSubscribeFrame.safeParse(raw) diff --git a/src/main/runtime/rpc/methods/terminal/terminal-query-methods.ts b/src/main/runtime/rpc/methods/terminal/terminal-query-methods.ts index 20fe9633b1b..152c0d61a49 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-query-methods.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-query-methods.ts @@ -8,7 +8,7 @@ import { TerminalRename, TerminalResolveActive, TerminalResolvePane -} from './unary-schemas' +} from '../../../../../shared/rpc-contract/terminal-unary-params' export const TERMINAL_QUERY_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/terminal/terminal-send-method.ts b/src/main/runtime/rpc/methods/terminal/terminal-send-method.ts index 7880713b131..f7ae8b29e5a 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-send-method.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-send-method.ts @@ -2,7 +2,7 @@ import { assertLegacyAiVaultResumeCommandAllowed } from '../../../../ai-vault/st import { InvalidArgumentError, defineMethod } from '../../core' import { isTerminalQueryReply } from '../../../../../shared/terminal-query-reply' import { assertTerminalAgentSendable } from '../../terminal-agent-send-guard' -import { TerminalSend } from './unary-schemas' +import { TerminalSend } from '../../../../../shared/rpc-contract/terminal-unary-params' import { assertTerminalSendExactPtyBinding, assertTerminalSendTextWithinLimit, diff --git a/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts b/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts index 5ffd1079a3c..8caca9d9132 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts @@ -97,37 +97,36 @@ export function appendAckPendingOutput( ): void { stream.ackPendingOutput.push(chunk) stream.ackPendingOutputBytes += chunk.bytes.byteLength - let omittedChunkCount = 0 - while ( - stream.ackPendingOutputBytes > TERMINAL_MULTIPLEX_PENDING_MAX_BYTES && - omittedChunkCount < stream.ackPendingOutput.length - ) { - stream.ackPendingOutputBytes -= stream.ackPendingOutput[omittedChunkCount]!.bytes.byteLength - omittedChunkCount += 1 - } - if (omittedChunkCount > 0) { - stream.ackPendingOutput.splice(0, omittedChunkCount) - stream.ackPendingOutputOverflowed = true - } + const trimmed = trimHeadToBudget( + stream.ackPendingOutput, + stream.ackPendingOutputBytes, + (pending) => pending.bytes.byteLength + ) + stream.ackPendingOutputBytes = trimmed.bytes + stream.ackPendingOutputOverflowed ||= trimmed.overflowed } export function trimPendingOutputToBudget( pendingOutput: TerminalOutputChunk[], pendingOutputBytes: number ): { bytes: number; overflowed: boolean } { - let omittedChunkCount = 0 - while ( - pendingOutputBytes > TERMINAL_MULTIPLEX_PENDING_MAX_BYTES && - omittedChunkCount < pendingOutput.length - ) { - const chunk = pendingOutput[omittedChunkCount] - pendingOutputBytes -= chunk.bytes - omittedChunkCount += 1 + return trimHeadToBudget(pendingOutput, pendingOutputBytes, (chunk) => chunk.bytes) +} + +function trimHeadToBudget( + items: T[], + bytes: number, + sizeOf: (item: T) => number +): { bytes: number; overflowed: boolean } { + let omittedCount = 0 + while (bytes > TERMINAL_MULTIPLEX_PENDING_MAX_BYTES && omittedCount < items.length) { + bytes -= sizeOf(items[omittedCount]!) + omittedCount += 1 } - if (omittedChunkCount > 0) { - pendingOutput.splice(0, omittedChunkCount) + if (omittedCount > 0) { + items.splice(0, omittedCount) } - return { bytes: pendingOutputBytes, overflowed: omittedChunkCount > 0 } + return { bytes, overflowed: omittedCount > 0 } } export function trimPendingOutputCoveredBySnapshot( diff --git a/src/main/runtime/rpc/methods/terminal/terminal-subscribe-method.ts b/src/main/runtime/rpc/methods/terminal/terminal-subscribe-method.ts index 1bd7cd268c7..0b1ae4e54b4 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-subscribe-method.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-subscribe-method.ts @@ -1,5 +1,5 @@ import { defineStreamingMethod } from '../../core' -import { TerminalSubscribe } from './stream-schemas' +import { TerminalSubscribe } from '../../../../../shared/rpc-contract/terminal-stream-params' import { isTerminalReadPayloadIncomplete } from './terminal-stream-replay' import { runTerminalBinarySubscription } from './terminal-legacy-subscribe-binary' import { diff --git a/src/main/runtime/rpc/methods/terminal/terminal-viewer-colors-method.ts b/src/main/runtime/rpc/methods/terminal/terminal-viewer-colors-method.ts index 095bc694bfa..598a0ea1b72 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-viewer-colors-method.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-viewer-colors-method.ts @@ -1,7 +1,7 @@ import { defineMethod } from '../../core' import { normalizeColorQueryReplyColors } from '../../../../../shared/pty-owner-color-query-colors' import { setPairedViewerColors } from '../../../terminal-view-attribute-store' -import { TerminalSetViewerColors } from './unary-schemas' +import { TerminalSetViewerColors } from '../../../../../shared/rpc-contract/terminal-unary-params' export const TERMINAL_VIEWER_COLORS_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/terminal/terminal-viewport-methods.ts b/src/main/runtime/rpc/methods/terminal/terminal-viewport-methods.ts index f24dea4602a..4fd5f62cf5d 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-viewport-methods.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-viewport-methods.ts @@ -1,11 +1,11 @@ import { defineMethod } from '../../core' -import { TerminalHandle } from './unary-schemas' +import { TerminalHandle } from '../../../../../shared/rpc-contract/terminal-unary-params' import { TerminalSetAutoRestoreFit, TerminalSetDisplayMode, TerminalUnsubscribe, TerminalUpdateViewport -} from './viewport-schemas' +} from '../../../../../shared/rpc-contract/terminal-viewport-schemas-params' import { updateViewportForClient } from './terminal-viewport-update' import { TerminalGetAutoRestoreFitParams } from '../../../../../shared/rpc-contract/terminal-viewport-methods-params' diff --git a/src/main/runtime/rpc/methods/terminal/unary-schemas.ts b/src/main/runtime/rpc/methods/terminal/unary-schemas.ts deleted file mode 100644 index 65787c8c405..00000000000 --- a/src/main/runtime/rpc/methods/terminal/unary-schemas.ts +++ /dev/null @@ -1,23 +0,0 @@ -export { - AgentTeamsPrepareLaunch, - AgentTeamsTmuxCompat, - TerminalCloseAll, - TerminalCreateParams, - TerminalFocus, - TerminalHandle, - TerminalInspectProcess, - TerminalListParams, - TerminalRead, - TerminalRecoverPane, - TerminalRename, - TerminalResolveActive, - TerminalResolvePane, - TerminalSend, - TerminalSetViewerColors, - TerminalSleep, - TerminalSplit, - TerminalStop, - TerminalStopExact, - TerminalViewport, - TerminalWait -} from '../../../../../shared/rpc-contract/terminal-unary-params' diff --git a/src/main/runtime/rpc/methods/terminal/viewport-schemas.ts b/src/main/runtime/rpc/methods/terminal/viewport-schemas.ts deleted file mode 100644 index 70ecd9037d1..00000000000 --- a/src/main/runtime/rpc/methods/terminal/viewport-schemas.ts +++ /dev/null @@ -1,6 +0,0 @@ -export { - TerminalSetAutoRestoreFit, - TerminalSetDisplayMode, - TerminalUnsubscribe, - TerminalUpdateViewport -} from '../../../../../shared/rpc-contract/terminal-viewport-schemas-params' diff --git a/src/main/runtime/rpc/methods/ui-update-value-tolerance.ts b/src/main/runtime/rpc/methods/ui-update-value-tolerance.ts deleted file mode 100644 index e03a9bfad06..00000000000 --- a/src/main/runtime/rpc/methods/ui-update-value-tolerance.ts +++ /dev/null @@ -1,4 +0,0 @@ -export { - omitUndefinedValues, - tolerateUnknownValues -} from '../../../../shared/rpc-contract/ui-update-value-tolerance-params' diff --git a/src/main/runtime/rpc/methods/worktree-catalog-methods.ts b/src/main/runtime/rpc/methods/worktree-catalog-methods.ts index dc0a01cfe60..29e61cfc64f 100644 --- a/src/main/runtime/rpc/methods/worktree-catalog-methods.ts +++ b/src/main/runtime/rpc/methods/worktree-catalog-methods.ts @@ -10,7 +10,7 @@ import { WorktreeDetectedListParams, WorktreeListParams, WorktreePsParams -} from './worktree-schemas' +} from '../../../../shared/rpc-contract/worktree-params' export const WORKTREE_CATALOG_METHODS = [ defineMethod({ diff --git a/src/main/runtime/rpc/methods/worktree-create-args.test.ts b/src/main/runtime/rpc/methods/worktree-create-args.test.ts index 49fab597781..19747628dae 100644 --- a/src/main/runtime/rpc/methods/worktree-create-args.test.ts +++ b/src/main/runtime/rpc/methods/worktree-create-args.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' import { buildManagedWorktreeCreateArgs } from './worktree-create-args' -import { WorktreeCreate } from './worktree-create-schemas' +import { WorktreeCreate } from '../../../../shared/rpc-contract/worktree-create-params' const PROVENANCE = { automationProvenance: undefined, diff --git a/src/main/runtime/rpc/methods/worktree-create-args.ts b/src/main/runtime/rpc/methods/worktree-create-args.ts index 54336dcbd68..173fa8ea7cd 100644 --- a/src/main/runtime/rpc/methods/worktree-create-args.ts +++ b/src/main/runtime/rpc/methods/worktree-create-args.ts @@ -1,7 +1,7 @@ import type { z } from 'zod' import { resolveRuntimeNavigationTarget } from '../../../../shared/runtime-navigation' import type { OrcaRuntimeService } from '../../orca-runtime' -import type { WorktreeCreate } from './worktree-create-schemas' +import type { WorktreeCreate } from '../../../../shared/rpc-contract/worktree-create-params' type WorktreeCreateParams = z.infer type ManagedWorktreeCreateArgs = Parameters[0] diff --git a/src/main/runtime/rpc/methods/worktree-create-schemas.ts b/src/main/runtime/rpc/methods/worktree-create-schemas.ts deleted file mode 100644 index 659f0c87fcf..00000000000 --- a/src/main/runtime/rpc/methods/worktree-create-schemas.ts +++ /dev/null @@ -1,4 +0,0 @@ -export { - WorktreeCreate, - WorktreePrefetchCreateBase -} from '../../../../shared/rpc-contract/worktree-create-params' diff --git a/src/main/runtime/rpc/methods/worktree-schemas.test.ts b/src/main/runtime/rpc/methods/worktree-schemas.test.ts index 8172a3cd6b0..52e04554888 100644 --- a/src/main/runtime/rpc/methods/worktree-schemas.test.ts +++ b/src/main/runtime/rpc/methods/worktree-schemas.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' -import { WorktreeCreate } from './worktree-create-schemas' -import { WorktreeActivate, WorktreeSet } from './worktree-schemas' +import { WorktreeCreate } from '../../../../shared/rpc-contract/worktree-create-params' +import { WorktreeActivate, WorktreeSet } from '../../../../shared/rpc-contract/worktree-params' describe('worktree RPC schemas', () => { it('accepts optional display-name provenance values', () => { diff --git a/src/main/runtime/rpc/methods/worktree-schemas.ts b/src/main/runtime/rpc/methods/worktree-schemas.ts deleted file mode 100644 index 41d2c38fec7..00000000000 --- a/src/main/runtime/rpc/methods/worktree-schemas.ts +++ /dev/null @@ -1,18 +0,0 @@ -export { - AutomationWorkspaceProvenanceRequest, - CliWorkspaceProvenanceRequest, - OptionalTuiAgent, - WorktreeActivate, - WorktreeDetectedListParams, - WorktreeForceDeleteBranch, - WorktreeListParams, - WorktreePsParams, - WorktreeRemove, - WorktreeResolveMrBase, - WorktreeResolvePrBase, - WorktreeSelector, - WorktreeSet, - WorktreeSortOrder, - WorktreeTeardownMissingTerminalsParams, - assertLinkedWorkItemSourceContextMatch -} from '../../../../shared/rpc-contract/worktree-params' diff --git a/src/main/runtime/rpc/methods/worktree-visibility-defaults-schema.ts b/src/main/runtime/rpc/methods/worktree-visibility-defaults-schema.ts deleted file mode 100644 index 19a6dba90be..00000000000 --- a/src/main/runtime/rpc/methods/worktree-visibility-defaults-schema.ts +++ /dev/null @@ -1 +0,0 @@ -export { WorktreeVisibilityDefaultsUpdate } from '../../../../shared/rpc-contract/worktree-visibility-defaults-params' diff --git a/src/main/runtime/rpc/methods/worktree.ts b/src/main/runtime/rpc/methods/worktree.ts index dabd845316c..9d044842673 100644 --- a/src/main/runtime/rpc/methods/worktree.ts +++ b/src/main/runtime/rpc/methods/worktree.ts @@ -14,7 +14,10 @@ import { buildManagedWorktreeCreateArgs } from './worktree-create-args' import { resolvePairedCallerHostId } from './paired-caller-host-id' import { resolveRuntimeNavigationTarget } from '../../../../shared/runtime-navigation' import { resolveRpcWorkspaceCreatorProvenance } from '../workspace-creator-context' -import { WorktreeCreate, WorktreePrefetchCreateBase } from './worktree-create-schemas' +import { + WorktreeCreate, + WorktreePrefetchCreateBase +} from '../../../../shared/rpc-contract/worktree-create-params' import { WorktreeActivate, WorktreeForceDeleteBranch, @@ -25,7 +28,7 @@ import { WorktreeSet, WorktreeSortOrder, WorktreeTeardownMissingTerminalsParams -} from './worktree-schemas' +} from '../../../../shared/rpc-contract/worktree-params' import { WORKTREE_CATALOG_METHODS } from './worktree-catalog-methods' import { readsWorktreeRemovalMarker } from '../worktree-removal-marker-projection' diff --git a/src/main/runtime/rpc/rpc-request-admission-prologue.ts b/src/main/runtime/rpc/rpc-request-admission-prologue.ts new file mode 100644 index 00000000000..90369e56b5a --- /dev/null +++ b/src/main/runtime/rpc/rpc-request-admission-prologue.ts @@ -0,0 +1,89 @@ +import type { + RpcEnvelopeMeta, + RpcMethod, + RpcRegistry, + RpcRequest, + RpcResponse, + RpcStreamingMethod +} from './core' +import { errorResponse } from './errors' +import type { OrcaRuntimeService } from '../orca-runtime' +import { orchestrationMigrationFence } from './orchestration-contract-fence' +import type { RpcDispatchStreamingOptions } from './dispatcher-stream-options' +import { mapDispatcherError } from './dispatcher-error-response' +import { parseRpcRequestParams } from './dispatcher-request-parsing' +import type { SshBridgeCallBinding } from './ssh-bridge-host-binding' +import { + bindRpcCallToCallerScope, + denyRpcMethodForCaller, + OWNER_RPC_CALLER_SCOPE, + type RpcCallerScope +} from './rpc-caller-scope' +import { + needsOrchestrationCallerResolution, + resolveOrchestrationSessionCaller, + type ResolvedOrchestrationRequest +} from './orchestration-session-caller' + +type RpcRequestAdmissionDeps = { + runtime: OrcaRuntimeService + registry: RpcRegistry + meta: RpcEnvelopeMeta + pinnedCallerScope?: RpcCallerScope +} + +// Why: both transports must reject, resolve and bind a request in the same order before any handler runs. +export function admitRpcRequest( + { runtime, registry, meta, pinnedCallerScope }: RpcRequestAdmissionDeps, + rawRequest: RpcRequest, + options: RpcDispatchStreamingOptions | undefined, + isTransportSupported: (method: RpcMethod | RpcStreamingMethod) => method is M +) { + const reject = (rejected: RpcResponse) => ({ rejected }) + const method = registry.get(rawRequest.method) + const callerScope = pinnedCallerScope ?? options?.callerScope ?? OWNER_RPC_CALLER_SCOPE + const denial = denyRpcMethodForCaller(callerScope, rawRequest.method, method?.permission) + if (denial) { + return reject(errorResponse(rawRequest.id, meta, 'forbidden', denial)) + } + if (!method) { + return reject( + errorResponse(rawRequest.id, meta, 'method_not_found', `Unknown method: ${rawRequest.method}`) + ) + } + const migrationFence = orchestrationMigrationFence(rawRequest, meta) + if (migrationFence) { + return reject(migrationFence) + } + + const admitResolved = ({ request, caller }: ResolvedOrchestrationRequest) => { + const parsedParams = parseRpcRequestParams(request, method, meta) + if (parsedParams.error) { + return reject(parsedParams.error) + } + if (!isTransportSupported(method)) { + return reject( + errorResponse( + request.id, + meta, + 'method_not_supported', + `Method ${request.method} requires a streaming transport` + ) + ) + } + const params = parsedParams.value + const admit = (binding: SshBridgeCallBinding | null) => + binding?.kind === 'denied' + ? reject(errorResponse(request.id, meta, 'forbidden', binding.message)) + : { method, request, caller, params, binding } + // Why: an unbound call stays synchronous, so dispatch gains no extra tick before the handler. + const pendingBinding = bindRpcCallToCallerScope(callerScope, runtime, request.method, params) + return pendingBinding ? pendingBinding.then(admit) : admit(null) + } + // Why: before params parse and the unary/streaming split, so both branches see one caller. + return needsOrchestrationCallerResolution(rawRequest) + ? resolveOrchestrationSessionCaller(runtime, rawRequest, options).then(admitResolved, (error) => + reject(mapDispatcherError(rawRequest, meta, error)) + ) + : admitResolved({ request: rawRequest }) +} diff --git a/src/main/runtime/rpc/rpc-streaming-dispatcher.ts b/src/main/runtime/rpc/rpc-streaming-dispatcher.ts index 14f960439e6..9a145ceed0f 100644 --- a/src/main/runtime/rpc/rpc-streaming-dispatcher.ts +++ b/src/main/runtime/rpc/rpc-streaming-dispatcher.ts @@ -1,38 +1,26 @@ import { - isRegistrationFencedUnsubscribe, isStreamingMethod, type RpcEnvelopeMeta, + type RpcMethod, type RpcRegistry, - type RpcRequest + type RpcRequest, + type RpcStreamingMethod } from './core' import { errorResponse, successResponse } from './errors' import type { OrcaRuntimeService } from '../orca-runtime' -import type { - OrchestrationMutationExecutor, - DurableMutationInvocation -} from './orchestration-mutation-executor' -import { orchestrationMigrationFence } from './orchestration-contract-fence' +import type { OrchestrationMutationExecutor } from './orchestration-mutation-executor' import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction' import type { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility' -import type { RpcDispatchStreamingOptions } from './dispatcher-stream-options' +import { + rpcContextFromTransport, + type RpcDispatchStreamingOptions +} from './dispatcher-stream-options' import { mapDispatcherError } from './dispatcher-error-response' -import { parseRpcRequestParams } from './dispatcher-request-parsing' -import { routeDispatcherClientHostedBrowserRpc } from './dispatcher-client-browser-routing' -import { needsLocalCallerFingerprint } from './dispatcher-caller-fingerprint' import { createDispatcherStreamingFeatureEmitter } from './dispatcher-streaming-feature-emitter' -import { resolveRpcCallerIdentity } from './rpc-caller-identity' -import { - bindRpcCallToCallerScope, - denyRpcMethodForCaller, - OWNER_RPC_CALLER_SCOPE, - type RpcCallerScope -} from './rpc-caller-scope' -import { - needsOrchestrationCallerResolution, - resolveOrchestrationSessionCaller, - type ResolvedOrchestrationRequest -} from './orchestration-session-caller' +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 @@ -52,145 +40,43 @@ export class RpcStreamingDispatcher { reply: (response: string) => void, options?: RpcDispatchStreamingOptions ): Promise { - let request = rawRequest const { runtime, registry, orchestrationMutations, legacyOrchestration, meta } = this.dependencies const envelopeMeta = meta() - const method = registry.get(request.method) - const callerScope = - this.dependencies.pinnedCallerScope ?? options?.callerScope ?? OWNER_RPC_CALLER_SCOPE - const denial = denyRpcMethodForCaller(callerScope, request.method, method?.permission) - if (denial) { - reply(JSON.stringify(errorResponse(request.id, envelopeMeta, 'forbidden', denial))) - return - } - if (!method) { - reply( - JSON.stringify( - errorResponse( - request.id, - envelopeMeta, - 'method_not_found', - `Unknown method: ${request.method}` - ) - ) - ) - return - } - - const migrationFence = orchestrationMigrationFence(request, envelopeMeta) - if (migrationFence) { - reply(JSON.stringify(migrationFence)) - return - } - - // Why: before params parse and the unary/streaming split, so both branches see one caller. - let resolved: ResolvedOrchestrationRequest = { request } - if (needsOrchestrationCallerResolution(request)) { - try { - resolved = await resolveOrchestrationSessionCaller(runtime, request, options) - } catch (error) { - reply(JSON.stringify(mapDispatcherError(request, envelopeMeta, error))) - return - } - } - request = resolved.request - const orchestrationCaller = resolved.caller - const parsedParams = parseRpcRequestParams(request, method, envelopeMeta) - if (parsedParams.error) { - reply(JSON.stringify(parsedParams.error)) - return - } - const params = parsedParams.value - const pendingBinding = bindRpcCallToCallerScope(callerScope, runtime, request.method, params) - const binding = pendingBinding ? await pendingBinding : null - if (binding?.kind === 'denied') { - reply(JSON.stringify(errorResponse(request.id, envelopeMeta, 'forbidden', binding.message))) + 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, binding, caller } = admission if (!isStreamingMethod(method)) { try { - // Session tabs always need this fence. COMPAT(terminal request-addressed unsubscribe): terminal only for phones without `requestId`. - // Capture before middleware yields to a replacement subscribe on the same connection. - const subscriptionRegistrationVersion = isRegistrationFencedUnsubscribe(request.method) - ? runtime.getSubscriptionRegistrationVersion() - : undefined - const clientHostedBrowser = await routeDispatcherClientHostedBrowserRpc( + const result = await invokeDispatcherUnaryMethod({ runtime, - request.method, - params - ) - if (clientHostedBrowser.handled) { - recordRuntimeFeatureInteraction( - runtime, - request.method, - clientHostedBrowser.result, - undefined, - request.params - ) - reply( - JSON.stringify(successResponse(request.id, envelopeMeta, clientHostedBrowser.result)) - ) - return - } - const compatibility = await legacyOrchestration.tryHandle(request, params, options?.signal) - if (compatibility.handled) { - reply(JSON.stringify(successResponse(request.id, envelopeMeta, compatibility.result))) - return - } - const effectiveParams = compatibility.params ?? params - const legacyCoordinator = legacyOrchestration.createCoordinatorInvocation( request, - compatibility.legacyCoordinatorAuthority - ) - const authenticatedCallerFingerprint = - options?.authenticatedCallerFingerprint ?? - (needsLocalCallerFingerprint(request, effectiveParams) - ? orchestrationMutations.getLocalAuthenticatedCallerFingerprint() - : undefined) - const invoke = (mutation?: DurableMutationInvocation) => { - const legacyCoordinatorRunId = legacyCoordinator?.revalidate() - return method.handler(effectiveParams, { - runtime, - signal: options?.signal, - requestId: request.id, - connectionId: options?.connectionId, - subscriptionRegistrationVersion, - clientId: options?.clientId, - pairedDeviceId: options?.pairedDeviceId, - caller: resolveRpcCallerIdentity(options), - clientKind: options?.clientKind, - clientCapabilities: options?.clientCapabilities, - updateClientCapabilities: options?.updateClientCapabilities, - authenticatedCallerFingerprint: - mutation?.identity.callerFingerprint ?? - legacyCoordinator?.mutationCallerFingerprint ?? - authenticatedCallerFingerprint, - recordMutationReceipt: mutation?.recordReceipt, - orchestrationMutation: mutation?.identity, - pairing: options?.pairing, - sendBinary: options?.sendBinary, - registerBinaryStreamHandler: options?.registerBinaryStreamHandler, - registerBinaryMessageHandler: options?.registerBinaryMessageHandler, - legacyCoordinatorRunId, - legacyCoordinatorAuthority: legacyCoordinator?.authority, - revalidateLegacyCoordinator: legacyCoordinator?.revalidate, - orchestrationCompatibilityCallerAuthority: - compatibility.orchestrationCompatibilityCallerAuthority, - orchestrationCompatibilityEvidence: request.orchestrationCompatibilityEvidence, - orchestrationCaller - }) + method, + params, + context: rpcContextFromTransport(runtime, request, options, caller), + orchestrationMutations, + legacyOrchestration + }) + const filtered = binding?.filterResult?.(result) ?? { + kind: 'allowed', + result } - const result = await orchestrationMutations.run( - request, - effectiveParams, - invoke, - legacyCoordinator?.mutationCallerFingerprint ?? authenticatedCallerFingerprint, - orchestrationCaller?.orcaSessionId - ) - recordRuntimeFeatureInteraction(runtime, request.method, result, undefined, request.params) - const filtered = binding?.filterResult?.(result) ?? { kind: 'allowed', result } reply( JSON.stringify( filtered.kind === 'denied' @@ -214,23 +100,7 @@ export class RpcStreamingDispatcher { try { const result = await method.handler( params, - { - runtime, - signal: options?.signal, - requestId: request.id, - connectionId: options?.connectionId, - clientId: options?.clientId, - pairedDeviceId: options?.pairedDeviceId, - caller: resolveRpcCallerIdentity(options), - clientKind: options?.clientKind, - clientCapabilities: options?.clientCapabilities, - updateClientCapabilities: options?.updateClientCapabilities, - pairing: options?.pairing, - sendBinary: options?.sendBinary, - registerBinaryStreamHandler: options?.registerBinaryStreamHandler, - registerBinaryMessageHandler: options?.registerBinaryMessageHandler, - orchestrationCaller - }, + rpcContextFromTransport(runtime, request, options, caller), emit ) recordRuntimeFeatureInteraction( diff --git a/src/main/runtime/rpc/schemas.test.ts b/src/main/runtime/rpc/schemas.test.ts index 8ba47496545..8de0752434f 100644 --- a/src/main/runtime/rpc/schemas.test.ts +++ b/src/main/runtime/rpc/schemas.test.ts @@ -7,7 +7,7 @@ import { OptionalPlainString, OptionalPositiveInt, OptionalString -} from './schemas' +} from '../../../shared/rpc-contract/rpc-param-primitives' import { InterceptEnable, Screenshot, diff --git a/src/main/runtime/rpc/schemas.ts b/src/main/runtime/rpc/schemas.ts deleted file mode 100644 index 2c469341be5..00000000000 --- a/src/main/runtime/rpc/schemas.ts +++ /dev/null @@ -1,17 +0,0 @@ -// Why: the RPC boundary ingests loosely-typed JSON from a CLI that grew -// organically, so these reusable pieces capture the validation shapes that -// recur across domains (optional worktree selector, bounded limit, browser -// target envelope, etc.). Methods compose these to declare their real -// contract without repeating the same `typeof` gymnastics 90 times. -export { - BrowserTarget, - OptionalBoolean, - OptionalFiniteNumber, - OptionalPlainString, - OptionalPositiveInt, - OptionalString, - TriStateLinkedIssue, - requiredNumber, - requiredString, - requiredStringAllowingEmpty -} from '../../../shared/rpc-contract/rpc-param-primitives' diff --git a/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts index f56516f3f27..4010d6e3ba6 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts @@ -61,7 +61,7 @@ export abstract class RemoteRuntimeTerminalFlowController extends RemoteRuntimeT stream: RemoteRuntimeMultiplexedTerminalState, paused: boolean ): boolean { - if (!stream.supportsOutputPause || this.streams.get(stream.streamId) !== stream) { + if (!stream.supportsOutputPause || !this.isRegisteredStream(stream)) { return false } if (stream.outputPaused === paused) { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts index 2f8e4ccae06..a8877a32e7d 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts @@ -9,12 +9,7 @@ import { recordE2eRemoteTransportSubscribe, unsubscribeRuntimeEnvironmentForE2e } from './remote-runtime-terminal-e2e-control' -import { - clearResyncTimer, - clearSnapshot, - discardOutputAcknowledgements, - rejectPendingSnapshotRequest -} from './remote-runtime-terminal-snapshot-state' +import { disposeRemoteTerminalStreamState } from './remote-runtime-terminal-snapshot-state' import type { RemoteRuntimeMultiplexedTerminalState, RuntimeEnvironmentSubscriptionHandle @@ -193,11 +188,7 @@ export abstract class RemoteRuntimeTerminalMultiplexerBase { // Why: close callbacks may resubscribe synchronously; release first so every replacement shares the new environment multiplexer. this.releaseIfCurrent(this.environmentId, this) for (const stream of streams) { - discardOutputAcknowledgements(stream) - stream.watchdog.dispose() - clearSnapshot(stream) - clearResyncTimer(stream) - rejectPendingSnapshotRequest(stream, message ?? 'Remote runtime connection closed.') + disposeRemoteTerminalStreamState(stream, message ?? 'Remote runtime connection closed.') const canHandleClose = Boolean(stream.callbacks.onTransportClose) stream.callbacks.onTransportClose?.({ recoverable }) if (message && !canHandleClose) { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts index 01dcac4d9b9..6e37cf12f8c 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts @@ -7,9 +7,7 @@ import { recordRendererCrashBreadcrumb } from '@/lib/crash-breadcrumb-recorder' import { RemoteRuntimeTerminalBinaryController } from './remote-runtime-terminal-binary-controller' import { CONTROL_STREAM_ID, - clearResyncTimer, - discardOutputAcknowledgements, - rejectPendingSnapshotRequest + disposeRemoteTerminalStreamState } from './remote-runtime-terminal-snapshot-state' import type { RemoteRuntimeMultiplexedTerminal, @@ -121,12 +119,10 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar serializeBuffer: (opts) => this.requestSnapshot(state, opts), serializeBufferOutcome: (opts) => this.requestSnapshotOutcome(state, opts), close: () => { - if (this.streams.get(streamId) === state) { - discardOutputAcknowledgements(state) - state.watchdog.dispose() + if (this.isRegisteredStream(state)) { + // Why after: a failed write closes the transport, whose message the pending snapshot must carry. this.sendFrame(streamId, TerminalStreamOpcode.Unsubscribe) - clearResyncTimer(state) - rejectPendingSnapshotRequest(state, 'Remote terminal stream closed.') + disposeRemoteTerminalStreamState(state, 'Remote terminal stream closed.') this.streams.delete(streamId) this.closeIfIdle() } diff --git a/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts index 5ee01d0cabb..a51b6ca9ae4 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts @@ -6,7 +6,7 @@ import { RemoteRuntimeTerminalFlowController } from './remote-runtime-terminal-f import { clearResyncTimer, clearSnapshot, - discardOutputAcknowledgements, + disposeRemoteTerminalStreamState, isTerminalDriverState, rejectPendingSnapshotRequest } from './remote-runtime-terminal-snapshot-state' @@ -87,11 +87,7 @@ export abstract class RemoteRuntimeTerminalResponseController extends RemoteRunt stream.callbacks.onOutputPauseCapability?.() } } else if (event.type === 'end') { - discardOutputAcknowledgements(stream) - stream.watchdog.dispose() - clearSnapshot(stream) - clearResyncTimer(stream) - rejectPendingSnapshotRequest(stream, 'Remote terminal stream ended.') + disposeRemoteTerminalStreamState(stream, 'Remote terminal stream ended.') this.streams.delete(event.streamId) if (stream.capacityRejected) { if (stream.callbacks.onTransportClose) { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts index c3cdc0c2190..7ec782cfaaa 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts @@ -12,9 +12,7 @@ import { REMOTE_TERMINAL_SNAPSHOT_REQUEST_TIMEOUT_MS, clearPendingSnapshotRequest, clearResyncTimer, - clearSnapshot, - discardOutputAcknowledgements, - rejectPendingSnapshotRequest, + disposeRemoteTerminalStreamState, retryWorthySnapshotOutcome } from './remote-runtime-terminal-snapshot-state' import type { @@ -254,11 +252,7 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt if (this.streams.get(stream.streamId) !== stream) { return } - stream.watchdog.dispose() - discardOutputAcknowledgements(stream) - clearSnapshot(stream) - clearResyncTimer(stream) - rejectPendingSnapshotRequest(stream, 'Remote terminal stream stopped responding.') + disposeRemoteTerminalStreamState(stream, 'Remote terminal stream stopped responding.') this.streams.delete(stream.streamId) this.sendFrame(stream.streamId, TerminalStreamOpcode.Unsubscribe) if (stream.callbacks.onTransportClose) { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts index 856413f9c54..852d2e57b78 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts @@ -82,6 +82,17 @@ export function rejectPendingSnapshotRequest( request.reject(new Error(message)) } +export function disposeRemoteTerminalStreamState( + stream: RemoteRuntimeMultiplexedTerminalState, + message: string +): void { + discardOutputAcknowledgements(stream) + stream.watchdog.dispose() + clearSnapshot(stream) + clearResyncTimer(stream) + rejectPendingSnapshotRequest(stream, message) +} + export function decodeSnapshotInfo( payload: Uint8Array ): RemoteRuntimeSnapshotInfo | null { diff --git a/tests/e2e/cross-version-wire/cross-version-browser-placement.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-browser-placement.unit.test.ts index 7a55bab6d50..ad72fd6e228 100644 --- a/tests/e2e/cross-version-wire/cross-version-browser-placement.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-browser-placement.unit.test.ts @@ -3,7 +3,7 @@ import { BROWSER_CLIENT_HOST_RUNTIME_CAPABILITY, BROWSER_NETWORK_TUNNEL_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version' -import { BrowserTabCreateParams } from '../../../src/main/runtime/rpc/methods/browser-tab-create-schema' +import { BrowserTabCreateParams } from '../../../src/shared/rpc-contract/browser-tab-create-params' import { importReleaseCheckoutModule, materializeReleaseCheckout,