starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-30 16:47:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(runtime-rpc-terminal-io): clean up Phase 3-3.8 code (#26895)

Browse Source 
* refactor(runtime-rpc-terminal-io): clean up Phase 3-3.8 code

- One admission prologue and one unary invocation shared by both RPC dispatchers
- Delete rpc schema re-export shims; import from the shared params catalog
- Delete the unregistered duplicate markdown tab method module
- Collapse multiplex installer stage types; add isMultiplexStreamAttached
- One source-identity guard in the ACK recovery snapshot
- Single stream teardown in the renderer multiplexer
- De-duplicate RPC method declarations; derive dispatch options from RpcContext
- Share viewport-frame decoding and one head-trim loop for pending queues
- Single sendTerminal call in sendTerminalStreamInput

* test: point cross-version and scope checks at the moved browser tab-create schema; run transport parity test under Node

---------

Co-authored-by: m4air <m4air@Mac.localdomain>
Co-authored-by: m4air <m4air@m4airs-MacBook-Air.local>
This commit is contained in:
OrcaWin
2026-10-09 11:50:31 -07:00
committed by GitHub
co-authored by m4air m4air
parent fbcf0edc60
commit 1f8dce85d4
100 changed files with 755 additions and 881 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/scripts/pr-code-change-scope.mjs
+1 -1
View File
@@ -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',
config/scripts/pr-code-change-scope.test.mjs
+1 -1
View File
@@ -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',
config/scripts/vitest-node-runtime-files.mjs
+1
View File
@@ -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',
mobile/src/session/session-read-reply-schema.ts
+1 -1
View File
@@ -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.
/**
mobile/src/tasks/mobile-agent-launch-architecture.test.ts
+1 -1
View File
@@ -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
src/main/runtime/rpc/core.ts
+28 -62
View File
@@ -125,8 +125,6 @@ export type RpcContext = {
) => () => void
}
export type RpcHandler<TParams, TResult> = (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 | null> = TSchema extends ZodType
? TSchema['_output']
@@ -138,92 +136,60 @@ export type RpcTypedMethod<TName extends string, TSchema extends ZodType | null,
// Why: required so a new method cannot ship without deciding which callers may reach it.
readonly permission: RpcMethodPermission
readonly params: TSchema
readonly handler: RpcHandler<RpcParsedParams<TSchema>, TResult>
readonly handler: (params: RpcParsedParams<TSchema>, ctx: RpcContext) => TResult
}
export function defineMethod<TName extends string, TSchema extends ZodType | null, TResult>(
spec: RpcTypedMethod<TName, TSchema, TResult>
): RpcTypedMethod<TName, TSchema, TResult> {
return {
name: spec.name,
permission: spec.permission,
params: spec.params,
handler: spec.handler
}
return spec
}
export type RpcStreamingHandler<TParams> = (
params: TParams,
ctx: RpcContext,
emit: (result: unknown) => void
) => Promise<void>
// Why: emitted values stay `unknown` — the emit callback is an input, so there is no return position to infer them from.
export type RpcTypedStreamingMethod<TName extends string, TSchema extends ZodType | null> = {
readonly name: TName
readonly permission: RpcMethodPermission
readonly params: TSchema
readonly stream: true
readonly handler: RpcStreamingHandler<RpcParsedParams<TSchema>>
readonly handler: (
params: RpcParsedParams<TSchema>,
ctx: RpcContext,
emit: (result: unknown) => void
) => Promise<void>
}
export function defineStreamingMethod<TName extends string, TSchema extends ZodType | null>(
spec: Omit<RpcTypedStreamingMethod<TName, TSchema>, 'stream'>
): RpcTypedStreamingMethod<TName, TSchema> {
return {
name: spec.name,
permission: spec.permission,
params: spec.params,
stream: true,
handler: spec.handler
}
return { ...spec, stream: true }
}
type RpcMethodWithParams<P> = {
readonly name: string
readonly permission: RpcMethodPermission
readonly params: ZodType | null
readonly handler: (params: P, ctx: RpcContext) => unknown
}
type RpcStreamingMethodWithParams<P> = {
readonly name: string
readonly permission: RpcMethodPermission
readonly params: ZodType | null
readonly stream: true
readonly handler: (params: P, ctx: RpcContext, emit: (result: unknown) => void) => Promise<void>
}
// 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<void>
}
export type RpcMethodDeclaration = RpcMethodWithParams<never>
type RpcStreamingMethodDeclaration = RpcStreamingMethodWithParams<never>
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<unknown>
// 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<void>
}
export type RpcAnyMethod = RpcMethod | RpcStreamingMethod
export type RpcStreamingMethod = RpcStreamingMethodWithParams<unknown>
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[]
src/main/runtime/rpc/dispatcher-stream-options.ts
+51 -24
View File
@@ -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<ArrayBufferLike>) => boolean | void
registerBinaryStreamHandler?: (
streamId: number,
handler: (frame: TerminalStreamFrame) => void
) => () => void
registerBinaryMessageHandler?: (
handler: (bytes: Uint8Array<ArrayBufferLike>) => 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
}
}
src/main/runtime/rpc/dispatcher-transport-parity.test.ts
+197
View File
@@ -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<RpcDispatcher['dispatch']>[1]
): Promise<RpcResponse> {
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' } })
})
})
src/main/runtime/rpc/dispatcher-unary-method-invocation.ts
+7 -6
View File
@@ -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)
src/main/runtime/rpc/dispatcher.ts
+40 -92
View File
@@ -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<RpcResponse> {
async dispatch(request: RpcRequest, options?: RpcDispatchStreamingOptions): Promise<RpcResponse> {
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)
}
src/main/runtime/rpc/methods/agent-launch-folder-workspace-creation.ts
+1 -1
View File
@@ -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'],
src/main/runtime/rpc/methods/agent-launch-intent-resolution.ts
+1 -1
View File
@@ -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'
/**
src/main/runtime/rpc/methods/agent-launch-replay.ts
+1 -1
View File
@@ -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
src/main/runtime/rpc/methods/agent-launch-schemas.ts
-6
View File
@@ -1,6 +0,0 @@
export {
AgentLaunch,
AgentLaunchFields,
AgentLaunchReplay,
type AgentLaunchParams
} from '../../../../shared/rpc-contract/agent-launch-params'
src/main/runtime/rpc/methods/agent-launch-tab-publication.ts
+1 -1
View File
@@ -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
src/main/runtime/rpc/methods/agent-launch-worktree-creation.ts
+1 -1
View File
@@ -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'
src/main/runtime/rpc/methods/agent-launch.ts
+5 -1
View File
@@ -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,
src/main/runtime/rpc/methods/agent-skill-sharing-capability-grant.test.ts
+1 -1
View File
@@ -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 }
src/main/runtime/rpc/methods/artifact-sharing-capability-grant.test.ts
+1 -1
View File
@@ -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({
src/main/runtime/rpc/methods/automation-schemas.ts
-10
View File
@@ -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'
src/main/runtime/rpc/methods/automations.ts
+1 -1
View File
@@ -7,7 +7,7 @@ import {
AutomationList,
AutomationRuns,
AutomationUpdate
} from './automation-schemas'
} from '../../../../shared/rpc-contract/automation-params'
function mutationOwner(
id: string,
src/main/runtime/rpc/methods/browser-core.ts
+5 -2
View File
@@ -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'
src/main/runtime/rpc/methods/browser-extras.ts
+1 -1
View File
@@ -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,
src/main/runtime/rpc/methods/browser-tab-create-schema.test.ts
+1 -1
View File
@@ -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', () => {
src/main/runtime/rpc/methods/browser-tab-create-schema.ts
-4
View File
@@ -1,4 +0,0 @@
export {
BrowserOpenUrlParams,
BrowserTabCreateParams
} from '../../../../shared/rpc-contract/browser-tab-create-params'
src/main/runtime/rpc/methods/client-settings-schemas.ts
-5
View File
@@ -1,5 +0,0 @@
export {
NativeChatSessionOptionsMutation,
PRBotAuthorOverrideUpdate,
SettingsUpdate
} from '../../../../shared/rpc-contract/client-settings-params'
src/main/runtime/rpc/methods/client-ui-workspace-filter-fields.ts
-1
View File
@@ -1 +0,0 @@
export { ClientUiWorkspaceFilterFields } from '../../../../shared/rpc-contract/client-ui-workspace-filter-fields-params'
src/main/runtime/rpc/methods/client-ui.ts
+2 -2
View File
@@ -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({
src/main/runtime/rpc/methods/computer-schemas.ts
-15
View File
@@ -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'
src/main/runtime/rpc/methods/computer.ts
+1 -1
View File
@@ -21,7 +21,7 @@ import {
Scroll,
SetValue,
TypeText
} from './computer-schemas'
} from '../../../../shared/rpc-contract/computer-schemas-params'
import {
ComputerCapabilitiesParams,
ComputerPermissionsStatusParams
src/main/runtime/rpc/methods/files-target-schemas.ts
-1
View File
@@ -1 +0,0 @@
export { FileOpen, WorktreeSelector } from '../../../../shared/rpc-contract/files-target-params'
src/main/runtime/rpc/methods/files.ts
+1 -1
View File
@@ -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,
src/main/runtime/rpc/methods/git-admission-tier-schema.ts
-1
View File
@@ -1 +0,0 @@
export { OptionalGitAdmissionTier } from '../../../../shared/rpc-contract/git-admission-tier-params'
src/main/runtime/rpc/methods/git-commit-message-generation-methods.ts
+1 -1
View File
@@ -6,7 +6,7 @@ import {
GitGenerateCommitMessage,
GitGeneratePullRequestFields,
WorktreeSelector
} from './git-params'
} from '../../../../shared/rpc-contract/git-params'
type CommitMessageGenerationOverride = {
commitMessageAi?: GlobalSettings['commitMessageAi']
src/main/runtime/rpc/methods/git-diff-methods.ts
+1 -1
View File
@@ -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.
src/main/runtime/rpc/methods/git-params.ts
-26
View File
@@ -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'
src/main/runtime/rpc/methods/git.ts
+1 -1
View File
@@ -20,7 +20,7 @@ import {
GitSubmoduleStatus,
GitTargetedRemote,
WorktreeSelector
} from './git-params'
} from '../../../../shared/rpc-contract/git-params'
export const GIT_METHODS = [
defineMethod({
src/main/runtime/rpc/methods/github-issue-update-schema.ts
-1
View File
@@ -1 +0,0 @@
export { IssueUpdate } from '../../../../shared/rpc-contract/github-issue-update-params'
src/main/runtime/rpc/methods/github-project-methods.ts
+1 -1
View File
@@ -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,
src/main/runtime/rpc/methods/github-repo-target-schemas.ts
-1
View File
@@ -1 +0,0 @@
export { RepoSelector, SlugRepo } from '../../../../shared/rpc-contract/github-repo-target-params'
src/main/runtime/rpc/methods/github-repo-work-item-methods.ts
+1 -1
View File
@@ -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,
src/main/runtime/rpc/methods/linear-issue-attribute-filter-schema.ts
-1
View File
@@ -1 +0,0 @@
export { LinearIssueAttributeFilterSchema } from '../../../../shared/rpc-contract/linear-issue-attribute-filter-params'
src/main/runtime/rpc/methods/mobile-markdown-tab-methods.ts
-24
View File
@@ -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
)
})
]
src/main/runtime/rpc/methods/orchestration/schemas.ts
+5 -1
View File
@@ -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'
src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts
+12 -11
View File
@@ -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,
src/main/runtime/rpc/methods/orchestration/worker/worker-release-schemas.ts
-6
View File
@@ -1,6 +0,0 @@
export {
WORKER_TERMINAL_LIST_STATES,
WorkerDispatchParams,
WorkerListParams,
WorkerRetainParams
} from '../../../../../../shared/rpc-contract/orchestration-worker-release-schemas-params'
src/main/runtime/rpc/methods/orchestration/worker/worker-release.ts
+4 -1
View File
@@ -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 = [
src/main/runtime/rpc/methods/repo-update-schema.ts
-4
View File
@@ -1,4 +0,0 @@
export {
RepoSourceControlAiOverrides,
createRepoUpdateSchema
} from '../../../../shared/rpc-contract/repo-update-params'
src/main/runtime/rpc/methods/repo.ts
+1 -1
View File
@@ -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
src/main/runtime/rpc/methods/session-tab-close-methods.ts
+4 -1
View File
@@ -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'
src/main/runtime/rpc/methods/session-tab-markdown-methods.ts
+4 -1
View File
@@ -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({
src/main/runtime/rpc/methods/session-tab-mutation-methods.ts
+6 -1
View File
@@ -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({
src/main/runtime/rpc/methods/session-tabs-pane-layout-intent.test.ts
+1 -1
View File
@@ -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',
src/main/runtime/rpc/methods/session-tabs-schemas.test.ts
+6 -1
View File
@@ -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'
src/main/runtime/rpc/methods/session-tabs-schemas.ts
-14
View File
@@ -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'
src/main/runtime/rpc/methods/session-tabs.ts
+1 -1
View File
@@ -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,
src/main/runtime/rpc/methods/terminal-quick-command-rpc-schema.ts
-1
View File
@@ -1 +0,0 @@
export { TerminalQuickCommandsUpdate } from '../../../../shared/rpc-contract/terminal-quick-command-params'
src/main/runtime/rpc/methods/terminal/stream-schemas.ts
+2 -7
View File
@@ -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),
src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts
+13 -16
View File
@@ -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') {
src/main/runtime/rpc/methods/terminal/terminal-inspect-process-params.test.ts
+4 -1
View File
@@ -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. */
src/main/runtime/rpc/methods/terminal/terminal-legacy-subscription-types.ts
+1 -1
View File
@@ -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'
src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts
+2 -2
View File
@@ -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({
src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts
+2 -9
View File
@@ -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)
src/main/runtime/rpc/methods/terminal/terminal-multiplex-connection.ts
+12 -13
View File
@@ -102,17 +102,16 @@ export type TerminalMultiplexSubscribeFrame = {
handleSubscribeFrame: (payload: Uint8Array<ArrayBufferLike>) => Promise<void>
}
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<TerminalMultiplexConnection, 'closed' | 'streams'>,
stream: TerminalMultiplexStream
): boolean {
return !state.closed && state.streams.get(stream.streamId) === stream
}
src/main/runtime/rpc/methods/terminal/terminal-multiplex-flow-control.ts
+44 -58
View File
@@ -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<NonNullable<SerializedSnapshot>, '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<void> => {
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()
}
src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts
+8 -13
View File
@@ -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 (
src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts
+7 -6
View File
@@ -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<MultiplexPublishedInitialState | null> {
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) {
src/main/runtime/rpc/methods/terminal/terminal-multiplex-live-stream.ts
+7 -10
View File
@@ -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
src/main/runtime/rpc/methods/terminal/terminal-multiplex-method.ts
+5 -4
View File
@@ -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)
src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts
+50 -64
View File
@@ -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<void> => {
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(() => {})
}
}
}
src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts
+6 -5
View File
@@ -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
src/main/runtime/rpc/methods/terminal/terminal-multiplex-subscribe-frame.ts
+2 -8
View File
@@ -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<unknown>(payload)
const parsed = TerminalMultiplexSubscribeFrame.safeParse(raw)
src/main/runtime/rpc/methods/terminal/terminal-query-methods.ts
+1 -1
View File
@@ -8,7 +8,7 @@ import {
TerminalRename,
TerminalResolveActive,
TerminalResolvePane
} from './unary-schemas'
} from '../../../../../shared/rpc-contract/terminal-unary-params'
export const TERMINAL_QUERY_METHODS = [
defineMethod({
src/main/runtime/rpc/methods/terminal/terminal-send-method.ts
+1 -1
View File
@@ -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,
src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts
+22 -23
View File
@@ -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<T>(
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(
src/main/runtime/rpc/methods/terminal/terminal-subscribe-method.ts
+1 -1
View File
@@ -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 {
src/main/runtime/rpc/methods/terminal/terminal-viewer-colors-method.ts
+1 -1
View File
@@ -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({
src/main/runtime/rpc/methods/terminal/terminal-viewport-methods.ts
+2 -2
View File
@@ -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'
src/main/runtime/rpc/methods/terminal/unary-schemas.ts
-23
View File
@@ -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'
src/main/runtime/rpc/methods/terminal/viewport-schemas.ts
-6
View File
@@ -1,6 +0,0 @@
export {
TerminalSetAutoRestoreFit,
TerminalSetDisplayMode,
TerminalUnsubscribe,
TerminalUpdateViewport
} from '../../../../../shared/rpc-contract/terminal-viewport-schemas-params'
src/main/runtime/rpc/methods/ui-update-value-tolerance.ts
-4
View File
@@ -1,4 +0,0 @@
export {
omitUndefinedValues,
tolerateUnknownValues
} from '../../../../shared/rpc-contract/ui-update-value-tolerance-params'
src/main/runtime/rpc/methods/worktree-catalog-methods.ts
+1 -1
View File
@@ -10,7 +10,7 @@ import {
WorktreeDetectedListParams,
WorktreeListParams,
WorktreePsParams
} from './worktree-schemas'
} from '../../../../shared/rpc-contract/worktree-params'
export const WORKTREE_CATALOG_METHODS = [
defineMethod({
src/main/runtime/rpc/methods/worktree-create-args.test.ts
+1 -1
View File
@@ -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,
src/main/runtime/rpc/methods/worktree-create-args.ts
+1 -1
View File
@@ -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<typeof WorktreeCreate>
type ManagedWorktreeCreateArgs = Parameters<OrcaRuntimeService['createManagedWorktree']>[0]
src/main/runtime/rpc/methods/worktree-create-schemas.ts
-4
View File
@@ -1,4 +0,0 @@
export {
WorktreeCreate,
WorktreePrefetchCreateBase
} from '../../../../shared/rpc-contract/worktree-create-params'
src/main/runtime/rpc/methods/worktree-schemas.test.ts
+2 -2
View File
@@ -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', () => {
src/main/runtime/rpc/methods/worktree-schemas.ts
-18
View File
@@ -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'
src/main/runtime/rpc/methods/worktree-visibility-defaults-schema.ts
-1
View File
@@ -1 +0,0 @@
export { WorktreeVisibilityDefaultsUpdate } from '../../../../shared/rpc-contract/worktree-visibility-defaults-params'
src/main/runtime/rpc/methods/worktree.ts
+5 -2
View File
@@ -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'
src/main/runtime/rpc/rpc-request-admission-prologue.ts
+89
View File
@@ -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<M extends RpcMethod | RpcStreamingMethod>(
{ 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 })
}
src/main/runtime/rpc/rpc-streaming-dispatcher.ts
+38 -168
View File
@@ -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<void> {
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(
src/main/runtime/rpc/schemas.test.ts
+1 -1
View File
@@ -7,7 +7,7 @@ import {
OptionalPlainString,
OptionalPositiveInt,
OptionalString
} from './schemas'
} from '../../../shared/rpc-contract/rpc-param-primitives'
import {
InterceptEnable,
Screenshot,
src/main/runtime/rpc/schemas.ts
-17
View File
@@ -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'
src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts
+1 -1
View File
@@ -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) {
src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts
+2 -11
View File
@@ -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) {
src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts
+4 -8
View File
@@ -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()
}
src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts
+2 -6
View File
@@ -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) {
src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts
+2 -8
View File
@@ -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) {
src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts
+11
View File
@@ -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<ArrayBufferLike>
): RemoteRuntimeSnapshotInfo | null {
tests/e2e/cross-version-wire/cross-version-browser-placement.unit.test.ts
+1 -1
View File
@@ -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,