Fix orchestration mail from WSL and SSH terminals (#3918)

This commit is contained in:
Jinwoo Hong
2026-05-31 15:13:15 -04:00
committed by GitHub
parent ce552c4a26
commit 4cc0147005
29 changed files with 1205 additions and 77 deletions
+43 -1
View File
@@ -31,6 +31,7 @@ type PendingRequest = {
export type NotificationHandler = (method: string, params: Record<string, unknown>) => void
export type MethodNotificationHandler = (params: Record<string, unknown>) => void
export type RequestHandler = (params: Record<string, unknown>) => Promise<unknown> | unknown
const REQUEST_TIMEOUT_MS = 30_000
@@ -44,6 +45,7 @@ export class SshChannelMultiplexer {
private lastReceivedAt = Date.now()
private pendingRequests = new Map<number, PendingRequest>()
private notificationHandlers: NotificationHandler[] = []
private requestHandlers = new Map<string, RequestHandler>()
// Why: per-method dispatch map keeps streaming consumers (fs.streamChunk,
// fs.streamEnd, fs.streamError) from accreting string-match logic in the
// generic notification listener that already serves fs.changed.
@@ -118,6 +120,15 @@ export class SshChannelMultiplexer {
}
}
onRequest(method: string, handler: RequestHandler): () => void {
this.requestHandlers.set(method, handler)
return () => {
if (this.requestHandlers.get(method) === handler) {
this.requestHandlers.delete(method)
}
}
}
// Why: the session needs to know when the relay channel dies so it can
// auto-reconnect. Without this, a relay channel close (e.g. --connect
// bridge exits) leaves the session in 'ready' state with a dead mux
@@ -341,10 +352,41 @@ export class SshChannelMultiplexer {
private handleMessage(msg: JsonRpcMessage): void {
if ('id' in msg && ('result' in msg || 'error' in msg)) {
this.handleResponse(msg as JsonRpcResponse)
} else if ('id' in msg && 'method' in msg) {
void this.handleRequest(msg as JsonRpcRequest)
} else if ('method' in msg && !('id' in msg)) {
this.handleNotification(msg as JsonRpcNotification)
}
// Requests from relay to client are not expected in Phase 2
}
private async handleRequest(msg: JsonRpcRequest): Promise<void> {
const handler = this.requestHandlers.get(msg.method)
if (!handler) {
this.sendMessage({
jsonrpc: '2.0',
id: msg.id,
error: { code: -32601, message: `Method not found: ${msg.method}` }
})
return
}
try {
const result = await handler(msg.params ?? {})
this.sendMessage({
jsonrpc: '2.0',
id: msg.id,
result: result ?? null
})
} catch (err) {
this.sendMessage({
jsonrpc: '2.0',
id: msg.id,
error: {
code: (err as { code?: number }).code ?? -32000,
message: err instanceof Error ? err.message : String(err)
}
})
}
}
private handleResponse(msg: JsonRpcResponse): void {
+16 -5
View File
@@ -34,6 +34,10 @@ import {
export type RelayDeployResult = {
transport: MultiplexerTransport
platform: RelayPlatform
remoteHome?: string
remoteRelayDir?: string
nodePath?: string
sockPath?: string
}
// Why: individual exec commands have 30s timeouts, but the full deploy
@@ -161,7 +165,7 @@ async function deployAndLaunchRelayInner(
onProgress?.('Starting relay...')
console.log('[ssh-relay] Launching relay...')
const transport = await launchRelay(conn, remoteRelayDir, graceTimeSeconds, relayInstanceId)
const launched = await launchRelay(conn, remoteRelayDir, graceTimeSeconds, relayInstanceId)
console.log('[ssh-relay] Relay started successfully')
// Why: best-effort cleanup of unreferenced sibling version dirs. Errors
@@ -169,7 +173,14 @@ async function deployAndLaunchRelayInner(
// can never block the user from connecting.
void gcOldRelayVersions(conn, remoteHome, remoteRelayDir).catch(() => {})
return { transport, platform }
return {
transport: launched.transport,
platform,
remoteHome,
remoteRelayDir,
nodePath: launched.nodePath,
sockPath: launched.sockPath
}
}
async function detectRemotePlatform(conn: SshConnection): Promise<RelayPlatform | null> {
@@ -423,7 +434,7 @@ async function launchRelay(
remoteDir: string,
graceTimeSeconds?: number,
relayInstanceId?: string
): Promise<MultiplexerTransport> {
): Promise<{ transport: MultiplexerTransport; nodePath: string; sockPath: string }> {
// Why: Phase 1 of the plan requires Node.js on the remote. We use the
// system `node` rather than bundling a node binary, keeping the relay
// package small (~100KB JS vs ~60MB with embedded node).
@@ -466,7 +477,7 @@ async function launchRelay(
)
const transport = await waitForSentinel(channel)
console.log('[ssh-relay] Reconnected to existing relay via socket')
return transport
return { transport, nodePath, sockPath: sockFile }
} catch (err) {
console.warn(
'[ssh-relay] Socket reconnect failed, launching fresh relay:',
@@ -555,5 +566,5 @@ async function launchRelay(
const channel = await conn.exec(
`cd ${escapedDir} && ${escapedNode} relay.js --connect --sock-path ${shellEscape(sockFile)}`
)
return waitForSentinel(channel)
return { transport: await waitForSentinel(channel), nodePath, sockPath: sockFile }
}
@@ -17,6 +17,7 @@ vi.mock('./ssh-channel-multiplexer', () => {
notify = vi.fn()
request = vi.fn().mockResolvedValue([])
onNotification = vi.fn().mockReturnValue(() => {})
onRequest = vi.fn().mockReturnValue(() => {})
onDispose = vi.fn().mockReturnValue(() => {})
dispose = vi.fn()
isDisposed = vi.fn().mockReturnValue(false)
+1
View File
@@ -24,6 +24,7 @@ vi.mock('./ssh-channel-multiplexer', () => {
notify = vi.fn()
request = muxRequestMock
onNotification = vi.fn().mockReturnValue(() => {})
onRequest = vi.fn().mockReturnValue(() => {})
onDispose = vi.fn().mockReturnValue(() => {})
dispose = vi.fn()
isDisposed = vi.fn().mockReturnValue(false)
+107 -13
View File
@@ -10,6 +10,7 @@
import type { BrowserWindow } from 'electron'
import { deployAndLaunchRelay } from './ssh-relay-deploy'
import { execCommand } from './ssh-relay-deploy-helpers'
import { isRelayVersionMismatchError } from './ssh-relay-version-mismatch-error'
import type { RelayVersionMismatchError } from './ssh-relay-version-mismatch-error'
import { SshChannelMultiplexer } from './ssh-channel-multiplexer'
@@ -49,6 +50,7 @@ import { notifyRemoteWorkspaceHandlers } from '../ipc/remote-workspace-events'
import { PortScanner } from './ssh-port-scanner'
import type { SshPortForwardManager } from './ssh-port-forward'
import type { SshConnection } from './ssh-connection'
import { shellEscape } from './ssh-connection-utils'
import {
DEFAULT_SSH_RELAY_GRACE_PERIOD_SECONDS,
type DetectedPort,
@@ -58,6 +60,7 @@ import {
} from '../../shared/ssh-types'
import type { Store } from '../persistence'
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
import { runRemoteOrcaCli } from './ssh-remote-orca-cli'
export type RelaySessionState = 'idle' | 'deploying' | 'ready' | 'reconnecting' | 'disposed'
@@ -97,6 +100,12 @@ export class SshRelaySession {
private _onReady: ((targetId: string) => void) | null = null
private portScanner: PortScanner | null = null
private currentConnection: SshConnection | null = null
private remoteCliBridgeEnv: {
binDir: string
relayDir: string
nodePath: string
sockPath: string
} | null = null
constructor(
readonly targetId: string,
@@ -184,12 +193,17 @@ export class SshRelaySession {
this.currentConnection = conn
try {
const { transport } = await deployAndLaunchRelay(
conn,
undefined,
graceTimeSeconds,
this.targetId
)
const { transport, remoteHome, remoteRelayDir, nodePath, sockPath } =
await deployAndLaunchRelay(conn, undefined, graceTimeSeconds, this.targetId)
this.remoteCliBridgeEnv =
remoteHome && remoteRelayDir && nodePath && sockPath
? {
binDir: `${remoteHome}/.orca-relay/bin`,
relayDir: remoteRelayDir,
nodePath,
sockPath
}
: null
// Why: dispose() can fire during the await above (e.g. user clicks
// disconnect while relay is deploying). If so, the session is already
@@ -303,12 +317,17 @@ export class SshRelaySession {
this.teardownProviders('connection_lost')
try {
const { transport } = await deployAndLaunchRelay(
conn,
undefined,
graceTimeSeconds,
this.targetId
)
const { transport, remoteHome, remoteRelayDir, nodePath, sockPath } =
await deployAndLaunchRelay(conn, undefined, graceTimeSeconds, this.targetId)
this.remoteCliBridgeEnv =
remoteHome && remoteRelayDir && nodePath && sockPath
? {
binDir: `${remoteHome}/.orca-relay/bin`,
relayDir: remoteRelayDir,
nodePath,
sockPath
}
: null
if (abortController.signal.aborted || this.isDisposed()) {
// Why: the relay is already running on the remote. Creating a temporary
@@ -487,7 +506,14 @@ export class SshRelaySession {
return false
}
const ptyProvider = new SshPtyProvider(this.targetId, mux)
await this.installRemoteOrcaCliShim()
if (shouldContinue && !shouldContinue()) {
return false
}
this.wireUpRemoteOrcaCli(mux)
const ptyProvider = new SshPtyProvider(this.targetId, mux, this.remoteCliBridgeEnv ?? undefined)
registerSshPtyProvider(this.targetId, ptyProvider)
const fsProvider = new SshFilesystemProvider(this.targetId, mux, () =>
@@ -559,6 +585,70 @@ export class SshRelaySession {
}
}
private async installRemoteOrcaCliShim(): Promise<void> {
if (!this.remoteCliBridgeEnv) {
return
}
const { binDir, relayDir, nodePath, sockPath } = this.remoteCliBridgeEnv
const shimPath = `${binDir}/orca`
const shim = [
'#!/usr/bin/env sh',
'set -eu',
`ORCA_RELAY_NODE_PATH=\${ORCA_RELAY_NODE_PATH:-${quoteSh(nodePath)}}`,
`ORCA_RELAY_DIR=\${ORCA_RELAY_DIR:-${quoteSh(relayDir)}}`,
`ORCA_RELAY_SOCKET_PATH=\${ORCA_RELAY_SOCKET_PATH:-${quoteSh(sockPath)}}`,
'if [ ! -S "$ORCA_RELAY_SOCKET_PATH" ]; then',
' echo "Orca SSH CLI bridge cannot find the relay socket: $ORCA_RELAY_SOCKET_PATH" >&2',
' exit 1',
'fi',
'exec "$ORCA_RELAY_NODE_PATH" "$ORCA_RELAY_DIR/relay.js" --sock-path "$ORCA_RELAY_SOCKET_PATH" --orca-cli "$@"',
''
].join('\n')
await execCommand(this.requireReadyConnection(), `mkdir -p ${shellEscape(binDir)}`)
const conn = this.requireReadyConnection()
if (typeof conn.writeFile === 'function') {
await conn.writeFile(shimPath, shim)
} else {
const sftp = await conn.sftp()
try {
await new Promise<void>((resolve, reject) => {
const ws = sftp.createWriteStream(shimPath)
sftp.once('error', reject)
ws.once('close', resolve)
ws.once('error', reject)
ws.end(shim)
})
} finally {
sftp.end()
}
}
await execCommand(conn, `chmod 755 ${shellEscape(shimPath)}`)
}
private wireUpRemoteOrcaCli(mux: SshChannelMultiplexer): void {
mux.onRequest('orca.cli', async (params) => {
if (!this.runtime) {
throw new Error('Orca runtime is unavailable')
}
const argv = Array.isArray(params.argv)
? params.argv.filter((item): item is string => typeof item === 'string')
: []
const cwd = typeof params.cwd === 'string' && params.cwd.length > 0 ? params.cwd : '/'
const rawEnv = params.env
const env =
rawEnv && typeof rawEnv === 'object' && !Array.isArray(rawEnv)
? Object.fromEntries(
Object.entries(rawEnv).filter(
(entry): entry is [string, string] =>
typeof entry[0] === 'string' && typeof entry[1] === 'string'
)
)
: {}
return await runRemoteOrcaCli(this.runtime, { argv, cwd, env })
})
}
// Why: ship the OpenCode plugin / Pi extension source bodies to the relay
// so it can materialize per-PTY overlay dirs and inject OPENCODE_CONFIG_DIR
// / PI_CODING_AGENT_DIR into spawn env. The strings change as we add agent
@@ -897,3 +987,7 @@ export class SshRelaySession {
}
}
}
function quoteSh(value: string): string {
return `'${value.replaceAll("'", `'\\''`)}'`
}
+125
View File
@@ -0,0 +1,125 @@
import { describe, expect, it, vi } from 'vitest'
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
import { runRemoteOrcaCli } from './ssh-remote-orca-cli'
describe('runRemoteOrcaCli', () => {
function createRuntime() {
const messages: {
id: string
from_handle: string
to_handle: string
subject: string
body?: string
read_at: string | null
}[] = []
let nextMessage = 1
const db = {
insertMessage: vi.fn(
(message: { from: string; to: string; subject: string; body?: string }) => {
const row = {
id: `msg_${nextMessage++}`,
from_handle: message.from,
to_handle: message.to,
subject: message.subject,
body: message.body,
read_at: null
}
messages.push(row)
return row
}
),
getUnreadMessages: vi.fn((handle: string) =>
messages.filter((message) => message.to_handle === handle && message.read_at === null)
),
getAllMessagesForHandle: vi.fn((handle: string) =>
messages.filter((message) => message.to_handle === handle)
),
markAsRead: vi.fn((ids: string[]) => {
for (const message of messages) {
if (ids.includes(message.id)) {
message.read_at = new Date(0).toISOString()
}
}
})
}
const runtime = {
getRuntimeId: () => 'runtime-test',
getStatus: () => ({
runtimeId: 'runtime-test',
rendererGraphEpoch: 1,
graphStatus: 'ready',
authoritativeWindowId: 1,
liveTabCount: 1,
liveLeafCount: 1
}),
getOrchestrationDb: () => db,
deliverPendingMessagesForHandle: vi.fn(),
notifyMessageArrived: vi.fn()
} as unknown as OrcaRuntimeService
return { runtime, db }
}
it('uses the remote ORCA_TERMINAL_HANDLE as orchestration sender identity', async () => {
const { runtime, db } = createRuntime()
const result = await runRemoteOrcaCli(runtime, {
argv: ['orchestration', 'send', '--to', 'term_windows', '--subject', 'ping', '--json'],
cwd: '/home/alice/repo',
env: { ORCA_TERMINAL_HANDLE: 'term_ssh' }
})
expect(result.exitCode).toBe(0)
const payload = JSON.parse(result.stdout) as { ok: boolean }
expect(payload.ok).toBe(true)
expect(db.getUnreadMessages('term_windows')[0]?.from_handle).toBe('term_ssh')
})
it('accepts equals-style orchestration flags in the remote shim', async () => {
const { runtime, db } = createRuntime()
const result = await runRemoteOrcaCli(runtime, {
argv: [
'orchestration',
'send',
'--to=term_windows',
'--subject=ping',
'--body=--literal-body',
'--json'
],
cwd: '/home/alice/repo',
env: { ORCA_TERMINAL_HANDLE: 'term_ssh' }
})
expect(result.exitCode).toBe(0)
const payload = JSON.parse(result.stdout) as { ok: boolean }
expect(payload.ok).toBe(true)
const message = db.getUnreadMessages('term_windows')[0]
expect(message?.from_handle).toBe('term_ssh')
expect(message?.body).toBe('--literal-body')
})
it('uses the remote ORCA_TERMINAL_HANDLE as orchestration check identity', async () => {
const { runtime, db } = createRuntime()
db.insertMessage({
from: 'term_windows',
to: 'term_ssh',
subject: 'pong',
body: 'hello'
})
const result = await runRemoteOrcaCli(runtime, {
argv: ['orchestration', 'check', '--all', '--json'],
cwd: '/home/alice/repo',
env: { ORCA_TERMINAL_HANDLE: 'term_ssh' }
})
expect(result.exitCode).toBe(0)
const payload = JSON.parse(result.stdout) as {
ok: boolean
result: { count: number; messages: { subject: string }[] }
}
expect(payload.ok).toBe(true)
expect(payload.result.count).toBe(1)
expect(payload.result.messages[0]?.subject).toBe('pong')
})
})
+216
View File
@@ -0,0 +1,216 @@
import type { CliStatusResult, RuntimeStatus } from '../../shared/runtime-types'
import { RpcDispatcher } from '../runtime/rpc/dispatcher'
import type { RpcResponse } from '../runtime/rpc/core'
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
export type RemoteOrcaCliRequest = {
argv: string[]
cwd: string
env: Record<string, string>
}
export type RemoteOrcaCliResult = {
stdout: string
stderr: string
exitCode: number
}
type ParsedRemoteCli = {
commandPath: string[]
flags: Map<string, string | boolean>
}
export async function runRemoteOrcaCli(
runtime: OrcaRuntimeService,
request: RemoteOrcaCliRequest
): Promise<RemoteOrcaCliResult> {
const dispatcher = new RpcDispatcher({ runtime })
const parsed = parseRemoteCliArgs(request.argv)
const json = parsed.flags.has('json')
try {
const response = await dispatchRemoteCli(dispatcher, parsed, request.env)
return {
stdout: json ? `${JSON.stringify(response, null, 2)}\n` : `${formatRemoteCli(response)}\n`,
stderr: '',
exitCode: response.ok ? 0 : 1
}
} catch (err) {
const message = err instanceof Error ? err.message : String(err)
if (json) {
return {
stdout: `${JSON.stringify(buildLocalError(message), null, 2)}\n`,
stderr: '',
exitCode: 1
}
}
return { stdout: '', stderr: `${message}\n`, exitCode: 1 }
}
}
async function dispatchRemoteCli(
dispatcher: RpcDispatcher,
parsed: ParsedRemoteCli,
env: Record<string, string>
): Promise<RpcResponse> {
const command = parsed.commandPath.join(' ')
switch (command) {
case 'status': {
const response = await call(dispatcher, 'status.get')
if (!response.ok) {
return response
}
const status = response.result as RuntimeStatus
const cliStatus: CliStatusResult = {
app: { running: true, pid: null },
runtime: {
state: status.graphStatus === 'ready' ? 'ready' : 'graph_not_ready',
reachable: true,
runtimeId: status.runtimeId
},
graph: { state: status.graphStatus }
}
return { ...response, result: cliStatus }
}
case 'terminal list':
return await call(dispatcher, 'terminal.list', {
worktree: optionalString(parsed.flags, 'worktree'),
limit: optionalNumber(parsed.flags, 'limit')
})
case 'orchestration send':
return await call(dispatcher, 'orchestration.send', {
from: resolveHandle(parsed.flags, env, 'from'),
to: requiredString(parsed.flags, 'to'),
subject: requiredString(parsed.flags, 'subject'),
body: optionalString(parsed.flags, 'body'),
type: optionalString(parsed.flags, 'type'),
priority: optionalString(parsed.flags, 'priority'),
threadId: optionalString(parsed.flags, 'thread-id'),
payload: optionalString(parsed.flags, 'payload')
})
case 'orchestration check':
return await call(dispatcher, 'orchestration.check', {
terminal: resolveHandle(parsed.flags, env, 'terminal'),
unread: parsed.flags.has('unread') ? true : undefined,
all: parsed.flags.has('all') ? true : undefined,
types: optionalString(parsed.flags, 'types'),
inject: parsed.flags.has('inject') ? true : undefined,
wait: parsed.flags.has('wait') ? true : undefined,
timeoutMs: optionalNumber(parsed.flags, 'timeout-ms')
})
case 'orchestration reply':
return await call(dispatcher, 'orchestration.reply', {
id: requiredString(parsed.flags, 'id'),
body: requiredString(parsed.flags, 'body'),
from: resolveHandle(parsed.flags, env, 'from')
})
case 'orchestration inbox':
return await call(dispatcher, 'orchestration.inbox', {
limit: optionalNumber(parsed.flags, 'limit'),
terminal: optionalString(parsed.flags, 'terminal')
})
default:
throw new Error(`Unsupported SSH Orca CLI command: ${command}`)
}
}
async function call(
dispatcher: RpcDispatcher,
method: string,
params?: Record<string, unknown>
): Promise<RpcResponse> {
return await dispatcher.dispatch({
id: `remote-cli-${Date.now()}`,
authToken: 'remote-cli',
method,
params
})
}
function parseRemoteCliArgs(argv: string[]): ParsedRemoteCli {
const commandPath: string[] = []
const flags = new Map<string, string | boolean>()
for (let i = 0; i < argv.length; i += 1) {
const token = argv[i]
if (!token.startsWith('--')) {
commandPath.push(token)
continue
}
const assignment = token.slice(2)
// Why: the SSH relay-backed shim should accept the same `--flag=value`
// form as the local CLI, including values that themselves start with `--`.
const equalsIndex = assignment.indexOf('=')
if (equalsIndex !== -1) {
flags.set(assignment.slice(0, equalsIndex), assignment.slice(equalsIndex + 1))
continue
}
const flag = assignment
const next = argv[i + 1]
if (next && !next.startsWith('--')) {
flags.set(flag, next)
i += 1
} else {
flags.set(flag, true)
}
}
return { commandPath, flags }
}
function resolveHandle(
flags: Map<string, string | boolean>,
env: Record<string, string>,
flagName: string
): string {
return optionalString(flags, flagName) ?? env.ORCA_TERMINAL_HANDLE ?? 'unknown'
}
function requiredString(flags: Map<string, string | boolean>, name: string): string {
const value = optionalString(flags, name)
if (!value) {
throw new Error(`Missing --${name}`)
}
return value
}
function optionalString(flags: Map<string, string | boolean>, name: string): string | undefined {
const value = flags.get(name)
return typeof value === 'string' && value.length > 0 ? value : undefined
}
function optionalNumber(flags: Map<string, string | boolean>, name: string): number | undefined {
const value = optionalString(flags, name)
if (value === undefined) {
return undefined
}
const parsed = Number(value)
return Number.isFinite(parsed) ? parsed : undefined
}
function formatRemoteCli(response: RpcResponse): string {
if (!response.ok) {
return response.error.message
}
const result = response.result as Record<string, unknown>
if ('app' in result && 'runtime' in result && 'graph' in result) {
const status = result as CliStatusResult
return [
`appRunning: ${status.app.running}`,
`pid: ${status.app.pid ?? 'none'}`,
`runtimeState: ${status.runtime.state}`,
`runtimeReachable: ${status.runtime.reachable}`,
`runtimeId: ${status.runtime.runtimeId ?? 'none'}`,
`graphState: ${status.graph.state}`
].join('\n')
}
return JSON.stringify(response.result)
}
function buildLocalError(message: string): RpcResponse {
return {
id: 'remote-cli-local',
ok: false,
error: { code: 'runtime_error', message },
_meta: { runtimeId: 'unknown' }
}
}