diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index be807c28a2f..18272f33e64 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -7,6 +7,48 @@ import { } from '../flags' import { getTerminalHandle } from '../selectors' +// Why: 15 s is well under Claude Code's empirical ~2 min Bash-tool silence +// budget and generates only ~40 lines per 10 min wait — enough to assure the +// parent process the subprocess is alive without flooding logs. See design +// doc §3.4. +const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000 + +// Why: test-only escape hatch so subprocess tests can verify the feature in +// under 10 s rather than needing a full 15 s silence window. Production users +// should never set this — there is no surface documentation. A bogus value +// falls back to the default rather than disabling the heartbeat. +function resolveHeartbeatIntervalMs(): number { + const raw = process.env.ORCA_HEARTBEAT_INTERVAL_MS + if (!raw) { + return DEFAULT_HEARTBEAT_INTERVAL_MS + } + const parsed = Number(raw) + if (!Number.isFinite(parsed) || parsed <= 0) { + return DEFAULT_HEARTBEAT_INTERVAL_MS + } + return parsed +} + +function startCheckHeartbeat(deadlineMs: number | undefined): () => void { + const startedAt = Date.now() + const interval = setInterval(() => { + const payload = { + _heartbeat: true, + elapsedMs: Date.now() - startedAt, + deadlineMs: deadlineMs ?? null + } + // Why: `process.stderr.write` is line-flushed per-call in Node, whereas a + // fully-buffered writer would hold all heartbeat lines until exit and + // silently defeat the whole point of the ping. Subprocess test asserts + // this by reading stderr incrementally. See §3.4. + process.stderr.write(`${JSON.stringify(payload)}\n`) + }, resolveHeartbeatIntervalMs()) + if (typeof interval.unref === 'function') { + interval.unref() + } + return () => clearInterval(interval) +} + type MessageSummary = { id: string from_handle: string @@ -62,18 +104,36 @@ export const ORCHESTRATION_HANDLERS: Record = { 'orchestration check': async ({ flags, client, cwd, json }) => { const terminal = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'terminal') - const result = await client.call<{ + const wait = flags.has('wait') + const timeoutMs = flags.has('timeout-ms') ? Number(flags.get('timeout-ms')) : undefined + + // Why: Claude Code's Bash tool auto-backgrounds subprocesses that produce + // no output for ~2 min (shorter on the non-interactive path). Emit a + // heartbeat line to stderr every HEARTBEAT_INTERVAL_MS while the wait is + // active so the parent process can see the subprocess is still alive. + // Stderr rather than stdout so stdout stays a single final JSON payload, + // and JSON-shaped rather than `# …` so `2>&1 | jq` pipelines still work + // (jq refuses `#`-prefixed lines). See design doc §3.4. + const stopHeartbeat = wait ? startCheckHeartbeat(timeoutMs) : null + type CheckResult = { messages: MessageSummary[] count: number formatted?: string - }>('orchestration.check', { - terminal, - unread: flags.has('unread') ? true : undefined, - types: getOptionalStringFlag(flags, 'types'), - inject: flags.has('inject') ? true : undefined, - wait: flags.has('wait') ? true : undefined, - timeoutMs: flags.has('timeout-ms') ? Number(flags.get('timeout-ms')) : undefined - }) + } + let result: Awaited>> + try { + result = await client.call('orchestration.check', { + terminal, + unread: flags.has('unread') ? true : undefined, + all: flags.has('all') ? true : undefined, + types: getOptionalStringFlag(flags, 'types'), + inject: flags.has('inject') ? true : undefined, + wait: wait ? true : undefined, + timeoutMs + }) + } finally { + stopHeartbeat?.() + } printResult(result, json, (r) => { if (r.formatted) { return r.formatted @@ -102,7 +162,8 @@ export const ORCHESTRATION_HANDLERS: Record = { messages: MessageSummary[] count: number }>('orchestration.inbox', { - limit: getOptionalPositiveIntegerFlag(flags, 'limit') + limit: getOptionalPositiveIntegerFlag(flags, 'limit'), + terminal: getOptionalStringFlag(flags, 'terminal') }) printResult(result, json, (r) => { if (r.count === 0) { diff --git a/src/cli/orchestration.subprocess.test.ts b/src/cli/orchestration.subprocess.test.ts new file mode 100644 index 00000000000..c8648ceafe6 --- /dev/null +++ b/src/cli/orchestration.subprocess.test.ts @@ -0,0 +1,158 @@ +// Why: subprocess-level test for the CLI heartbeat behavior described in +// design doc §3.4. Spawns the real compiled CLI with no TTY, points it at a +// real in-process runtime via ORCA_USER_DATA_PATH, and asserts: +// - the first heartbeat line appears on stderr well under Claude Code's +// ~2 min Bash-tool silence budget (we verify with a shortened interval; +// production uses 15 s via the same code path) +// - ≥3 heartbeats arrive during the wait window +// - stderr is line-flushed (we observe each heartbeat as a separate chunk +// before the process exits — not in one burst at the end) +// - stdout stays a single clean JSON payload (no heartbeats leak to stdout) +// - a `jq "select(._heartbeat|not)"` filter on the merged stream would +// yield exactly the final result +// +// This test is skipped if the CLI hasn't been built yet (out/cli/index.js +// missing) so `pnpm test` works on a fresh checkout without requiring a prior +// `pnpm run build:cli`. The verification gate explicitly builds the CLI +// before running this file. +import { spawn } from 'child_process' +import { existsSync, mkdtempSync } from 'fs' +import { tmpdir } from 'os' +import { join } from 'path' +import { describe, expect, it } from 'vitest' +import { OrcaRuntimeService } from '../main/runtime/orca-runtime' +import { OrchestrationDb } from '../main/runtime/orchestration/db' +import { OrcaRuntimeRpcServer } from '../main/runtime/runtime-rpc' + +// Why: Vitest runs tests with `process.cwd()` pinned to the repo root, so +// join against it to locate the compiled CLI regardless of where this test +// file itself lives. +const CLI_PATH = join(process.cwd(), 'out', 'cli', 'index.js') + +const describeIfBuilt = existsSync(CLI_PATH) ? describe : describe.skip + +describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => { + it('emits newline-flushed JSON heartbeats to stderr while waiting', async () => { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-cli-sub-')) + const runtime = new OrcaRuntimeService() + const db = new OrchestrationDb(':memory:') + runtime.setOrchestrationDb(db) + const server = new OrcaRuntimeRpcServer({ runtime, userDataPath }) + await server.start() + + try { + // Why: use the ORCA_HEARTBEAT_INTERVAL_MS escape hatch to shrink the + // test to ~1 s wall time. Production callers never set this; the + // production default (15 s) is exercised by §3.4's own unit tests + // and by the fact that this same code path runs with the real + // constant when the env var is absent. + const heartbeatMs = 200 + const waitTimeoutMs = 1200 + + const child = spawn( + process.execPath, + [ + CLI_PATH, + 'orchestration', + 'check', + '--wait', + '--timeout-ms', + String(waitTimeoutMs), + '--json' + ], + { + env: { + ...process.env, + ORCA_USER_DATA_PATH: userDataPath, + ORCA_TERMINAL_HANDLE: 'term_nobody', + ORCA_HEARTBEAT_INTERVAL_MS: String(heartbeatMs) + }, + // Why: explicit pipe for all three fds so we can watch stderr + // in real time; no TTY attached (Bash-tool parity). + stdio: ['ignore', 'pipe', 'pipe'] + } + ) + + const stderrChunks: { at: number; data: string }[] = [] + const stdoutChunks: { at: number; data: string }[] = [] + const startedAt = Date.now() + child.stderr.setEncoding('utf8') + child.stdout.setEncoding('utf8') + child.stderr.on('data', (d) => stderrChunks.push({ at: Date.now() - startedAt, data: d })) + child.stdout.on('data', (d) => stdoutChunks.push({ at: Date.now() - startedAt, data: d })) + + const exitCode = await new Promise((resolveExit, rejectExit) => { + child.once('exit', (code) => resolveExit(code ?? 1)) + child.once('error', rejectExit) + }) + + expect(exitCode).toBe(0) + + const stderr = stderrChunks.map((c) => c.data).join('') + const stdout = stdoutChunks.map((c) => c.data).join('') + + const heartbeatLines = stderr + .split('\n') + .filter((line) => line.trim().length > 0) + .map((line) => { + try { + return JSON.parse(line) as Record + } catch { + return null + } + }) + .filter((p): p is Record => p !== null && p._heartbeat === true) + + // ≥3 heartbeats in a 1.2s window with a 200ms interval + expect(heartbeatLines.length).toBeGreaterThanOrEqual(3) + expect(heartbeatLines[0]).toHaveProperty('elapsedMs') + expect(heartbeatLines[0]).toHaveProperty('deadlineMs', waitTimeoutMs) + + // Why: first heartbeat must arrive within one interval + scheduler + // slack (300ms is generous); if the stream were fully buffered we'd + // see everything only after exit. + const firstHeartbeatChunk = stderrChunks.find((c) => c.data.includes('_heartbeat')) + expect(firstHeartbeatChunk).toBeDefined() + expect(firstHeartbeatChunk!.at).toBeLessThan(heartbeatMs + 300) + + // Why: line-flushing proof — the *first* heartbeat chunk must arrive + // strictly before the exit chunk; i.e. we got at least two separate + // stderr data events (heartbeat + final). A single-chunk delivery + // would indicate stderr was buffered until exit. + const lastStderrAt = stderrChunks.at(-1)?.at ?? 0 + const firstStderrAt = stderrChunks.at(0)?.at ?? 0 + expect(lastStderrAt).toBeGreaterThan(firstStderrAt) + + // Stdout: exactly one JSON payload, the terminal result. No heartbeats + // leak, and the content parses as valid JSON. + const stdoutTrimmed = stdout.trim() + const stdoutPayload = JSON.parse(stdoutTrimmed) as Record + expect(stdoutPayload).not.toHaveProperty('_heartbeat') + expect(stdoutTrimmed).not.toContain('_heartbeat') + // Why: result should be an RPC success envelope with the expected + // shape. `count: 0` and `messages: []` because the wait timed out + // with no message for term_nobody. + expect(stdoutPayload).toMatchObject({ ok: true }) + + // Why: the heartbeats-on-stderr design is meant to pair with shell + // filters like `2>&1 | jq "select(._heartbeat|not)"`. jq is + // line-oriented by default, but also accepts pretty-printed JSON + // across multiple lines. What matters here is that every + // heartbeat line on stderr is a standalone JSON object (so jq can + // match it) and doesn't span multiple lines — assert that each + // heartbeat is a single-line JSON with no embedded newlines. + for (const line of stderr.split('\n')) { + if (line.trim().length === 0) { + continue + } + if (line.includes('_heartbeat')) { + expect(() => JSON.parse(line)).not.toThrow() + expect(line).not.toContain('\n') + } + } + } finally { + db.close() + await server.stop() + } + }, 30_000) +}) diff --git a/src/cli/runtime/client.ts b/src/cli/runtime/client.ts index e256bcdabd0..ae65cd4f34c 100644 --- a/src/cli/runtime/client.ts +++ b/src/cli/runtime/client.ts @@ -5,6 +5,14 @@ import { getCliStatus } from './status' import { sendRequest } from './transport' import { RuntimeClientError, RuntimeRpcFailureError, type RuntimeRpcSuccess } from './types' +// Why: for `orchestration.check --wait` the caller's method-level +// `params.timeoutMs` is the inner waiter budget; we extend the client-side +// socket timeout to `timeoutMs + GRACE_MS` so the client's own idle timer +// never fires before the server-side waiter has had a chance to resolve and +// emit its terminal frame. The 10 s grace absorbs round-trip + one final +// keepalive window. See design doc §3.1. +const LONG_POLL_CLIENT_GRACE_MS = 10_000 + export class RuntimeClient { private readonly userDataPath: string private readonly requestTimeoutMs: number @@ -25,18 +33,30 @@ export class RuntimeClient { } ): Promise> { const metadata = readMetadata(this.userDataPath) - const response = await sendRequest( - metadata, - method, - params, - options?.timeoutMs ?? this.requestTimeoutMs - ) + const effectiveTimeoutMs = options?.timeoutMs ?? this.resolveMethodTimeoutMs(method, params) + const response = await sendRequest(metadata, method, params, effectiveTimeoutMs) if (!response.ok) { throw new RuntimeRpcFailureError(response) } return response } + // Why: centralises the per-method timeout policy. `orchestration.check` with + // `wait: true` is the only long-poll today, and its inner waiter budget + // lives in `params.timeoutMs`. We widen the client-side socket timeout to + // `timeoutMs + grace` so it doesn't fire before the server has a chance to + // resolve. Without this, a 5 min wait would still die at the 60 s default. + // See design doc §3.1. + private resolveMethodTimeoutMs(method: string, params?: unknown): number { + if (method === 'orchestration.check' && isWaitingCheck(params)) { + const inner = Number((params as { timeoutMs?: unknown }).timeoutMs) + if (Number.isFinite(inner) && inner > 0) { + return Math.max(inner + LONG_POLL_CLIENT_GRACE_MS, this.requestTimeoutMs) + } + } + return this.requestTimeoutMs + } + async getCliStatus(): Promise> { return getCliStatus(this.userDataPath) } @@ -67,3 +87,12 @@ export class RuntimeClient { function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)) } + +function isWaitingCheck(params: unknown): boolean { + return ( + typeof params === 'object' && + params !== null && + 'wait' in params && + (params as { wait: unknown }).wait === true + ) +} diff --git a/src/cli/runtime/envelope-schema.test.ts b/src/cli/runtime/envelope-schema.test.ts index 57f92ada298..b7315644f4b 100644 --- a/src/cli/runtime/envelope-schema.test.ts +++ b/src/cli/runtime/envelope-schema.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { RuntimeRpcEnvelopeSchema } from './envelope-schema' +import { RuntimeRpcEnvelopeSchema, isKeepaliveFrame } from './envelope-schema' describe('RuntimeRpcEnvelopeSchema', () => { it('accepts a well-formed success envelope', () => { @@ -57,4 +57,37 @@ describe('RuntimeRpcEnvelopeSchema', () => { const parsed = RuntimeRpcEnvelopeSchema.safeParse({ hello: 'world' }) expect(parsed.success).toBe(false) }) + + it('accepts a keepalive frame', () => { + const parsed = RuntimeRpcEnvelopeSchema.safeParse({ _keepalive: true }) + expect(parsed.success).toBe(true) + }) + + it('rejects a keepalive frame with _keepalive !== true', () => { + const parsed = RuntimeRpcEnvelopeSchema.safeParse({ _keepalive: false }) + expect(parsed.success).toBe(false) + }) +}) + +describe('isKeepaliveFrame', () => { + it('returns true for a well-formed keepalive', () => { + expect(isKeepaliveFrame({ _keepalive: true })).toBe(true) + }) + + it('returns false for a success envelope', () => { + expect(isKeepaliveFrame({ id: 'x', ok: true, result: {}, _meta: { runtimeId: 'r' } })).toBe( + false + ) + }) + + it('returns false for non-object inputs', () => { + expect(isKeepaliveFrame(null)).toBe(false) + expect(isKeepaliveFrame(undefined)).toBe(false) + expect(isKeepaliveFrame('keepalive')).toBe(false) + }) + + it('returns false when _keepalive is not strictly true', () => { + expect(isKeepaliveFrame({ _keepalive: 1 })).toBe(false) + expect(isKeepaliveFrame({ _keepalive: 'true' })).toBe(false) + }) }) diff --git a/src/cli/runtime/envelope-schema.ts b/src/cli/runtime/envelope-schema.ts index eeec3536005..f674aab11af 100644 --- a/src/cli/runtime/envelope-schema.ts +++ b/src/cli/runtime/envelope-schema.ts @@ -36,4 +36,27 @@ const Failure = z.object({ _meta: MetaFailure }) -export const RuntimeRpcEnvelopeSchema = z.discriminatedUnion('ok', [Success, Failure]) +// Why: transport-layer keepalive frame (server→client only). Not a terminal +// frame — the client reads past it and keeps waiting for the real +// success/failure. `id` and `_meta` are deliberately absent: keepalives carry +// no method-level semantics and aren't tied to a particular request (one +// connection handles one request today). See design doc §3.1. +const Keepalive = z.object({ + _keepalive: z.literal(true) +}) + +// Why: switched from z.discriminatedUnion('ok', …) to z.union because +// keepalives have no `ok` field. Client code must branch on +// `'_keepalive' in frame` before treating the frame as Success/Failure. +export const RuntimeRpcEnvelopeSchema = z.union([Success, Failure, Keepalive]) + +export type RuntimeRpcKeepaliveFrame = z.infer + +export function isKeepaliveFrame(frame: unknown): frame is RuntimeRpcKeepaliveFrame { + return ( + typeof frame === 'object' && + frame !== null && + '_keepalive' in frame && + (frame as { _keepalive: unknown })._keepalive === true + ) +} diff --git a/src/cli/runtime/transport.ts b/src/cli/runtime/transport.ts index bc39de0d332..7209a274c0f 100644 --- a/src/cli/runtime/transport.ts +++ b/src/cli/runtime/transport.ts @@ -1,7 +1,7 @@ import { createConnection } from 'net' import { randomUUID } from 'crypto' import type { RuntimeMetadata, RuntimeTransportMetadata } from '../../shared/runtime-bootstrap' -import { RuntimeRpcEnvelopeSchema } from './envelope-schema' +import { isKeepaliveFrame, RuntimeRpcEnvelopeSchema } from './envelope-schema' import { RuntimeClientError, type RuntimeRpcResponse } from './types' export async function sendRequest( @@ -13,9 +13,14 @@ export async function sendRequest( return await new Promise((resolve, reject) => { const socket = createConnection(getTransportEndpoint(metadata.transport!)) let buffer = '' + let settled = false const requestId = randomUUID() const timeout = setTimeout(() => { + if (settled) { + return + } + settled = true socket.destroy() reject( new RuntimeClientError( @@ -25,28 +30,72 @@ export async function sendRequest( ) }, timeoutMs) + const finish = ( + result: { ok: true; response: RuntimeRpcResponse } | { ok: false; error: Error } + ): void => { + if (settled) { + return + } + settled = true + clearTimeout(timeout) + socket.end() + if (result.ok) { + resolve(result.response) + } else { + reject(result.error) + } + } + socket.setEncoding('utf8') socket.once('error', () => { - clearTimeout(timeout) - reject( - new RuntimeClientError( + finish({ + ok: false, + error: new RuntimeClientError( 'runtime_unavailable', 'Could not connect to the running Orca app. Restart Orca and try again.' ) - ) + }) }) socket.on('data', (chunk) => { buffer += chunk - const newlineIndex = buffer.indexOf('\n') - if (newlineIndex === -1) { - return - } - const message = buffer.slice(0, newlineIndex) - socket.end() - clearTimeout(timeout) - let response: RuntimeRpcResponse - try { - const raw: unknown = JSON.parse(message) + // Why: the server may interleave `{"_keepalive":true}\n` frames with the + // final success/failure frame to keep both idle timers alive during a + // long-poll (see design doc §3.1). Read frames in a loop until we see a + // terminal frame. Each keepalive refreshes the client-side timer so a + // 10 min wait doesn't trip the 60 s default ceiling. + let newlineIndex = buffer.indexOf('\n') + while (newlineIndex !== -1 && !settled) { + const line = buffer.slice(0, newlineIndex) + buffer = buffer.slice(newlineIndex + 1) + if (line.trim().length === 0) { + newlineIndex = buffer.indexOf('\n') + continue + } + + let raw: unknown + try { + raw = JSON.parse(line) + } catch { + finish({ + ok: false, + error: new RuntimeClientError( + 'invalid_runtime_response', + 'The Orca runtime returned an invalid response frame.' + ) + }) + return + } + + // Fast-path: ignore keepalives without running the full schema. + // setTimeout().refresh() is stable since Node 10 (Orca ships on + // Node 20+ via Electron and the standalone CLI targets the same + // major). See §7 risk #9. + if (isKeepaliveFrame(raw)) { + timeout.refresh() + newlineIndex = buffer.indexOf('\n') + continue + } + // Why: validate the envelope shape (id, ok, result/error, _meta) at // the decode boundary so version skew between the CLI and the Orca // main runtime surfaces as a single invalid_runtime_response instead @@ -54,43 +103,49 @@ export async function sendRequest( // unknown — the TResult generic is the caller's responsibility. const parsed = RuntimeRpcEnvelopeSchema.safeParse(raw) if (!parsed.success) { - reject( - new RuntimeClientError( + finish({ + ok: false, + error: new RuntimeClientError( 'invalid_runtime_response', 'The Orca runtime returned an invalid response frame.' ) - ) + }) return } - response = parsed.data as RuntimeRpcResponse - } catch { - reject( - new RuntimeClientError( - 'invalid_runtime_response', - 'The Orca runtime returned an invalid response frame.' - ) - ) + + // Narrow out keepalive (already filtered above) so TS can see a + // Success|Failure shape here. + const frame = parsed.data + if ('_keepalive' in frame) { + timeout.refresh() + newlineIndex = buffer.indexOf('\n') + continue + } + + const response = frame as RuntimeRpcResponse + if (response.id !== requestId) { + finish({ + ok: false, + error: new RuntimeClientError( + 'invalid_runtime_response', + 'The Orca runtime returned a mismatched response id.' + ) + }) + return + } + if (response._meta?.runtimeId && response._meta.runtimeId !== metadata.runtimeId) { + finish({ + ok: false, + error: new RuntimeClientError( + 'runtime_unavailable', + 'The Orca runtime changed while the request was in flight. Retry the command.' + ) + }) + return + } + finish({ ok: true, response }) return } - if (response.id !== requestId) { - reject( - new RuntimeClientError( - 'invalid_runtime_response', - 'The Orca runtime returned a mismatched response id.' - ) - ) - return - } - if (response._meta?.runtimeId && response._meta.runtimeId !== metadata.runtimeId) { - reject( - new RuntimeClientError( - 'runtime_unavailable', - 'The Orca runtime changed while the request was in flight. Retry the command.' - ) - ) - return - } - resolve(response) }) socket.on('connect', () => { socket.write( diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts index e527fe8b953..fe27807b10d 100644 --- a/src/cli/specs/orchestration.ts +++ b/src/cli/specs/orchestration.ts @@ -23,8 +23,23 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'check'], summary: 'Check messages for a terminal', usage: - 'orca orchestration check [--terminal ] [--unread] [--types ] [--inject] [--wait] [--timeout-ms ] [--json]', - allowedFlags: [...GLOBAL_FLAGS, 'terminal', 'unread', 'types', 'inject', 'wait', 'timeout-ms'] + 'orca orchestration check [--terminal ] [--unread | --all] [--types ] [--inject] [--wait] [--timeout-ms ] [--json]\n' + + ' --unread (default): return only unread messages and mark them read.\n' + + ' --all: return every message for the handle; does not mark read.\n' + + ' --wait: block until a matching message arrives or --timeout-ms expires.\n' + + ' Emits JSON heartbeat lines to stderr every 15s so the caller can\n' + + ' tell the process is alive. Filter with `grep -v _heartbeat` or\n' + + ' `jq "select(._heartbeat|not)"` when merging streams with 2>&1.', + allowedFlags: [ + ...GLOBAL_FLAGS, + 'terminal', + 'unread', + 'all', + 'types', + 'inject', + 'wait', + 'timeout-ms' + ] }, { path: ['orchestration', 'reply'], @@ -34,9 +49,9 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ }, { path: ['orchestration', 'inbox'], - summary: 'Show all messages across recipients', - usage: 'orca orchestration inbox [--limit ] [--json]', - allowedFlags: [...GLOBAL_FLAGS, 'limit'] + summary: 'Show messages across (or for) recipients', + usage: 'orca orchestration inbox [--limit ] [--terminal ] [--json]', + allowedFlags: [...GLOBAL_FLAGS, 'limit', 'terminal'] }, { path: ['orchestration', 'task-create'], diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index 744568a7a64..463b778b709 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -471,32 +471,44 @@ describe('OrcaRuntimeService', () => { it('delivers pending orchestration messages to an already-idle agent', async () => { vi.useFakeTimers() - const runtime = new OrcaRuntimeService(store) - const db = new OrchestrationDb(':memory:') - const write = vi.fn().mockReturnValue(true) - runtime.setOrchestrationDb(db) - runtime.setPtyController({ - write, - kill: vi.fn(), - getForegroundProcess: async () => null - }) - syncSinglePty(runtime) + try { + const runtime = new OrcaRuntimeService(store) + const db = new OrchestrationDb(':memory:') + const write = vi.fn().mockReturnValue(true) + runtime.setOrchestrationDb(db) + runtime.setPtyController({ + write, + kill: vi.fn(), + getForegroundProcess: async () => null + }) + syncSinglePty(runtime) - const [terminal] = (await runtime.listTerminals()).terminals - runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) - runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101) - db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello' }) + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101) + db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello' }) - runtime.deliverPendingMessagesForHandle(terminal.handle) + runtime.deliverPendingMessagesForHandle(terminal.handle) - expect(write).toHaveBeenCalledWith('pty-1', expect.stringContaining('Subject: hello')) - // Why: markAsRead is deferred until the 500ms delayed Enter is confirmed, - // so we must advance timers past the split-write delay. - await vi.advanceTimersByTimeAsync(500) - expect(write).toHaveBeenCalledWith('pty-1', '\r') - expect(db.getUnreadMessages(terminal.handle)).toHaveLength(0) - db.close() - vi.useRealTimers() + expect(write).toHaveBeenCalledWith('pty-1', expect.stringContaining('Subject: hello')) + // Why: the split Enter write lands after the 500ms delay, so we advance + // past it before asserting on delivered_at. + await vi.advanceTimersByTimeAsync(500) + expect(write).toHaveBeenCalledWith('pty-1', '\r') + + // Why: design doc §3.2 splits delivered vs. read — push-on-idle stamps + // `delivered_at` but must *not* flip `read`, since only the check caller + // (the agent) is authorized to consume messages from its queue. The + // injected banner is a courtesy; the rows stay unread so the agent can + // still observe them via `check` and resolve the consumption race. + const unread = db.getUnreadMessages(terminal.handle) + expect(unread).toHaveLength(1) + expect(unread[0].read).toBe(0) + expect(unread[0].delivered_at).not.toBeNull() + db.close() + } finally { + vi.useRealTimers() + } }) it('adopts preallocated ORCA_TERMINAL_HANDLE as a valid runtime handle', async () => { diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 3ba6281df65..85028967f20 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -1733,7 +1733,7 @@ export class OrcaRuntimeService { waitForMessage( handle: string, - options?: { typeFilter?: string[]; timeoutMs?: number } + options?: { typeFilter?: string[]; timeoutMs?: number; signal?: AbortSignal } ): Promise { return new Promise((resolve) => { const timeoutMs = options?.timeoutMs ?? MESSAGE_WAIT_DEFAULT_TIMEOUT_MS @@ -1745,7 +1745,27 @@ export class OrcaRuntimeService { timeout: null } + // Why: if the caller aborts (socket closed on the RPC side — see design + // doc §3.1 counter-lifecycle), resolve immediately so the long-poll slot + // is released instead of counting down the full timeoutMs with a dead + // client on the other end. + const signal = options?.signal + const onAbort = (): void => { + this.removeMessageWaiter(waiter) + resolve() + } + if (signal) { + if (signal.aborted) { + resolve() + return + } + signal.addEventListener('abort', onAbort, { once: true }) + } + waiter.timeout = setTimeout(() => { + if (signal) { + signal.removeEventListener('abort', onAbort) + } this.removeMessageWaiter(waiter) resolve() }, timeoutMs) @@ -2002,8 +2022,15 @@ export class OrcaRuntimeService { // Why: Claude Code treats large single PTY writes as paste events and // swallows a \r included in the same write. Send Enter separately after - // a delay so the agent processes the pasted message first. Mark messages - // as read only after \r is confirmed, so failed deliveries stay queued. + // a delay so the agent processes the pasted message first. Stamp + // `delivered_at` only after \r is confirmed, so failed deliveries stay + // queued. + // + // Important (design doc §3.2, feedback #2): we stamp `delivered_at` here + // instead of flipping `read`. `read` is reserved for "a check-caller + // consumed this message." Flipping `read` on push-on-idle would hide the + // message from the coordinator's next `check --unread`, which is the + // exact bug feedback #2 reported. The two bits must stay independent. const ptyId = leaf.ptyId setTimeout(() => { try { @@ -2012,11 +2039,12 @@ export class OrcaRuntimeService { } const submitted = this.ptyController?.write(ptyId, '\r') ?? false if (submitted) { - this._orchestrationDb?.markAsRead(unread.map((m) => m.id)) + this._orchestrationDb?.markAsDelivered(unread.map((m) => m.id)) } } catch { - // Terminal may have closed during the delay — messages stay unread - // and will be re-delivered on the next idle transition. + // Terminal may have closed during the delay — messages stay queued + // (delivered_at still NULL) and will be re-delivered on the next + // idle transition. } }, 500) } diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index b7c74bee72e..79a6fc97f1d 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -42,6 +42,24 @@ export class OrchestrationDb { this.db.pragma('synchronous = NORMAL') this.db.pragma('busy_timeout = 5000') this.createTables() + // Why: migrations must run synchronously before any method can be called + // on this instance. The migration is idempotent (PRAGMA check) and + // hard-fails on error — a DB missing `delivered_at` would cause push-on-idle + // to throw inside markAsDelivered, which combined with the long-poll + // counter risks leaking slots. Better to crash startup than serve traffic + // on a broken schema. See design doc §3.2 / §3.3 / §7 risk #3. + this.runMigrations() + } + + private runMigrations(): void { + const columns = this.db.prepare('PRAGMA table_info(messages)').all() as { name: string }[] + const hasDeliveredAt = columns.some((c) => c.name === 'delivered_at') + if (!hasDeliveredAt) { + // ALTER TABLE throws synchronously; any error other than + // "duplicate column name" (which is ruled out by the PRAGMA check above) + // must propagate so startup fails loudly. + this.db.exec('ALTER TABLE messages ADD COLUMN delivered_at TEXT') + } } private createTables(): void { @@ -195,12 +213,37 @@ export class OrchestrationDb { this.db.prepare(`UPDATE messages SET read = 1 WHERE id IN (${placeholders})`).run(...ids) } + // Why: `delivered_at` is stamped via SQLite's datetime('now') rather than a + // JS ISO string so it uses the same 'YYYY-MM-DD HH:MM:SS' UTC shape as the + // other SQL-default timestamps on this table. A future ORDER BY or + // comparison against created_at relies on this format consistency. + // See design doc §3.2. + markAsDelivered(ids: string[]): void { + if (ids.length === 0) { + return + } + const placeholders = ids.map(() => '?').join(',') + this.db + .prepare(`UPDATE messages SET delivered_at = datetime('now') WHERE id IN (${placeholders})`) + .run(...ids) + } + getInbox(limit = 20): MessageRow[] { return this.db .prepare('SELECT * FROM messages ORDER BY sequence DESC LIMIT ?') .all(limit) as MessageRow[] } + // Why: used by `check --all` and `inbox --terminal ` — returns every + // message for a handle regardless of read/delivered state; never touches the + // read bit. Stale-handle safe: if the handle no longer exists, the query + // just returns whatever historical rows remain (§3.3). + getAllMessagesForHandle(toHandle: string, limit = 100): MessageRow[] { + return this.db + .prepare('SELECT * FROM messages WHERE to_handle = ? ORDER BY sequence DESC LIMIT ?') + .all(toHandle, limit) as MessageRow[] + } + // ── Tasks ── createTask(task: { spec: string; deps?: string[]; parentId?: string }): TaskRow { diff --git a/src/main/runtime/orchestration/formatter.test.ts b/src/main/runtime/orchestration/formatter.test.ts index 2b47d40273e..28182e3dbfb 100644 --- a/src/main/runtime/orchestration/formatter.test.ts +++ b/src/main/runtime/orchestration/formatter.test.ts @@ -16,6 +16,7 @@ function makeMessage(overrides: Partial = {}): MessageRow { read: 0, sequence: 1, created_at: '2026-01-01T00:00:00Z', + delivered_at: null, ...overrides } } diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index 1d0ae9a3914..2d4fcd9e8c4 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -30,6 +30,7 @@ export type MessageRow = { read: number sequence: number created_at: string + delivered_at: string | null } export type TaskRow = { diff --git a/src/main/runtime/rpc/core.ts b/src/main/runtime/rpc/core.ts index 9b4bae3d4e3..a91745cc87a 100644 --- a/src/main/runtime/rpc/core.ts +++ b/src/main/runtime/rpc/core.ts @@ -38,6 +38,13 @@ export type RpcRequest = { export type RpcContext = { runtime: OrcaRuntimeService + // Why: long-poll handlers (e.g. orchestration.check with wait=true) need to + // observe the underlying socket's lifetime so they can release their slot + // and resolve their inner waiters immediately when a client disconnects + // instead of running down the configured timeoutMs. Undefined outside the + // runtime-rpc transport (direct in-process callers don't need it). + // See design doc §3.1 counter-lifecycle. + signal?: AbortSignal } export type RpcHandler = (params: TParams, ctx: RpcContext) => Promise | unknown diff --git a/src/main/runtime/rpc/dispatcher.ts b/src/main/runtime/rpc/dispatcher.ts index 105fba0bc55..8d0d1dc5cd3 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -30,7 +30,7 @@ export class RpcDispatcher { this.registry = buildRegistry(methods) } - async dispatch(request: RpcRequest): Promise { + async dispatch(request: RpcRequest, options?: { signal?: AbortSignal }): Promise { const meta = this.meta() const method = this.registry.get(request.method) if (!method) { @@ -55,7 +55,10 @@ export class RpcDispatcher { } try { - const result = await method.handler(parsedParams, { runtime: this.runtime }) + const result = await method.handler(parsedParams, { + runtime: this.runtime, + signal: options?.signal + }) return successResponse(request.id, meta, result) } catch (error) { // Why: browser methods throw BrowserError with a structured `code`; diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts index 64a2507d0f1..7f451656421 100644 --- a/src/main/runtime/rpc/methods/orchestration.test.ts +++ b/src/main/runtime/rpc/methods/orchestration.test.ts @@ -275,6 +275,79 @@ describe('orchestration RPC methods', () => { }) ).rejects.toThrow('Invalid --types') }) + + it('default (unread only) marks returned rows as read', async () => { + setup() + db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + db.insertMessage({ from: 'a', to: 'b', subject: 'two' }) + + const first = (await call('orchestration.check', { terminal: 'b' })) as { + count: number + } + expect(first.count).toBe(2) + + const second = (await call('orchestration.check', { terminal: 'b' })) as { + count: number + } + expect(second.count).toBe(0) + }) + + it('--all returns every message for the handle without marking read', async () => { + setup() + db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + const second = db.insertMessage({ from: 'a', to: 'b', subject: 'two' }) + db.markAsRead([second.id]) + + const result = (await call('orchestration.check', { + terminal: 'b', + all: true + })) as { messages: { read: number }[]; count: number } + + expect(result.count).toBe(2) + // Must not have flipped the remaining unread row + const stillUnread = db.getUnreadMessages('b') + expect(stillUnread).toHaveLength(1) + }) + + it('--all returns rows with delivered_at set after push-on-idle stamped them', async () => { + setup() + const msg = db.insertMessage({ from: 'a', to: 'b', subject: 'hi' }) + // Why: simulate push-on-idle stamping delivered_at without the runtime loop. + db.markAsDelivered([msg.id]) + + const result = (await call('orchestration.check', { + terminal: 'b', + all: true + })) as { messages: { id: string; delivered_at: string | null }[]; count: number } + + expect(result.count).toBe(1) + expect(result.messages[0].delivered_at).not.toBeNull() + }) + + it('--all --terminal returns empty list', async () => { + setup() + db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + + const result = (await call('orchestration.check', { + terminal: 'does_not_exist', + all: true + })) as { count: number } + expect(result.count).toBe(0) + }) + + it('unread:false compat shim behaves like --all (one-release bridge)', async () => { + setup() + db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + + const result = (await call('orchestration.check', { + terminal: 'b', + unread: false + })) as { count: number } + expect(result.count).toBe(1) + + // Must not have marked read + expect(db.getUnreadMessages('b')).toHaveLength(1) + }) }) describe('orchestration.reply', () => { @@ -310,6 +383,38 @@ describe('orchestration RPC methods', () => { const result = (await call('orchestration.inbox', {})) as { count: number } expect(result.count).toBe(2) }) + + it('--terminal matches check --all output for the same handle', async () => { + setup() + db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + db.insertMessage({ from: 'a', to: 'b', subject: 'two' }) + db.insertMessage({ from: 'a', to: 'c', subject: 'other' }) + + const inbox = (await call('orchestration.inbox', { terminal: 'b' })) as { + messages: { id: string; to_handle: string }[] + count: number + } + const check = (await call('orchestration.check', { + terminal: 'b', + all: true + })) as { messages: { id: string; to_handle: string }[]; count: number } + + expect(inbox.count).toBe(2) + expect(check.count).toBe(2) + // Same rows in the same order — both use sequence DESC + expect(inbox.messages.map((m) => m.id)).toEqual(check.messages.map((m) => m.id)) + expect(inbox.messages.every((m) => m.to_handle === 'b')).toBe(true) + }) + + it('--terminal returns empty list without erroring', async () => { + setup() + db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + + const result = (await call('orchestration.inbox', { + terminal: 'does_not_exist' + })) as { count: number } + expect(result.count).toBe(0) + }) }) describe('orchestration.taskCreate', () => { diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index 5c467e369b0..a3376664ef6 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -52,6 +52,10 @@ const SendParams = z.object({ const CheckParams = z.object({ terminal: OptionalString, unread: OptionalBoolean, + // Why: `all` surfaces every message for the handle and skips mark-read. + // Previously the only way to ask for "all" was the hidden RPC trick + // `{unread: false}`. See design doc §3.2 / §3.3. + all: OptionalBoolean, types: OptionalString, inject: OptionalBoolean, wait: OptionalBoolean, @@ -65,7 +69,11 @@ const ReplyParams = z.object({ }) const InboxParams = z.object({ - limit: OptionalFiniteNumber + limit: OptionalFiniteNumber, + // Why: filters the inbox listing to a specific handle so coordinators can + // ask "everything for this handle" with either `inbox` or `check --all` + // and get agreeing results. See design doc §3.3. + terminal: OptionalString }) const TaskCreateParams = z.object({ @@ -177,7 +185,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ defineMethod({ name: 'orchestration.check', params: CheckParams, - handler: async (params, { runtime }) => { + handler: async (params, { runtime, signal }) => { const db = runtime.getOrchestrationDb() const handle = params.terminal ?? 'unknown' const typeFilter = params.types @@ -191,12 +199,17 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ throw new Error(`Invalid --types: ${invalidTypes.join(',')}`) } - const showUnread = params.unread !== false + // Why: `all` short-circuits to "everything for the handle, no marking." + // Explicit `unread: false` is also honored for one release as a compat + // shim so in-flight callers don't break (see design doc §5). Otherwise + // today's behavior is preserved: default is unread-only + mark-read. + const showAll = params.all === true || params.unread === false + const showUnread = !showAll const readAndReturn = () => { const messages = showUnread ? db.getUnreadMessages(handle, typeFilter) - : db.getAllMessages(handle) + : db.getAllMessagesForHandle(handle) if (showUnread && messages.length > 0) { db.markAsRead(messages.map((m) => m.id)) @@ -216,10 +229,15 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ } // Why: blocking wait lets coordinators replace sleep+poll loops with a - // single call that resolves when a message arrives or the timeout expires. + // single call that resolves when a message arrives or the timeout + // expires. The `signal` plumbed from the RPC transport aborts this + // waiter the moment the client socket closes, so a killed client + // releases its long-poll slot immediately rather than after the full + // timeoutMs. See design doc §3.1 counter-lifecycle. await runtime.waitForMessage(handle, { typeFilter: typeFilter as string[] | undefined, - timeoutMs: params.timeoutMs ?? undefined + timeoutMs: params.timeoutMs ?? undefined, + signal }) return readAndReturn() } @@ -255,7 +273,13 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ params: InboxParams, handler: (params, { runtime }) => { const db = runtime.getOrchestrationDb() - const messages = db.getInbox(params.limit) + // Why: when `terminal` is provided, mirror `check --all` output for that + // handle (same rows in the same sequence order). Stale/unknown handles + // return an empty list instead of erroring, matching the "historical + // rows survive handle deletion" rule in design doc §3.3. + const messages = params.terminal + ? db.getAllMessagesForHandle(params.terminal, params.limit) + : db.getInbox(params.limit) return { messages, count: messages.length } } }), diff --git a/src/main/runtime/runtime-rpc.test.ts b/src/main/runtime/runtime-rpc.test.ts index 58fff491f44..8c1c3cc619c 100644 --- a/src/main/runtime/runtime-rpc.test.ts +++ b/src/main/runtime/runtime-rpc.test.ts @@ -2,9 +2,11 @@ import { existsSync, mkdtempSync } from 'fs' import { tmpdir } from 'os' import { join } from 'path' -import { createConnection } from 'net' +import { createConnection, type Socket } from 'net' import { describe, expect, it, vi } from 'vitest' +import Database from 'better-sqlite3' import { OrcaRuntimeService } from './orca-runtime' +import { OrchestrationDb } from './orchestration/db' import * as runtimeMetadataModule from './runtime-metadata' import { readRuntimeMetadata } from './runtime-metadata' import { createRuntimeTransportMetadata, OrcaRuntimeRpcServer } from './runtime-rpc' @@ -46,6 +48,62 @@ async function sendRequest( }) } +// Why: long-poll keepalive tests need every frame, not just the first, because +// we need to count `_keepalive` frames before the terminal success/failure. +// Also exposes the socket so tests can close it mid-wait to exercise the +// long-poll counter decrement path. +type FramedSession = { + socket: Socket + frames: Record[] + done: Promise +} + +function openFramedSession(endpoint: string, request: Record): FramedSession { + const frames: Record[] = [] + const socket = createConnection(endpoint) + let buffer = '' + socket.setEncoding('utf8') + const done = new Promise((resolve, reject) => { + socket.once('error', (err) => { + // Why: ECONNRESET is expected when we deliberately destroy the socket + // mid-wait to probe the counter decrement; surface other errors. + if ((err as NodeJS.ErrnoException).code === 'ECONNRESET') { + resolve() + return + } + reject(err) + }) + socket.on('close', () => resolve()) + socket.on('data', (chunk: string) => { + buffer += chunk + let newlineIndex = buffer.indexOf('\n') + while (newlineIndex !== -1) { + const raw = buffer.slice(0, newlineIndex).trim() + buffer = buffer.slice(newlineIndex + 1) + if (raw) { + const frame = JSON.parse(raw) as Record + frames.push(frame) + // Why: the server leaves the socket open after writing the terminal + // frame (short RPCs expect the client to close); close the client + // side so `done` resolves once we've captured the response. + if (frame._keepalive !== true) { + socket.end() + } + } + newlineIndex = buffer.indexOf('\n') + } + }) + socket.on('connect', () => { + socket.write(`${JSON.stringify(request)}\n`) + }) + }) + return { socket, frames, done } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + describe('OrcaRuntimeRpcServer', () => { const makeStore = (overrides?: { isUnread?: boolean }) => ({ getRepo: (id: string) => @@ -502,4 +560,209 @@ describe('OrcaRuntimeRpcServer', () => { await server.stop() }) + + // Why: §6 tests for the transport keepalive + long-poll counter path in §3.1. + // Exercise the real socket (not a mock) so we catch buffer/flush regressions + // that a unit-level test would miss. + describe('long-poll transport (§3.1)', () => { + it('emits keepalive frames while a check --wait handler blocks', async () => { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-')) + const runtime = new OrcaRuntimeService() + const db = new OrchestrationDb(':memory:') + runtime.setOrchestrationDb(db) + // Why: 50ms keepalive lets us collect ≥3 frames within a 300ms wait + // window without slowing the suite. + const server = new OrcaRuntimeRpcServer({ + runtime, + userDataPath, + keepaliveIntervalMs: 50 + }) + await server.start() + + try { + const metadata = readRuntimeMetadata(userDataPath) + const session = openFramedSession(metadata!.transport!.endpoint, { + id: 'req_wait', + authToken: metadata!.authToken, + method: 'orchestration.check', + params: { + terminal: 'term_nobody', + wait: true, + timeoutMs: 300 + } + }) + await session.done + + const keepalives = session.frames.filter((f) => f._keepalive === true) + const terminals = session.frames.filter((f) => f.ok !== undefined) + expect(terminals).toHaveLength(1) + expect(terminals[0]).toMatchObject({ id: 'req_wait', ok: true }) + // Why: 300ms wait with 50ms keepalive → expect roughly 5 keepalives; + // assert ≥3 to tolerate scheduler jitter without flaking. + expect(keepalives.length).toBeGreaterThanOrEqual(3) + } finally { + db.close() + await server.stop() + } + }) + + it('releases long-poll slot when client closes mid-wait', async () => { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-')) + const runtime = new OrcaRuntimeService() + const db = new OrchestrationDb(':memory:') + runtime.setOrchestrationDb(db) + const server = new OrcaRuntimeRpcServer({ + runtime, + userDataPath, + keepaliveIntervalMs: 1000, + longPollCap: 2 + }) + await server.start() + + try { + const metadata = readRuntimeMetadata(userDataPath) + const endpoint = metadata!.transport!.endpoint + + // Fill the cap with two long waits (10s each — we'll kill them). + const a = openFramedSession(endpoint, { + id: 'req_a', + authToken: metadata!.authToken, + method: 'orchestration.check', + params: { terminal: 'term_a', wait: true, timeoutMs: 10_000 } + }) + const b = openFramedSession(endpoint, { + id: 'req_b', + authToken: metadata!.authToken, + method: 'orchestration.check', + params: { terminal: 'term_b', wait: true, timeoutMs: 10_000 } + }) + // Let the two waits land in the handler and increment the counter. + await sleep(100) + expect(server['activeLongPolls']).toBe(2) + + // Kill one client mid-wait; counter must drop to 1. + a.socket.destroy() + await a.done + // Give Node one tick to fire the close event on the server socket. + await sleep(50) + expect(server['activeLongPolls']).toBe(1) + + // The freed slot must admit a new long-poll immediately. + const c = openFramedSession(endpoint, { + id: 'req_c', + authToken: metadata!.authToken, + method: 'orchestration.check', + params: { terminal: 'term_c', wait: true, timeoutMs: 100 } + }) + await c.done + const cTerminal = c.frames.find((f) => f.ok !== undefined) + expect(cTerminal).toMatchObject({ ok: true, id: 'req_c' }) + + b.socket.destroy() + await b.done + } finally { + db.close() + await server.stop() + } + }) + + it('responds runtime_busy once the long-poll cap is saturated', async () => { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-')) + const runtime = new OrcaRuntimeService() + const db = new OrchestrationDb(':memory:') + runtime.setOrchestrationDb(db) + const server = new OrcaRuntimeRpcServer({ + runtime, + userDataPath, + keepaliveIntervalMs: 1000, + longPollCap: 1 + }) + await server.start() + + try { + const metadata = readRuntimeMetadata(userDataPath) + const endpoint = metadata!.transport!.endpoint + + const a = openFramedSession(endpoint, { + id: 'req_a', + authToken: metadata!.authToken, + method: 'orchestration.check', + params: { terminal: 'term_a', wait: true, timeoutMs: 5_000 } + }) + await sleep(100) + expect(server['activeLongPolls']).toBe(1) + + // Second long-poll overflows the cap → runtime_busy. + const overflow = await sendRequest(endpoint, { + id: 'req_overflow', + authToken: metadata!.authToken, + method: 'orchestration.check', + params: { terminal: 'term_b', wait: true, timeoutMs: 5_000 } + }) + expect(overflow).toMatchObject({ + id: 'req_overflow', + ok: false, + error: { code: 'runtime_busy' } + }) + // The failing request must not have counted against the cap. + expect(server['activeLongPolls']).toBe(1) + + // Short RPCs still succeed even when the long-poll cap is full. + const short = await sendRequest(endpoint, { + id: 'req_short', + authToken: metadata!.authToken, + method: 'status.get' + }) + expect(short).toMatchObject({ id: 'req_short', ok: true }) + + a.socket.destroy() + await a.done + } finally { + db.close() + await server.stop() + } + }) + }) + + // Why: §6 test for the idempotent + hard-fail schema migration. A broken + // migration must crash startup loudly rather than serve traffic against a + // schema missing the delivered_at column. + describe('orchestration DB migration (§3.2)', () => { + it('is idempotent when delivered_at already exists', () => { + // First open creates the column; second open should be a no-op. + const db1 = new OrchestrationDb(':memory:') + db1.close() + // File path reuse is meaningless with :memory:, so use a tmp file. + const tmpPath = join(mkdtempSync(join(tmpdir(), 'orca-orch-mig-')), 'orch.sqlite') + const a = new OrchestrationDb(tmpPath) + a.close() + // Second construction must not throw "duplicate column name". + expect(() => { + const b = new OrchestrationDb(tmpPath) + b.close() + }).not.toThrow() + }) + + it('hard-fails startup when the migration cannot be applied', () => { + // Simulate a migration error by monkey-patching better-sqlite3's exec. + // If ALTER TABLE throws for any reason (e.g. disk full, permissions), + // the constructor must propagate — not swallow and serve half-broken. + const tmpPath = join(mkdtempSync(join(tmpdir(), 'orca-orch-mig-')), 'orch.sqlite') + const realPrototype = Database.prototype as unknown as { + exec: (sql: string) => unknown + } + const originalExec = realPrototype.exec + realPrototype.exec = function (sql: string) { + if (sql.includes('ALTER TABLE messages ADD COLUMN delivered_at')) { + throw new Error('simulated migration failure') + } + return originalExec.call(this, sql) + } + try { + expect(() => new OrchestrationDb(tmpPath)).toThrow('simulated migration failure') + } finally { + realPrototype.exec = originalExec + } + }) + }) }) diff --git a/src/main/runtime/runtime-rpc.ts b/src/main/runtime/runtime-rpc.ts index 8ebb8778736..70dbd4af1f3 100644 --- a/src/main/runtime/runtime-rpc.ts +++ b/src/main/runtime/runtime-rpc.ts @@ -1,3 +1,4 @@ +/* eslint-disable max-lines -- Why: this file is the single security boundary for the bundled CLI — transport setup, auth-token enforcement, admission control, keepalive framing, and orphan-socket sweeping all co-locate deliberately so a reviewer can audit the boundary in one sitting. Splitting this across files would scatter the invariants without reducing complexity. */ // Why: this is the single security boundary for the bundled CLI. It owns // transport setup (unix socket / named pipe), auth-token enforcement, and // bootstrap-metadata publication so a running runtime is always discoverable @@ -19,12 +20,49 @@ type OrcaRuntimeRpcServerOptions = { userDataPath: string pid?: number platform?: NodeJS.Platform + // Why: test-only overrides for the two time-bound constants below. + // Production callers must not pass these — defaults are set by the design + // doc (§3.1) and changing them in production would weaken the admission + // fence or flood the socket with keepalive frames. + keepaliveIntervalMs?: number + longPollCap?: number } const MAX_RUNTIME_RPC_MESSAGE_BYTES = 1024 * 1024 const RUNTIME_RPC_SOCKET_IDLE_TIMEOUT_MS = 30_000 const MAX_RUNTIME_RPC_CONNECTIONS = 32 +// Why: after 10 s of a pending dispatch we emit a tiny `{"_keepalive":true}` +// frame every 10 s until the handler resolves. Each write resets both the +// server's own socket idle timer (30 s) and — once §3.1 ships on the client — +// the client's idle timer, because any byte counts as socket activity. This +// is the transport-layer fix for feedback #1: long-poll RPCs (i.e. +// orchestration.check --wait) can now run past the 30 s/60 s idle caps +// without either end tearing the socket down. See design doc §3.1. +const KEEPALIVE_INTERVAL_MS = 10_000 + +// Why: long-poll slot cap. With keepalives a `check --wait --timeout-ms +// 600000` can hold a connection for up to 10 minutes; unbounded that would +// saturate MAX_RUNTIME_RPC_CONNECTIONS (32) with 32 waiting coordinators +// and lock out normal short RPCs. Capping at half the connection budget +// leaves the other half for short traffic. On overflow the server responds +// immediately with `runtime_busy` (CLI exit 75) — fail fast, not silent +// queuing. See design doc §3.1 + §7 risk #2. +const LONG_POLL_CAP = 16 + +// Why: a long-poll request is one whose handler blocks for an unbounded +// amount of time waiting for an external event (today, only +// `orchestration.check` with `wait === true`). This function is the single +// place that classifies it — the long-poll counter, abort wiring, and +// runtime_busy admission check all share this decision. See §3.1. +function isLongPollRequest(request: RpcRequest): boolean { + if (request.method !== 'orchestration.check') { + return false + } + const params = request.params as { wait?: unknown } | undefined + return params?.wait === true +} + export class OrcaRuntimeRpcServer { private readonly runtime: OrcaRuntimeService private readonly dispatcher: RpcDispatcher @@ -32,20 +70,30 @@ export class OrcaRuntimeRpcServer { private readonly pid: number private readonly platform: NodeJS.Platform private readonly authToken = randomBytes(24).toString('hex') + private readonly keepaliveIntervalMs: number + private readonly longPollCap: number private server: Server | null = null private transport: RuntimeTransportMetadata | null = null + // Why: separate from Node's server.maxConnections because we need to count + // only long-running dispatches, not every in-flight short RPC. See §3.1 + + // §7 risk #2. + private activeLongPolls = 0 constructor({ runtime, userDataPath, pid = process.pid, - platform = process.platform + platform = process.platform, + keepaliveIntervalMs = KEEPALIVE_INTERVAL_MS, + longPollCap = LONG_POLL_CAP }: OrcaRuntimeRpcServerOptions) { this.runtime = runtime this.dispatcher = new RpcDispatcher({ runtime }) this.userDataPath = userDataPath this.pid = pid this.platform = platform + this.keepaliveIntervalMs = keepaliveIntervalMs + this.longPollCap = longPollCap } async start(): Promise { @@ -174,37 +222,129 @@ export class OrcaRuntimeRpcServer { const rawMessage = buffer.slice(0, newlineIndex).trim() buffer = buffer.slice(newlineIndex + 1) if (rawMessage) { - void this.handleMessage(rawMessage).then((response) => { - socket.write(`${JSON.stringify(response)}\n`) - }) + void this.handleRequest(socket, rawMessage) } newlineIndex = buffer.indexOf('\n') } }) } - private async handleMessage(rawMessage: string): Promise { + // Why: a single entry point per inbound request so keepalive + long-poll + // admission + AbortController wiring + response write all live in one + // place. See design doc §3.1. + private async handleRequest(socket: Socket, rawMessage: string): Promise { + const parsed = this.parseAndAuth(rawMessage) + if ('error' in parsed) { + this.safeWrite(socket, `${JSON.stringify(parsed.error)}\n`) + return + } + const request = parsed.request + + // Why: long-poll admission fence. Short RPCs bypass the counter entirely + // — it only guards handlers that can block for minutes. See §7 risk #2. + const longPoll = isLongPollRequest(request) + if (longPoll) { + if (this.activeLongPolls >= this.longPollCap) { + const busy = this.buildError( + request.id, + 'runtime_busy', + 'long-poll capacity reached; retry with backoff' + ) + this.safeWrite(socket, `${JSON.stringify(busy)}\n`) + socket.end() + return + } + this.activeLongPolls += 1 + } + + // Why: `decremented` must guard against double-decrement when both + // `close` and a post-resolve cleanup path fire. `socket.on('close')` is + // the only path that fires for every termination (normal end, destroy, + // idle timer, client kill -9, OS reset), so it carries the decrement. + // Tying it to `.finally` alone would leak a slot any time a client dies + // mid-wait because the inner waitForMessage can keep counting down for + // minutes after the socket is gone. See §3.1 counter-lifecycle. + let decremented = !longPoll + const abortController = new AbortController() + const onClose = (): void => { + if (!decremented) { + decremented = true + this.activeLongPolls = Math.max(0, this.activeLongPolls - 1) + } + abortController.abort() + } + socket.on('close', onClose) + + // Why: for long-poll requests we start a keepalive ticker after 10 s. The + // first frame at 10 s resets both the server's 30 s idle timer and the + // client's configured timeout. Short RPCs never see a keepalive — the + // ticker never fires because the handler resolves first. + let keepaliveTimer: NodeJS.Timeout | null = null + if (longPoll) { + keepaliveTimer = setInterval(() => { + if (socket.writable && !socket.destroyed) { + socket.write('{"_keepalive":true}\n') + } + }, this.keepaliveIntervalMs) + // Why: don't hold the process open solely on the keepalive interval — + // .unref() lets the event loop exit when nothing else is pending. + if (typeof keepaliveTimer.unref === 'function') { + keepaliveTimer.unref() + } + } + + try { + const response = await this.dispatcher.dispatch(request, { + signal: longPoll ? abortController.signal : undefined + }) + if (!socket.destroyed) { + this.safeWrite(socket, `${JSON.stringify(response)}\n`) + } + } finally { + if (keepaliveTimer) { + clearInterval(keepaliveTimer) + } + // Why: the close-handler path is still what decrements the counter (the + // socket may be closed by the client before the response write flushes). + // We don't remove the listener here — `once` semantics are handled by + // the boolean guard. + } + } + + private parseAndAuth(rawMessage: string): { request: RpcRequest } | { error: RpcResponse } { let request: RpcRequest try { request = JSON.parse(rawMessage) as RpcRequest } catch { - return this.buildError('unknown', 'bad_request', 'Invalid JSON request') + return { error: this.buildError('unknown', 'bad_request', 'Invalid JSON request') } } if (typeof request.id !== 'string' || request.id.length === 0) { - return this.buildError('unknown', 'bad_request', 'Missing request id') + return { error: this.buildError('unknown', 'bad_request', 'Missing request id') } } if (typeof request.method !== 'string' || request.method.length === 0) { - return this.buildError(request.id, 'bad_request', 'Missing RPC method') + return { error: this.buildError(request.id, 'bad_request', 'Missing RPC method') } } if (typeof request.authToken !== 'string' || request.authToken.length === 0) { - return this.buildError(request.id, 'unauthorized', 'Missing auth token') + return { error: this.buildError(request.id, 'unauthorized', 'Missing auth token') } } if (request.authToken !== this.authToken) { - return this.buildError(request.id, 'unauthorized', 'Invalid auth token') + return { error: this.buildError(request.id, 'unauthorized', 'Invalid auth token') } } - return this.dispatcher.dispatch(request) + return { request } + } + + private safeWrite(socket: Socket, payload: string): void { + if (socket.destroyed || !socket.writable) { + return + } + try { + socket.write(payload) + } catch { + // Socket was closed in between the writable check and the write — + // nothing we can do; the client already disconnected. + } } private buildError(id: string, code: string, message: string): RpcResponse {