diff --git a/packages/cli/src/__tests__/cli.test.ts b/packages/cli/src/__tests__/cli.test.ts index 0d5e917949..97c8a82ad3 100644 --- a/packages/cli/src/__tests__/cli.test.ts +++ b/packages/cli/src/__tests__/cli.test.ts @@ -13,6 +13,19 @@ describe('Maka CLI args', () => { { kind: 'eval', args: ['task-run', 'inspect', 'run-1'] }, ], [['inspect', 'run-1', '--json'], { kind: 'inspect', args: ['run-1', '--json'] }], + [['runtime-host', 'serve'], { kind: 'runtime-host-serve' }], + [ + ['runtime-host', 'serve', '--root', '/srv/maka'], + { kind: 'runtime-host-serve', rootPath: '/srv/maka' }, + ], + [ + ['runtime-host'], + { kind: 'error', message: 'runtime-host requires the serve command', exitCode: 2 }, + ], + [ + ['runtime-host', 'serve', '--root'], + { kind: 'error', message: '--root requires a directory', exitCode: 2 }, + ], [['run', 'hello', '--max-steps', '3'], { kind: 'run', args: ['hello', '--max-steps', '3'] }], [['-p', 'hello', '--max-steps', '3'], { kind: 'run', args: ['hello', '--max-steps', '3'] }], [['--version'], { kind: 'version', text: '0.1.0' }], diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index ca7d3b1ceb..67f0b14059 100644 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -11,6 +11,7 @@ export type MakaCliCommand = | { kind: 'activate'; args: string[] } | { kind: 'eval'; args: string[] } | { kind: 'inspect'; args: string[] } + | { kind: 'runtime-host-serve'; rootPath?: string } | { kind: 'help'; text: string } | { kind: 'version'; text: string } | { kind: 'error'; message: string; exitCode: number }; @@ -43,6 +44,7 @@ export function parseMakaCliArgs(argv: string[], version: string): MakaCliComman if (first === 'activate') return { kind: 'activate', args: argv.slice(1) }; if (first === 'eval') return { kind: 'eval', args: argv.slice(1) }; if (first === 'inspect') return { kind: 'inspect', args: argv.slice(1) }; + if (first === 'runtime-host') return parseRuntimeHostCommand(argv.slice(1)); return { kind: 'error', message: `Unexpected argument: ${first ?? ''}`, @@ -96,6 +98,7 @@ function helpText(): string { ' maka -p ... Alias for maka run', ' maka eval ... Run evaluation and autonomous task commands', ' maka inspect ... Inspect Session, AgentRun, or TaskRun evidence', + ' maka runtime-host serve [--root ] Run a local Runtime Host service', '', 'Options:', ' -h, --help Show help', @@ -130,6 +133,10 @@ export async function runMakaCli(argv: string[] = process.argv.slice(2)): Promis const { runMakaInspectCli } = await import('./inspect-command.js'); return runMakaInspectCli(command.args); } + case 'runtime-host-serve': { + const { runRuntimeHostServiceCli } = await import('./runtime-host-service-command.js'); + return runRuntimeHostServiceCli(command.rootPath ?? resolveMakaWorkspaceRoot()); + } case 'help': process.stdout.write(`${command.text}\n`); return 0; @@ -153,6 +160,30 @@ export async function runMakaCli(argv: string[] = process.argv.slice(2)): Promis } } +function parseRuntimeHostCommand(argv: string[]): MakaCliCommand { + if (argv[0] !== 'serve') { + return { + kind: 'error', + message: argv[0] + ? `Unexpected runtime-host command: ${argv[0]}` + : 'runtime-host requires the serve command', + exitCode: 2, + }; + } + if (argv[1] === undefined) return { kind: 'runtime-host-serve' }; + if (argv[1] !== '--root') { + return { kind: 'error', message: `Unexpected argument: ${argv[1]}`, exitCode: 2 }; + } + const rootPath = argv[2]; + if (!rootPath || rootPath.startsWith('-')) { + return { kind: 'error', message: '--root requires a directory', exitCode: 2 }; + } + if (argv[3] !== undefined) { + return { kind: 'error', message: `Unexpected argument: ${argv[3]}`, exitCode: 2 }; + } + return { kind: 'runtime-host-serve', rootPath }; +} + async function readPackageVersion(): Promise { const raw = await readFile(new URL('../package.json', import.meta.url), 'utf8'); const parsed = JSON.parse(raw) as { version?: unknown }; diff --git a/packages/cli/src/runtime-host-service-command.ts b/packages/cli/src/runtime-host-service-command.ts new file mode 100644 index 0000000000..73483383b6 --- /dev/null +++ b/packages/cli/src/runtime-host-service-command.ts @@ -0,0 +1,14 @@ +import { + installRuntimeHostLogCapture, + runRuntimeHostProcessLifecycle, + startExecutionRuntimeHostService, +} from '@maka/runtime-host/server'; + +export async function runRuntimeHostServiceCli(rootPath: string): Promise { + installRuntimeHostLogCapture(); + const host = await startExecutionRuntimeHostService({ rootPath }); + await runRuntimeHostProcessLifecycle(host, { + onReady: () => process.stdout.write(`Runtime Host service is ready at ${host.endpoint}\n`), + }); + return 0; +} diff --git a/packages/runtime-host/src/__tests__/artifact-protocol.test.ts b/packages/runtime-host/src/__tests__/artifact-protocol.test.ts index 139374ebe5..fd601d68b7 100644 --- a/packages/runtime-host/src/__tests__/artifact-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/artifact-protocol.test.ts @@ -13,9 +13,9 @@ import { decodeClientFrame, decodeHostFrame, encodeArtifactQueryResult, - encodeProtocolFrame, + encodeProtocolMessage, HOST_OPERATION_SPECS, - RUNTIME_HOST_MAX_FRAME_BYTES, + RUNTIME_HOST_MAX_MESSAGE_BYTES, RuntimeHostProtocolError, } from '../protocol/index.js'; import { encodeArtifactProjection } from '../protocol/artifact.js'; @@ -66,7 +66,7 @@ describe('Artifact protocol', () => { ); }); - test('bounds sequential Artifact read chunks below the frame limit', () => { + test('bounds sequential Artifact read chunks below the message limit', () => { const bytes = Buffer.alloc(ARTIFACT_READ_CHUNK_MAX_BYTES, 9); assert.doesNotThrow(() => response('artifact.query', { @@ -107,7 +107,7 @@ describe('Artifact protocol', () => { ); }); - test('bounds chunked attachment publication below the frame limit', () => { + test('bounds chunked attachment publication below the message limit', () => { const bytes = Buffer.alloc(ARTIFACT_INGEST_CHUNK_MAX_BYTES, 7); const digest = `sha256:${'a'.repeat(64)}`; assert.doesNotThrow(() => @@ -169,7 +169,7 @@ describe('Artifact protocol', () => { uploadId: 'upload-1', }), ); - const frame = encodeProtocolFrame({ + const frame = encodeProtocolMessage({ requestId: 'artifact-ingest-chunk', operation: 'artifact.ingest', input: { @@ -180,7 +180,7 @@ describe('Artifact protocol', () => { chunkBase64: bytes.toString('base64'), }, }); - assert.ok(frame.byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES); + assert.ok(frame.byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES); for (const chunkBase64 of [ Buffer.alloc(ARTIFACT_INGEST_CHUNK_MAX_BYTES + 1).toString('base64'), @@ -368,12 +368,12 @@ describe('Artifact protocol', () => { Buffer.byteLength(JSON.stringify(maximumBinary), 'utf8') <= ARTIFACT_RESULT_MAX_BYTES, ); assert.ok( - encodeProtocolFrame({ + encodeProtocolMessage({ requestId: 'artifact-binary', operation: 'artifact.query', ok: true, result: maximumBinary, - }).byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES, + }).byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES, ); }); diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index 9772b0b71a..95f1565a6d 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -13,14 +13,17 @@ import { readHostRegistration } from '../control/registration.js'; import { connectRuntimeHost, type RuntimeHostConnection } from '../client/index.js'; import { decodeHostFrame, + encodeProtocolMessage, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_MAX_IN_FLIGHT_DOMAIN_REQUESTS, RUNTIME_HOST_PROTOCOL_VERSION, + type ClientFrame, type HostFrame, type ResponseFrame, type TurnSnapshot, } from '../protocol/index.js'; import { RuntimeHostKernel, type RuntimeHostComposition } from '../server/index.js'; +import { LOCAL_OWNER_CONNECTION_AUTHORITY } from '../server/connection-authority.js'; import { RuntimeHostConnectionSession } from '../server/connection-session.js'; import { createUnavailableDomainOperationHandlers, @@ -42,6 +45,16 @@ const CURRENT_PROTOCOL = { max: RUNTIME_HOST_PROTOCOL_VERSION, } as const; +function acceptedConnection(connectionId: string) { + return { + hostEpoch: 'host-epoch', + connectionId, + clientInstanceId: 'test-client', + surface: 'tui' as const, + authority: LOCAL_OWNER_CONNECTION_AUTHORITY, + }; +} + type TurnQueryHandler = RuntimeHostComposition['handlers']['turn.query']; test('concurrent responses remain framed and correlated in reverse completion order', async () => { @@ -162,7 +175,7 @@ test('serial outbound writer fails once when its real transport is closed', asyn failureCalls += 1; }); try { - pair.clientTransport.destroy(); + pair.clientTransport.abort(); await pair.clientTransport.closed; const receipt = writer.enqueue(statusResponse('closed-transport')); await assert.rejects(receipt.flushed); @@ -236,7 +249,7 @@ test('clean read EOF drains an already dispatched response before closing', asyn test('a fatal transport close during clean EOF drain tears down exactly once', async () => { const fixture = await openHalfClosedDispatchedSession('fatal-close-after-eof'); try { - fixture.pair.serverTransport.destroy(new Error('forced transport failure')); + fixture.pair.serverTransport.abort(new Error('forced transport failure')); await withTimeout( fixture.teardownObserved.promise, 1_000, @@ -289,7 +302,7 @@ test('a connection accepted before composition exists resolves ready handlers wi assert.equal(registration.state, 'recovering'); transport = await openAcceptedTransport(registration.endpoint, 'pre-ready-client'); - await transport.write({ + await writeProtocolFrame(transport, { requestId: 'before-ready', operation: 'turn.query', input: { sessionId: 'session', turnId: 'turn' }, @@ -301,7 +314,7 @@ test('a connection accepted before composition exists resolves ready handlers wi releaseFactory.resolve(); host = await withTimeout(hostTask, 1_000, 'Runtime Host did not become ready'); - await transport.write({ + await writeProtocolFrame(transport, { requestId: 'after-ready', operation: 'turn.query', input: { sessionId: 'session', turnId: 'turn' }, @@ -314,7 +327,7 @@ test('a connection accepted before composition exists resolves ready handlers wi assert.equal(afterReady.result.runId, 'run-turn'); } finally { releaseFactory.resolve(); - transport?.destroy(); + transport?.abort(); host ??= await hostTask.catch(() => undefined); await host?.close().catch(() => undefined); await rm(join(resolveRootControlNamespace(), capability.rootId), { @@ -354,12 +367,7 @@ test('connection reset while operation admission is pending does not execute the }; const session = new RuntimeHostConnectionSession({ transport: pair.serverTransport, - connection: { - hostEpoch: 'host-epoch', - connectionId: 'pending-admission', - surface: 'tui', - principal: 'local_os_user', - }, + connection: acceptedConnection('pending-admission'), resolveHandlers: () => handlers, resolveContinuity: () => undefined, beginOperation: async () => { @@ -377,7 +385,7 @@ test('connection reset while operation admission is pending does not execute the }); const run = session.run(); try { - await pair.clientTransport.write({ + await writeProtocolFrame(pair.clientTransport, { requestId: 'pending-request', operation: 'turn.query', input: { sessionId: 'session', turnId: 'turn' }, @@ -395,7 +403,7 @@ test('connection reset while operation admission is pending does not execute the assert.equal(finishCalls, 1); } finally { releaseAdmission.resolve(); - pair.clientTransport.destroy(); + pair.clientTransport.abort(); await Promise.allSettled([run, pair.close()]); } }); @@ -471,13 +479,13 @@ test('a duplicate active request id tears down only the offending connection', a async ({ connectClient, endpoint }) => { const transport = await openAcceptedTransport(endpoint, 'duplicate-request-client'); try { - await transport.write({ + await writeProtocolFrame(transport, { requestId: 'duplicate-request', operation: 'turn.query', input: { sessionId: 'session', turnId: 'first' }, }); await withTimeout(handlerEntered.promise, 1_000, 'first request was not admitted'); - await transport.write({ + await writeProtocolFrame(transport, { requestId: 'duplicate-request', operation: 'turn.query', input: { sessionId: 'session', turnId: 'second' }, @@ -490,7 +498,7 @@ test('a duplicate active request id tears down only the offending connection', a assert.equal(handlerCalls, 1); } finally { releaseHandler.resolve(); - transport.destroy(); + transport.abort(); } const observer = await connectClient(); @@ -533,7 +541,7 @@ test('reserves liveness status at the domain request limit and rejects another d value.activeOperations === 65 && value.activeResidencies === 0, ); - await transport.write({ + await writeProtocolFrame(transport, { requestId: 'overflow-status', operation: 'host.status', input: {}, @@ -546,7 +554,7 @@ test('reserves liveness status at the domain request limit and rejects another d assert.equal(statusResponse.ok, true); } await new Promise((resolve) => setImmediate(resolve)); - await transport.write({ + await writeProtocolFrame(transport, { requestId: 'overflow-64', operation: 'turn.query', input: { sessionId: 'session', turnId: 'turn-64' }, @@ -558,7 +566,7 @@ test('reserves liveness status at the domain request limit and rejects another d ); } finally { releaseHandlers.resolve(); - transport.destroy(); + transport.abort(); } const status = await waitForStatus( observer, @@ -604,12 +612,7 @@ test('an in-flight status does not consume the final domain request slot', async }; const session = new RuntimeHostConnectionSession({ transport: pair.serverTransport, - connection: { - hostEpoch: 'host-epoch', - connectionId: 'status-before-final-domain-client', - surface: 'tui', - principal: 'local_os_user', - }, + connection: acceptedConnection('status-before-final-domain-client'), resolveHandlers: () => handlers, resolveContinuity: () => undefined, beginOperation: async () => ({ @@ -634,13 +637,13 @@ test('an in-flight status does not consume the final domain request slot', async 1_000, 'initial domain handlers were not admitted', ); - await pair.clientTransport.write({ + await writeProtocolFrame(pair.clientTransport, { requestId: 'status-first-probe', operation: 'host.status', input: {}, }); await withTimeout(statusEntered.promise, 1_000, 'status handler was not admitted'); - await pair.clientTransport.write({ + await writeProtocolFrame(pair.clientTransport, { requestId: 'status-first-63', operation: 'turn.query', input: { sessionId: 'session', turnId: 'turn-63' }, @@ -655,7 +658,7 @@ test('an in-flight status does not consume the final domain request slot', async assert.equal(response.operation, 'host.status'); assert.equal(response.ok, true); } - await pair.clientTransport.write({ + await writeProtocolFrame(pair.clientTransport, { requestId: 'status-first-overflow', operation: 'turn.query', input: { sessionId: 'session', turnId: 'turn-64' }, @@ -668,7 +671,7 @@ test('an in-flight status does not consume the final domain request slot', async } finally { releaseStatus.resolve(); releaseDomains.resolve(); - pair.clientTransport.destroy(); + pair.clientTransport.abort(); await Promise.allSettled([run, pair.close()]); } }); @@ -700,12 +703,7 @@ test('evicting one slow subscription keeps sibling subscriptions and requests us }; const session = new RuntimeHostConnectionSession({ transport: pair.serverTransport, - connection: { - hostEpoch: 'host-epoch', - connectionId: 'shared-subscription-connection', - surface: 'tui', - principal: 'local_os_user', - }, + connection: acceptedConnection('shared-subscription-connection'), resolveHandlers: () => handlers, resolveContinuity: () => coordinator, beginOperation: async () => ({ @@ -718,13 +716,13 @@ test('evicting one slow subscription keeps sibling subscriptions and requests us const run = session.run(); const slow = await openSubscription(pair.clientTransport, 'slow-session', 'open-slow'); const sibling = await openSubscription(pair.clientTransport, 'sibling-session', 'open-sibling'); - const originalWrite = pair.serverTransport.writeEncoded.bind(pair.serverTransport); + const originalWrite = pair.serverTransport.write.bind(pair.serverTransport); const writeBlocked = deferred(); const releaseWrite = deferred(); - pair.serverTransport.writeEncoded = async (encoded) => { + pair.serverTransport.write = async (message) => { writeBlocked.resolve(); await releaseWrite.promise; - return originalWrite(encoded); + return originalWrite(message); }; try { @@ -743,7 +741,7 @@ test('evicting one slow subscription keeps sibling subscriptions and requests us 'run-sibling-session', connectionTextEvent('sibling-session', 1), ); - await pair.clientTransport.write({ + await writeProtocolFrame(pair.clientTransport, { requestId: 'status-after-eviction', operation: 'host.status', input: {}, @@ -782,8 +780,8 @@ test('evicting one slow subscription keeps sibling subscriptions and requests us assert.equal(pair.serverTransport.socket.destroyed, false); } finally { releaseWrite.resolve(); - pair.serverTransport.writeEncoded = originalWrite; - pair.clientTransport.destroy(); + pair.serverTransport.write = originalWrite; + pair.clientTransport.abort(); await Promise.allSettled([run, pair.close()]); coordinator.close(); } @@ -852,7 +850,7 @@ async function openAcceptedTransport( socket.once('error', reject); }); const transport = new FramedTransport(socket); - await transport.write({ + await writeProtocolFrame(transport, { kind: 'hello', clientInstanceId, surface: 'tui', @@ -899,8 +897,8 @@ async function openTransportPair(): Promise { clientTransport, serverTransport, close: async () => { - clientTransport.destroy(); - serverTransport.destroy(); + clientTransport.abort(); + serverTransport.abort(); await Promise.all([clientTransport.closed, serverTransport.closed]); await closeServer(listener); }, @@ -917,12 +915,7 @@ async function openHalfClosedDispatchedSession( let teardownCalls = 0; const session = new RuntimeHostConnectionSession({ transport: pair.serverTransport, - connection: { - hostEpoch: 'host-epoch', - connectionId: `${turnId}-client`, - surface: 'tui', - principal: 'local_os_user', - }, + connection: acceptedConnection(`${turnId}-client`), resolveHandlers: () => ({ 'host.status': async () => ({ ok: true, @@ -957,7 +950,7 @@ async function openHalfClosedDispatchedSession( }); const run = session.run(); try { - await pair.clientTransport.write({ + await writeProtocolFrame(pair.clientTransport, { requestId: `${turnId}-request`, operation: 'turn.query', input: { sessionId: 'session', turnId }, @@ -974,13 +967,13 @@ async function openHalfClosedDispatchedSession( teardownCalls: () => teardownCalls, close: async () => { releaseHandler.resolve(); - pair.clientTransport.destroy(); + pair.clientTransport.abort(); await Promise.allSettled([run, pair.close()]); }, }; } catch (error) { releaseHandler.resolve(); - pair.clientTransport.destroy(); + pair.clientTransport.abort(); await Promise.allSettled([run, pair.close()]); throw error; } @@ -1106,7 +1099,7 @@ function runningSnapshot(sessionId: string, turnId: string): TurnSnapshot { } async function openSubscription(transport: FramedTransport, sessionId: string, requestId: string) { - await transport.write({ + await writeProtocolFrame(transport, { requestId, operation: 'subscription.open', input: { sessionId }, @@ -1118,6 +1111,13 @@ async function openSubscription(transport: FramedTransport, sessionId: string, r return response.result; } +function writeProtocolFrame( + transport: FramedTransport, + frame: ClientFrame | HostFrame, +): Promise { + return transport.write(encodeProtocolMessage(frame)); +} + function canonicalProjection(sessionId: string): CanonicalSessionProjection { return { session: { diff --git a/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts b/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts index c0f7e5883c..0378c98239 100644 --- a/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts @@ -104,7 +104,7 @@ test('retry after a discarded turn.start response reuses the durable semantic ad }); const observer = await connectClient(fixture.root, 'tui'); const committed = await waitForTurn(observer, fixture.sessionId, turnId); - dropped.destroy(); + dropped.abort(); const retried = requireStartedTurn( await observer.startTurn({ diff --git a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts index eaaa34a772..3ca8ce2bf4 100644 --- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts +++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts @@ -61,9 +61,11 @@ import { } from '../../client/index.js'; import { decodeHostFrame, + encodeProtocolMessage, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, TASK_LEDGER_PAGE_MAX_ITEMS, + type ClientFrame, type ConnectionCatalogQueryResult, type InteractionPendingSnapshot, type SubscriptionFrame, @@ -1118,7 +1120,7 @@ export async function sendStartWithoutReadingResponse( input: { sessionId: string; turnId: string; text: string }, ): Promise { const transport = new FramedTransport(await openSocket(endpoint)); - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: randomUUID(), surface: 'desktop', @@ -1129,7 +1131,7 @@ export async function sendStartWithoutReadingResponse( const handshake = decodeHostFrame(await transport.read(2_000)); assert.ok('kind' in handshake); assert.equal(handshake.kind, 'accepted'); - await transport.write({ + await writeClientFrame(transport, { requestId: randomUUID(), operation: 'turn.start', input: { @@ -1141,6 +1143,10 @@ export async function sendStartWithoutReadingResponse( return transport; } +function writeClientFrame(transport: FramedTransport, frame: ClientFrame): Promise { + return transport.write(encodeProtocolMessage(frame)); +} + function openSocket(path: string): Promise { return new Promise((resolve, reject) => { const socket = connect(path); diff --git a/packages/runtime-host/src/__tests__/framed-transport.test.ts b/packages/runtime-host/src/__tests__/framed-transport.test.ts index 7108330a83..a8446ea930 100644 --- a/packages/runtime-host/src/__tests__/framed-transport.test.ts +++ b/packages/runtime-host/src/__tests__/framed-transport.test.ts @@ -1,8 +1,15 @@ import assert from 'node:assert/strict'; import { createServer, Socket, type Server } from 'node:net'; import { test } from 'node:test'; -import { RUNTIME_HOST_MAX_FRAME_BYTES, RuntimeHostProtocolError } from '../protocol/index.js'; +import { + encodeProtocolMessage, + RUNTIME_HOST_COMPATIBILITY_EPOCH, + RUNTIME_HOST_MAX_MESSAGE_BYTES, + RUNTIME_HOST_PROTOCOL_VERSION, + RuntimeHostProtocolError, +} from '../protocol/index.js'; import { FramedTransport, RuntimeHostTransportError } from '../transport/framed-transport.js'; +import { frameLocalIpcProtocolMessage } from '../transport/local-ipc-framing.js'; test('drains clean half-open input before reporting typed read EOF', async () => { await withSocketPair(async (transport, peer) => { @@ -13,14 +20,14 @@ test('drains clean half-open input before reporting typed read EOF', async () => (error: unknown) => error instanceof RuntimeHostTransportError && error.code === 'read_eof', ); - const reply = Buffer.from(`${JSON.stringify({ reply: true })}\n`); - const received = readSocket(peer, reply.byteLength); - await transport.writeEncoded(reply); - assert.deepEqual(await received, reply); + const reply = encodeProtocolMessage({ kind: 'draining', hostEpoch: 'host-1' }); + const received = readSocket(peer, reply.byteLength + 1); + await transport.write(reply); + assert.deepEqual(await received, Buffer.concat([reply, Buffer.from('\n')])); await ended; const failure = new Error('forced transport failure'); - transport.destroy(failure); + transport.abort(failure); await transport.closed; await assert.rejects(transport.read(0), (error: unknown) => error === failure); }); @@ -70,7 +77,7 @@ test('applies byte backpressure without dropping large valid frames', async () = test('fails closed on an oversized unterminated frame over a real socket', async () => { await withSocketPair(async (transport, peer) => { const read = transport.read(1_000); - peer.write(Buffer.alloc(RUNTIME_HOST_MAX_FRAME_BYTES + 1, 0x61)); + peer.write(Buffer.alloc(RUNTIME_HOST_MAX_MESSAGE_BYTES + 1, 0x61)); await assert.rejects( read, (error: unknown) => @@ -80,16 +87,29 @@ test('fails closed on an oversized unterminated frame over a real socket', async }); }); -test('returns a rejected Promise for an oversized outbound frame', async () => { - await withSocketPair(async (transport) => { - await assert.rejects( - transport.write({ - kind: 'draining', - hostEpoch: 'x'.repeat(RUNTIME_HOST_MAX_FRAME_BYTES), - }), - (error: unknown) => - error instanceof RuntimeHostProtocolError && error.code === 'frame_too_large', - ); +test('decodes split UTF-8 and coalesced Local IPC frames', async () => { + await withSocketPair(async (transport, peer) => { + const hello = { + kind: 'hello' as const, + clientInstanceId: '客户端', + surface: 'tui' as const, + protocolMin: RUNTIME_HOST_PROTOCOL_VERSION, + protocolMax: RUNTIME_HOST_PROTOCOL_VERSION, + compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH, + }; + const status = { requestId: 'status-1', operation: 'host.status', input: {} } as const; + const wire = Buffer.concat([ + frameLocalIpcProtocolMessage(encodeProtocolMessage(hello)), + frameLocalIpcProtocolMessage(encodeProtocolMessage(status)), + ]); + const split = wire.indexOf(Buffer.from('端')) + 1; + + peer.write(wire.subarray(0, split)); + await new Promise((resolve) => setImmediate(resolve)); + peer.write(wire.subarray(split)); + + assert.deepEqual(await transport.read(1_000), hello); + assert.deepEqual(await transport.read(1_000), status); }); }); @@ -109,7 +129,7 @@ async function withSocketPair( try { await run(transport, peer); } finally { - transport.destroy(); + transport.abort(); peer.destroy(); await transport.closed; await closeServer(server); diff --git a/packages/runtime-host/src/__tests__/host-kernel.test.ts b/packages/runtime-host/src/__tests__/host-kernel.test.ts index 5dce17450e..c0d6ebefc4 100644 --- a/packages/runtime-host/src/__tests__/host-kernel.test.ts +++ b/packages/runtime-host/src/__tests__/host-kernel.test.ts @@ -35,10 +35,12 @@ import { readHostRegistration } from '../control/registration.js'; import { removePosixEndpointDirectories } from './fixtures/endpoint-hygiene.js'; import { decodeHostFrame, + encodeProtocolMessage, RUNTIME_HOST_COMPATIBILITY_EPOCH, - RUNTIME_HOST_MAX_FRAME_BYTES, + RUNTIME_HOST_MAX_MESSAGE_BYTES, RUNTIME_HOST_PROTOCOL_VERSION, RuntimeHostProtocolError, + type ClientFrame, type ClientSurface, } from '../protocol/index.js'; import { @@ -73,6 +75,39 @@ const require = createRequire(import.meta.url); const execFileAsync = promisify(execFile); describe('non-serving Runtime Host kernel', () => { + test('service lifecycle remains ready until explicitly closed', async () => { + await withHostPaths(async (paths) => { + const capability = await resolveStorageRoot({ path: paths.root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + const host = await RuntimeHostKernel.start({ + owner, + lifecycleMode: 'service', + }); + + await sleep(25); + assert.equal(host.state, 'ready'); + assert.equal(await tryAcquireInteractiveRootOwner(capability), undefined); + + const incompatible = await connectOrSpawnRuntimeHost({ + ...paths, + rootPath: paths.root, + surface: 'tui', + protocol: LEGACY_PROTOCOL, + electionDeadlineMs: 500, + }); + assert.equal(incompatible.kind, 'incompatible'); + if (incompatible.kind === 'incompatible') { + assert.equal(incompatible.handshake.replacement, 'blocked_by_residency'); + } + + await host.close(); + const successor = await tryAcquireInteractiveRootOwner(capability); + assert.ok(successor); + await successor.close(); + }); + }); + test('elects one owner, serves status and diagnostics, and releases ownership after true-idle shutdown', async () => { await withHostPaths(async (paths) => { const winner = await startTestRuntimeHostCandidate(paths, { @@ -179,7 +214,7 @@ describe('non-serving Runtime Host kernel', () => { assert.ok(registration); assert.equal(registration.state, 'recovering'); transport = new FramedTransport(await openSocket(registration.endpoint)); - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: 'lifecycle-test', surface: 'inspect', @@ -190,14 +225,18 @@ describe('non-serving Runtime Host kernel', () => { const handshake = decodeHostFrame(await transport.read(1_000)); assert.ok('kind' in handshake && handshake.kind === 'accepted'); - await transport.write({ requestId: 'status', operation: 'host.status', input: {} }); + await writeClientFrame(transport, { + requestId: 'status', + operation: 'host.status', + input: {}, + }); const status = decodeHostFrame(await transport.read(1_000)); assert.ok(!('kind' in status) && status.operation === 'host.status' && status.ok); if (!('kind' in status) && status.operation === 'host.status' && status.ok) { assert.equal(status.result.state, 'recovering'); } - await transport.write({ + await writeClientFrame(transport, { requestId: 'query', operation: 'turn.query', input: { sessionId: 'session', turnId: 'turn' }, @@ -209,7 +248,7 @@ describe('non-serving Runtime Host kernel', () => { } } finally { releaseFactory(); - transport?.destroy(); + transport?.abort(); host = await hostTask.catch(() => undefined); await host?.close().catch(() => undefined); } @@ -408,7 +447,8 @@ describe('non-serving Runtime Host kernel', () => { if (resident.kind !== 'connected') return; const staleWhileResident = new FramedTransport(await openSocket(candidate.host.endpoint)); - await staleWhileResident.writeEncoded( + await writeRawLocalIpc( + staleWhileResident, encodeLegacyProtocolFrame({ kind: 'hello', clientInstanceId: 'stale-schema-resident', @@ -426,7 +466,7 @@ describe('non-serving Runtime Host kernel', () => { state: 'ready', replacement: 'blocked_by_residency', }); - staleWhileResident.destroy(); + staleWhileResident.abort(); await staleWhileResident.closed; const blocked = await connectOrSpawnRuntimeHost({ @@ -442,7 +482,8 @@ describe('non-serving Runtime Host kernel', () => { await resident.connection.close(); const staleAtIdle = new FramedTransport(await openSocket(candidate.host.endpoint)); - await staleAtIdle.writeEncoded( + await writeRawLocalIpc( + staleAtIdle, encodeLegacyProtocolFrame({ kind: 'hello', clientInstanceId: 'stale-schema-idle', @@ -456,7 +497,7 @@ describe('non-serving Runtime Host kernel', () => { if ('kind' in staleIdleResponse && staleIdleResponse.kind === 'incompatible') { assert.equal(staleIdleResponse.replacement, 'wait_for_idle_exit'); } - staleAtIdle.destroy(); + staleAtIdle.abort(); await staleAtIdle.closed; const replaceable = await Promise.all([ @@ -1100,7 +1141,7 @@ describe('non-serving Runtime Host kernel', () => { const transport = new FramedTransport(socket); await new Promise((resolve) => setImmediate(resolve)); const closing = candidate.host.close(); - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: 'draining-client', surface: 'tui', @@ -1110,7 +1151,7 @@ describe('non-serving Runtime Host kernel', () => { }); const response = decodeHostFrame(await transport.read(2_000)); assert.deepEqual(response, { kind: 'draining', hostEpoch: candidate.host.hostEpoch }); - transport.destroy(); + transport.abort(); await transport.closed; await closing; }); @@ -1162,7 +1203,7 @@ describe('non-serving Runtime Host kernel', () => { }); const transport = new FramedTransport(await openSocket(host.endpoint)); try { - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: 'composition-connection-release', surface: 'tui', @@ -1174,7 +1215,7 @@ describe('non-serving Runtime Host kernel', () => { assert.ok('kind' in handshake && handshake.kind === 'accepted'); if (!('kind' in handshake) || handshake.kind !== 'accepted') return; - await transport.write({ + await writeClientFrame(transport, { requestId: 'blocked-memory-mutation', operation: 'memory.mutate', input: { @@ -1187,7 +1228,7 @@ describe('non-serving Runtime Host kernel', () => { const admittedConnectionId = await handlerEntered; assert.equal(admittedConnectionId, handshake.connectionId); - transport.destroy(); + transport.abort(); await transport.closed; await new Promise((resolve) => setImmediate(resolve)); assert.deepEqual(releasedConnectionIds, []); @@ -1196,7 +1237,7 @@ describe('non-serving Runtime Host kernel', () => { assert.equal(await connectionReleased, handshake.connectionId); } finally { releaseHandler(); - transport.destroy(); + transport.abort(); await host.close().catch(() => undefined); } }); @@ -1290,7 +1331,7 @@ describe('non-serving Runtime Host kernel', () => { const transport = new FramedTransport(await openHalfOpenSocket(candidate.host.endpoint)); const incompleteSocket = await openHalfOpenSocket(candidate.host.endpoint); try { - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: 'half-open-client', surface: 'tui', @@ -1314,7 +1355,7 @@ describe('non-serving Runtime Host kernel', () => { assert.ok(owner); await owner?.close(); } finally { - transport.destroy(); + transport.abort(); incompleteSocket.destroy(); } }); @@ -1420,7 +1461,7 @@ describe('non-serving Runtime Host kernel', () => { try { const ready = await waitForUncooperativeHostMessage(child, 'ready'); transport = new FramedTransport(await openSocket(ready.endpoint)); - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: 'bounded-shutdown-test', surface: 'tui', @@ -1433,7 +1474,7 @@ describe('non-serving Runtime Host kernel', () => { assert.equal(handshake.kind, 'accepted'); const blocked = waitForUncooperativeHostMessage(child, 'operation-blocked'); - await transport.write({ + await writeClientFrame(transport, { requestId: 'blocked-turn-start', operation: 'turn.start', input: { @@ -1447,7 +1488,7 @@ describe('non-serving Runtime Host kernel', () => { child.send({ type: 'shutdown' }); await shutdownRequested; - await transport.write({ + await writeClientFrame(transport, { requestId: 'post-drain-status', operation: 'host.status', input: {}, @@ -1463,7 +1504,7 @@ describe('non-serving Runtime Host kernel', () => { const rejectedHandshakeTransport = new FramedTransport(await openSocket(ready.endpoint)); try { - await rejectedHandshakeTransport.write({ + await writeClientFrame(rejectedHandshakeTransport, { kind: 'hello', clientInstanceId: 'post-drain-client', surface: 'inspect', @@ -1476,7 +1517,7 @@ describe('non-serving Runtime Host kernel', () => { hostEpoch: ready.hostEpoch, }); } finally { - rejectedHandshakeTransport.destroy(); + rejectedHandshakeTransport.abort(); } assert.equal(child.exitCode, null); @@ -1509,7 +1550,7 @@ describe('non-serving Runtime Host kernel', () => { await connected.connection.close(); await successor.host.close(); } finally { - transport?.destroy(); + transport?.abort(); if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); await withTimeout(waitForExit(child), 1_000, 'uncooperative Host cleanup did not exit'); } @@ -1683,7 +1724,7 @@ describe('non-serving Runtime Host kernel', () => { await sendInvalidBootstrap(candidate.host.endpoint, Buffer.from('not-json\n')); await sendInvalidBootstrap( candidate.host.endpoint, - Buffer.alloc(RUNTIME_HOST_MAX_FRAME_BYTES + 1, 0x61), + Buffer.alloc(RUNTIME_HOST_MAX_MESSAGE_BYTES + 1, 0x61), ); const connected = await retryConnect(paths, CURRENT_PROTOCOL); @@ -2293,6 +2334,16 @@ function encodeLegacyProtocolFrame(frame: unknown): Buffer { return Buffer.from(`${JSON.stringify(frame)}\n`, 'utf8'); } +function writeRawLocalIpc(transport: FramedTransport, frame: Uint8Array): Promise { + return new Promise((resolve, reject) => { + transport.socket.write(frame, (error) => (error ? reject(error) : resolve())); + }); +} + +function writeClientFrame(transport: FramedTransport, frame: ClientFrame): Promise { + return transport.write(encodeProtocolMessage(frame)); +} + async function removeControlDirectoriesForRootsUnder(base: string): Promise { const rootIds = new Set(); await collectRootIds(base, rootIds); diff --git a/packages/runtime-host/src/__tests__/memory-protocol.test.ts b/packages/runtime-host/src/__tests__/memory-protocol.test.ts index ac2b5d157c..f452ac1e10 100644 --- a/packages/runtime-host/src/__tests__/memory-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/memory-protocol.test.ts @@ -3,12 +3,12 @@ import { describe, test } from 'node:test'; import { decodeClientFrame, decodeHostFrame, - encodeProtocolFrame, + encodeProtocolMessage, HOST_OPERATION_SPECS, MEMORY_DOCUMENT_CHUNK_MAX_BYTES, MEMORY_ENTRY_PAGE_MAX_ITEMS, MEMORY_RESULT_MAX_BYTES, - RUNTIME_HOST_MAX_FRAME_BYTES, + RUNTIME_HOST_MAX_MESSAGE_BYTES, RuntimeHostProtocolError, } from '../protocol/index.js'; @@ -88,12 +88,12 @@ describe('Memory protocol', () => { assert.doesNotThrow(() => response('memory.query', result)); assert.ok(Buffer.byteLength(JSON.stringify(result), 'utf8') <= MEMORY_RESULT_MAX_BYTES); assert.ok( - encodeProtocolFrame({ + encodeProtocolMessage({ requestId: 'memory-page', operation: 'memory.query', ok: true, result, - }).byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES, + }).byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES, ); const input = { @@ -104,11 +104,11 @@ describe('Memory protocol', () => { }; assert.doesNotThrow(() => request('memory.mutate', input)); assert.ok( - encodeProtocolFrame({ + encodeProtocolMessage({ requestId: 'memory-chunk', operation: 'memory.mutate', input, - }).byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES, + }).byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES, ); assert.throws( () => diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 8974d78ba6..0e4de5ea80 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -8,13 +8,12 @@ import { decodeHostRegistration, decodeSessionMessageQueueProjection, decodeSessionContinuitySnapshot, - encodeProtocolFrame, + encodeProtocolMessage, HOST_OPERATION_SPECS, MESSAGE_OPERATION_RESULT_MAX_BYTES, MESSAGE_QUEUE_MAX_ENTRIES, negotiateProtocol, - ProtocolFrameDecoder, - RUNTIME_HOST_MAX_FRAME_BYTES, + RUNTIME_HOST_MAX_MESSAGE_BYTES, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, SESSION_CONTINUITY_SCHEMA_VERSION, @@ -407,7 +406,7 @@ describe('Runtime Host bootstrap protocol', () => { } }); - test('enforces UTF-8 snapshot, live field, and whole-frame byte bounds', () => { + test('enforces UTF-8 snapshot, live field, and whole-message byte bounds', () => { const snapshot = continuitySnapshot('epoch-1'); assert.ok(Buffer.byteLength(JSON.stringify(snapshot)) < SESSION_CONTINUITY_SNAPSHOT_MAX_BYTES); assert.throws( @@ -470,7 +469,7 @@ describe('Runtime Host bootstrap protocol', () => { () => decodeHostFrame({ ...frame, - privatePadding: 'x'.repeat(RUNTIME_HOST_MAX_FRAME_BYTES), + privatePadding: 'x'.repeat(RUNTIME_HOST_MAX_MESSAGE_BYTES), }), isInvalidFrame, ); @@ -544,7 +543,7 @@ describe('Runtime Host bootstrap protocol', () => { }), ); assert.doesNotThrow(() => - encodeProtocolFrame({ + encodeProtocolMessage({ requestId: 'credential-export', operation: 'configuration.credentials.export', ok: true, @@ -646,14 +645,12 @@ describe('Runtime Host bootstrap protocol', () => { }, }; - const encoded = encodeProtocolFrame(frame); + const encoded = encodeProtocolMessage(frame); assert.ok( - encoded.byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES, - `${label} envelope exceeds the protocol frame limit`, + encoded.byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES, + `${label} envelope exceeds the protocol message limit`, ); - const decodedFrames = new ProtocolFrameDecoder().push(encoded); - assert.equal(decodedFrames.length, 1); - const decoded = decodeHostFrame(decodedFrames[0]); + const decoded = decodeHostFrame(JSON.parse(encoded.toString('utf8'))); assert.ok('kind' in decoded); if (!('kind' in decoded)) continue; assert.equal(decoded.kind, 'subscription.session_event'); @@ -704,36 +701,9 @@ describe('Runtime Host bootstrap protocol', () => { const canonical = decodeHostFrame(frame); assert.ok(Buffer.byteLength(`${JSON.stringify(canonical)}\n`, 'utf8') > 64 * 1024); - const encoded = encodeProtocolFrame(canonical); - assert.ok(encoded.byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES); - const [decoded] = new ProtocolFrameDecoder().push(encoded); - assert.deepEqual(decodeHostFrame(decoded), canonical); - }); - - test('decodes split UTF-8 and multiple newline-delimited frames without an unbounded tail', () => { - const decoder = new ProtocolFrameDecoder(); - const wire = Buffer.from( - `${JSON.stringify({ kind: 'hello', clientInstanceId: '客户端', surface: 'tui', protocolMin: RUNTIME_HOST_PROTOCOL_VERSION, protocolMax: RUNTIME_HOST_PROTOCOL_VERSION, compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH })}\n` + - `${JSON.stringify({ requestId: 'status-1', operation: 'host.status', input: {} })}\n`, - ); - const split = wire.indexOf(Buffer.from('端')) + 1; - assert.deepEqual(decoder.push(wire.subarray(0, split)), []); - const frames = decoder.push(wire.subarray(split)); - assert.equal(frames.length, 2); - assert.deepEqual(decodeClientFrame(frames[0]), { - kind: 'hello', - clientInstanceId: '客户端', - surface: 'tui', - protocolMin: RUNTIME_HOST_PROTOCOL_VERSION, - protocolMax: RUNTIME_HOST_PROTOCOL_VERSION, - compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH, - }); - assert.deepEqual(decodeClientFrame(frames[1]), { - requestId: 'status-1', - operation: 'host.status', - input: {}, - }); - decoder.end(); + const encoded = encodeProtocolMessage(canonical); + assert.ok(encoded.byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES); + assert.deepEqual(decodeHostFrame(JSON.parse(encoded.toString('utf8'))), canonical); }); test('accepts protocol v0 in handshakes and Host registration while rejecting negatives', () => { @@ -1038,7 +1008,7 @@ describe('Runtime Host bootstrap protocol', () => { maxSteps: 4, }, }; - const start = decodeClientFrame(JSON.parse(encodeProtocolFrame(startWire).toString('utf8'))); + const start = decodeClientFrame(JSON.parse(encodeProtocolMessage(startWire).toString('utf8'))); assert.deepEqual(start, { requestId: 'start-request-1', operation: 'turn.start', @@ -1064,7 +1034,7 @@ describe('Runtime Host bootstrap protocol', () => { }, }; assert.deepEqual( - decodeClientFrame(JSON.parse(encodeProtocolFrame(submitWire).toString('utf8'))), + decodeClientFrame(JSON.parse(encodeProtocolMessage(submitWire).toString('utf8'))), submitWire, ); assert.throws( @@ -1210,7 +1180,7 @@ describe('Runtime Host bootstrap protocol', () => { }, }; assert.deepEqual(decodeHostFrame(response), response); - assert.ok(encodeProtocolFrame(response).byteLength < RUNTIME_HOST_MAX_FRAME_BYTES); + assert.ok(encodeProtocolMessage(response).byteLength < RUNTIME_HOST_MAX_MESSAGE_BYTES); const request = 'r'.repeat(TURN_SKILL_ID_MAX_LENGTH); const id = 'i'.repeat(81); @@ -1360,7 +1330,7 @@ describe('Runtime Host bootstrap protocol', () => { operation: 'turn.message.submit', input, }); - assert.ok(encodeProtocolFrame(frame).byteLength < RUNTIME_HOST_MAX_FRAME_BYTES); + assert.ok(encodeProtocolMessage(frame).byteLength < RUNTIME_HOST_MAX_MESSAGE_BYTES); assert.throws( () => decodeClientFrame({ @@ -1558,10 +1528,19 @@ describe('Runtime Host bootstrap protocol', () => { ); }); - test('rejects a frame before buffering more than the byte cap', () => { - const decoder = new ProtocolFrameDecoder(); + test('bounds encoded protocol messages', () => { + const empty = { kind: 'draining', hostEpoch: '' } as const; + const overhead = Buffer.byteLength(JSON.stringify(empty), 'utf8'); + const value = { + ...empty, + hostEpoch: 'x'.repeat(RUNTIME_HOST_MAX_MESSAGE_BYTES - overhead), + }; + const message = encodeProtocolMessage(value); + + assert.equal(message.byteLength, RUNTIME_HOST_MAX_MESSAGE_BYTES); + assert.notEqual(message.at(-1), 0x0a); assert.throws( - () => decoder.push(Buffer.alloc(RUNTIME_HOST_MAX_FRAME_BYTES + 1, 0x61)), + () => encodeProtocolMessage({ ...value, hostEpoch: `${value.hostEpoch}x` }), (error: unknown) => error instanceof RuntimeHostProtocolError && error.code === 'frame_too_large', ); diff --git a/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts index 73b35508e5..c33295a8b3 100644 --- a/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts @@ -24,8 +24,10 @@ import { } from '../client/index.js'; import { decodeHostFrame, + encodeProtocolMessage, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, + type ClientFrame, type SessionCatalogItem, type SessionCatalogProjection, type SessionCreateInput, @@ -693,7 +695,7 @@ test('stable Session creation survives response loss and Host restart', { dropped = await sendCreateWithoutReadingResponse(host.endpoint, input); const observer = await connectClient(root, 'tui'); const committed = await waitForSession(observer, input.sessionId); - dropped.destroy(); + dropped.abort(); dropped = undefined; await observer.close(); @@ -711,7 +713,7 @@ test('stable Session creation survives response loss and Host restart', { await stopHost(host); host = undefined; } finally { - dropped?.destroy(); + dropped?.abort(); await terminateHost(host); await rm(join(resolveRootControlNamespace(), capability.rootId), { recursive: true, @@ -1060,7 +1062,7 @@ async function sendCreateWithoutReadingResponse( input: SessionCreateInput, ): Promise { const transport = new FramedTransport(await openSocket(endpoint)); - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: randomUUID(), surface: 'desktop', @@ -1071,7 +1073,7 @@ async function sendCreateWithoutReadingResponse( const handshake = decodeHostFrame(await transport.read(2_000)); assert.ok('kind' in handshake); assert.equal(handshake.kind, 'accepted'); - await transport.write({ + await writeClientFrame(transport, { requestId: randomUUID(), operation: 'session.create', input, @@ -1079,6 +1081,10 @@ async function sendCreateWithoutReadingResponse( return transport; } +function writeClientFrame(transport: FramedTransport, frame: ClientFrame): Promise { + return transport.write(encodeProtocolMessage(frame)); +} + function openSocket(path: string): Promise { return new Promise((resolve, reject) => { const socket = connect(path); diff --git a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts index 9d37b123bd..700adfd9c9 100644 --- a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts @@ -6,9 +6,8 @@ import { TOOL_OUTPUT_DELTA_MAX_CHARS } from '@maka/core/events'; import { PiAgentBackend, type PiAgentTransport } from '@maka/runtime'; import { decodeHostFrame, - encodeProtocolFrame, - ProtocolFrameDecoder, - RUNTIME_HOST_MAX_FRAME_BYTES, + encodeProtocolMessage, + RUNTIME_HOST_MAX_MESSAGE_BYTES, type SessionTranscriptCursor, type SubscriptionFrame, } from '../protocol/index.js'; @@ -239,11 +238,9 @@ test('tool output preserves one domain event and one wire frame', async () => { await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', output); await waitFor(() => sink.frames.length === 1); - const encoded = encodeProtocolFrame(sink.frames[0]!); - assert.ok(encoded.byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES); - const decodedFrames = new ProtocolFrameDecoder().push(encoded); - assert.equal(decodedFrames.length, 1); - const decoded = decodeHostFrame(decodedFrames[0]); + const encoded = encodeProtocolMessage(sink.frames[0]!); + assert.ok(encoded.byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES); + const decoded = decodeHostFrame(JSON.parse(encoded.toString('utf8'))); assert.ok('kind' in decoded); if (!('kind' in decoded)) return; assert.equal(decoded.kind, 'subscription.session_event'); diff --git a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts index d252670c39..78a188041c 100644 --- a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts +++ b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts @@ -19,15 +19,17 @@ import { prepareRuntimeHostEndpoint } from '../control/endpoint.js'; import { removeHostRegistration, writeHostRegistration } from '../control/registration.js'; import { decodeClientFrame, - encodeProtocolFrame, + encodeProtocolMessage, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION, SESSION_CONTINUITY_SCHEMA_VERSION, + type HostFrame, type RequestFrame, type SubscriptionFrame, } from '../protocol/index.js'; import { FramedTransport } from '../transport/framed-transport.js'; +import { frameLocalIpcProtocolMessage } from '../transport/local-ipc-framing.js'; const PROTOCOL = { min: RUNTIME_HOST_PROTOCOL_VERSION, @@ -39,15 +41,16 @@ test('registers a subscription before receiving a coalesced first frame', async async (transport, hostEpoch) => { const request = await acceptConnectionAndReadOpen(transport, hostEpoch); const opened = openResult(hostEpoch, 'subscription-ordered'); - await transport.writeEncoded( + await writeRawLocalIpc( + transport, Buffer.concat([ - encodeProtocolFrame({ + encodeLocalIpcTestFrame({ requestId: request.requestId, operation: 'subscription.open', ok: true, result: opened, }), - encodeProtocolFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)), + encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)), ]), ); await answerClose(transport, opened.subscriptionId); @@ -80,15 +83,16 @@ test('delivers Runtime Resource PTY frames without closing the connection', asyn ptySequence: 7, data: 'ready', }; - await transport.writeEncoded( + await writeRawLocalIpc( + transport, Buffer.concat([ - encodeProtocolFrame({ + encodeLocalIpcTestFrame({ requestId: request.requestId, operation: 'subscription.open', ok: true, result: opened, }), - encodeProtocolFrame(frame), + encodeLocalIpcTestFrame(frame), ]), ); await answerClose(transport, opened.subscriptionId); @@ -118,15 +122,16 @@ test('isolates a sequence gap and continues requests on the same connection', as async (transport, hostEpoch) => { const request = await acceptConnectionAndReadOpen(transport, hostEpoch); const opened = openResult(hostEpoch, 'subscription-gap'); - await transport.writeEncoded( + await writeRawLocalIpc( + transport, Buffer.concat([ - encodeProtocolFrame({ + encodeLocalIpcTestFrame({ requestId: request.requestId, operation: 'subscription.open', ok: true, result: opened, }), - encodeProtocolFrame(deltaFrame(hostEpoch, opened.subscriptionId, 2)), + encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 2)), ]), ); await answerClose(transport, opened.subscriptionId); @@ -151,13 +156,14 @@ test('rejects epoch and Session correlation changes per subscription', async () async (transport, hostEpoch) => { const request = await acceptConnectionAndReadOpen(transport, hostEpoch); const opened = openResult(hostEpoch, `subscription-${changed}`); - await transport.write({ + await writeProtocolFrame(transport, { requestId: request.requestId, operation: 'subscription.open', ok: true, result: opened, }); - await transport.write( + await writeProtocolFrame( + transport, changed === 'graph' ? { kind: 'subscription.agent_graph_changed', @@ -201,7 +207,7 @@ test('evicts a locally slow iterator and keeps the connection usable', async () const request = await acceptConnectionAndReadOpen(transport, hostEpoch); const opened = openResult(hostEpoch, 'subscription-slow'); const frames = [ - encodeProtocolFrame({ + encodeLocalIpcTestFrame({ requestId: request.requestId, operation: 'subscription.open', ok: true, @@ -209,9 +215,11 @@ test('evicts a locally slow iterator and keeps the connection usable', async () }), ]; for (let sequence = 1; sequence <= 33; sequence += 1) { - frames.push(encodeProtocolFrame(deltaFrame(hostEpoch, opened.subscriptionId, sequence))); + frames.push( + encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, sequence)), + ); } - await transport.writeEncoded(Buffer.concat(frames)); + await writeRawLocalIpc(transport, Buffer.concat(frames)); await answerClose(transport, opened.subscriptionId, closeObserved.resolve); await answerStatus(transport, hostEpoch); }, @@ -233,13 +241,13 @@ test('ends every active subscription with connection_closed on EOF', async () => await withProtocolPeer( async (transport, hostEpoch) => { const request = await acceptConnectionAndReadOpen(transport, hostEpoch); - await transport.write({ + await writeProtocolFrame(transport, { requestId: request.requestId, operation: 'subscription.open', ok: true, result: openResult(hostEpoch, 'subscription-eof'), }); - transport.destroyAfterFlush(); + transport.closeAfterFlush(); }, async (connection) => { const subscription = await connection.openSessionSubscription({ @@ -266,7 +274,7 @@ test('loads a canonical transcript while live frames continue on the same connec async (transport, hostEpoch) => { const openRequest = await acceptConnectionAndReadOpen(transport, hostEpoch); const opened = openResult(hostEpoch, 'subscription-transcript'); - await transport.write({ + await writeProtocolFrame(transport, { requestId: openRequest.requestId, operation: 'subscription.open', ok: true, @@ -279,10 +287,11 @@ test('loads a canonical transcript while live frames continue on the same connec kind: 'start', subscriptionId: opened.subscriptionId, }); - await transport.writeEncoded( + await writeRawLocalIpc( + transport, Buffer.concat([ - encodeProtocolFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)), - encodeProtocolFrame({ + encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)), + encodeLocalIpcTestFrame({ requestId: transcriptRequest.requestId, operation: 'session.transcript.query', ok: true, @@ -327,7 +336,7 @@ test('restarts transcript loading after an expired snapshot', async () => { async (transport, hostEpoch) => { const openRequest = await acceptConnectionAndReadOpen(transport, hostEpoch); const opened = openResult(hostEpoch, 'subscription-retry'); - await transport.write({ + await writeProtocolFrame(transport, { requestId: openRequest.requestId, operation: 'subscription.open', ok: true, @@ -339,7 +348,7 @@ test('restarts transcript loading after an expired snapshot', async () => { kind: 'start', subscriptionId: opened.subscriptionId, }); - await transport.write({ + await writeProtocolFrame(transport, { requestId: startRequest.requestId, operation: 'session.transcript.query', ok: true, @@ -363,7 +372,7 @@ test('restarts transcript loading after an expired snapshot', async () => { messageIndex: 0, byteOffset: splitAt, }); - await transport.write({ + await writeProtocolFrame(transport, { requestId: continuationRequest.requestId, operation: 'session.transcript.query', ok: true, @@ -375,7 +384,7 @@ test('restarts transcript loading after an expired snapshot', async () => { kind: 'start', subscriptionId: opened.subscriptionId, }); - await transport.write({ + await writeProtocolFrame(transport, { requestId: retryStartRequest.requestId, operation: 'session.transcript.query', ok: true, @@ -424,7 +433,8 @@ test('forces a same-v0 pre-epoch Host through its incompatible replacement path' { min: hello.protocolMin, max: hello.protocolMax }, { min: RUNTIME_HOST_PROTOCOL_VERSION + 1, max: RUNTIME_HOST_PROTOCOL_VERSION + 1 }, ); - await transport.writeEncoded( + await writeRawLocalIpc( + transport, encodeLegacyProtocolFrame({ kind: 'incompatible', hostEpoch, @@ -434,7 +444,7 @@ test('forces a same-v0 pre-epoch Host through its incompatible replacement path' replacement: 'wait_for_idle_exit', }), ); - transport.destroyAfterFlush(); + transport.closeAfterFlush(); await transport.closed; })().then(serverTask.resolve, serverTask.reject); }); @@ -537,7 +547,7 @@ async function acceptConnectionAndReadOpen( ): Promise> { const hello = decodeClientFrame(await transport.read(1_000)); assert.ok('kind' in hello && hello.kind === 'hello'); - await transport.write({ + await writeProtocolFrame(transport, { kind: 'accepted', hostEpoch, connectionId: 'connection-1', @@ -561,7 +571,7 @@ async function answerClose( assert.equal(request.operation, 'subscription.close'); assert.deepEqual(request.input, { subscriptionId }); onObserved?.(); - await transport.write({ + await writeProtocolFrame(transport, { requestId: request.requestId, operation: 'subscription.close', ok: true, @@ -573,7 +583,7 @@ async function answerStatus(transport: FramedTransport, hostEpoch: string): Prom const request = decodeClientFrame(await transport.read(1_000)); assert.ok(!('kind' in request)); assert.equal(request.operation, 'host.status'); - await transport.write({ + await writeProtocolFrame(transport, { requestId: request.requestId, operation: 'host.status', ok: true, @@ -643,6 +653,20 @@ function hasSubscriptionReason(reason: RuntimeHostSubscriptionError['reason']) { error instanceof RuntimeHostSubscriptionError && error.reason === reason; } +function writeProtocolFrame(transport: FramedTransport, frame: HostFrame): Promise { + return transport.write(encodeProtocolMessage(frame)); +} + +function encodeLocalIpcTestFrame(frame: HostFrame): Buffer { + return frameLocalIpcProtocolMessage(encodeProtocolMessage(frame)); +} + +function writeRawLocalIpc(transport: FramedTransport, frame: Uint8Array): Promise { + return new Promise((resolve, reject) => { + transport.socket.write(frame, (error) => (error ? reject(error) : resolve())); + }); +} + function encodeLegacyProtocolFrame(frame: unknown): Buffer { return Buffer.from(`${JSON.stringify(frame)}\n`, 'utf8'); } diff --git a/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts b/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts index 18fb067e9d..56ad3ffac3 100644 --- a/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts +++ b/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts @@ -14,10 +14,12 @@ import { prepareRuntimeHostEndpoint } from '../control/endpoint.js'; import { removeHostRegistration, writeHostRegistration } from '../control/registration.js'; import { decodeClientFrame, + encodeProtocolMessage, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION, RuntimeHostProtocolError, + type HostFrame, type OperationInput, type OperationOutput, type RequestFrame, @@ -154,7 +156,7 @@ describe('Usage/Pricing client response correlation', () => { async (transport, hostEpoch) => { const request = await acceptConnectionAndReadRequest(transport, hostEpoch); assert.equal(request.operation, mismatch.operation); - await transport.write({ + await writeHostFrame(transport, { requestId: request.requestId, operation: mismatch.operation, ok: true, @@ -186,7 +188,7 @@ describe('Usage/Pricing client response correlation', () => { offset: 50, limit: 10, }); - await transport.write({ + await writeHostFrame(transport, { requestId: request.requestId, operation: 'usage.query', ok: true, @@ -313,7 +315,7 @@ async function acceptConnectionAndReadRequest( ): Promise { const hello = decodeClientFrame(await transport.read(REQUEST_TIMEOUT_MS)); assert.ok('kind' in hello && hello.kind === 'hello'); - await transport.write({ + await writeHostFrame(transport, { kind: 'accepted', hostEpoch, connectionId: 'usage-pricing-correlation', @@ -326,6 +328,10 @@ async function acceptConnectionAndReadRequest( return request as RequestFrame; } +function writeHostFrame(transport: FramedTransport, frame: HostFrame): Promise { + return transport.write(encodeProtocolMessage(frame)); +} + function listen(server: Server, path: string): Promise { return new Promise((resolve, reject) => { server.once('error', reject); diff --git a/packages/runtime-host/src/__tests__/usage-pricing-protocol.test.ts b/packages/runtime-host/src/__tests__/usage-pricing-protocol.test.ts index 90bd1cdbd5..c8cc96b7a6 100644 --- a/packages/runtime-host/src/__tests__/usage-pricing-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/usage-pricing-protocol.test.ts @@ -19,11 +19,11 @@ import { decodeHostFrame, decodeUsageQueryInput, encodePricingQueryResult, - encodeProtocolFrame, + encodeProtocolMessage, HOST_OPERATION_SPECS, PRICING_PAGE_MAX_BYTES, PRICING_PAGE_MAX_ITEMS, - RUNTIME_HOST_MAX_FRAME_BYTES, + RUNTIME_HOST_MAX_MESSAGE_BYTES, RuntimeHostProtocolError, USAGE_PAGE_MAX_BYTES, USAGE_PAGE_MAX_ITEMS, @@ -559,7 +559,7 @@ describe('Usage/Pricing protocol', () => { ); }); - test('bounds a page of maximum-length CJK pricing items below the frame limit', () => { + test('bounds a page of maximum-length CJK pricing items below the message limit', () => { const cjkEntries = Array.from({ length: PRICING_PAGE_MAX_ITEMS }, (_, index) => customPricingConfigEntry(maximumCjkPricing(index)), ).sort((left, right) => comparePricingModelKeys(left.pricing.modelKey, right.pricing.modelKey)); @@ -577,12 +577,12 @@ describe('Usage/Pricing protocol', () => { const pageBytes = Buffer.byteLength(JSON.stringify(maximumPage), 'utf8'); assert.ok(pageBytes <= PRICING_PAGE_MAX_BYTES); assert.ok( - encodeProtocolFrame({ + encodeProtocolMessage({ requestId: 'maximum-cjk-pricing-page', operation: 'pricing.query', ok: true, result: maximumPage, - }).byteLength <= RUNTIME_HOST_MAX_FRAME_BYTES, + }).byteLength <= RUNTIME_HOST_MAX_MESSAGE_BYTES, ); assert.throws( () => @@ -758,7 +758,7 @@ async function queryUsageRows( ); const frame = decodeHostFrame( JSON.parse( - encodeProtocolFrame({ + encodeProtocolMessage({ requestId: `usage-${source}-identity-query`, operation: 'usage.query', ...outcome, @@ -786,7 +786,7 @@ async function queryUsageBuckets( ); const frame = decodeHostFrame( JSON.parse( - encodeProtocolFrame({ + encodeProtocolMessage({ requestId: 'usage-bucket-identity-query', operation: 'usage.query', ...outcome, diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index d164f797a2..e7c2b8c6b7 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -11,7 +11,9 @@ import { import { readHostRegistration, RuntimeHostRegistrationError } from '../control/registration.js'; import { decodeHostFrame, + encodeProtocolMessage, isClientCapabilityHostFrameKind, + type ClientFrame, type ClientCapabilityHostFrame, type ClientCapabilityReplaceResult, type ClientCapabilityUnregisterResult, @@ -68,6 +70,7 @@ import { validateProtocolRange, } from '../protocol/index.js'; import { FramedTransport, RuntimeHostTransportError } from '../transport/framed-transport.js'; +import type { RuntimeHostMessageTransport } from '../transport/message-transport.js'; import type { OperationSpec } from '../protocol/operation-spec.js'; import { ClientSessionSubscription, @@ -258,7 +261,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { readonly connectionId: string; readonly selectedProtocol: number; readonly closed: Promise; - readonly #transport: FramedTransport; + readonly #transport: RuntimeHostMessageTransport; readonly #pendingRequests = new Map(); readonly #retiredRequests = new Map(); readonly #queuedDomainFrames: QueuedDomainFrame[] = []; @@ -275,7 +278,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { readonly #onLivenessProbe: (() => void) | undefined; constructor( - transport: FramedTransport, + transport: RuntimeHostMessageTransport, accepted: { hostEpoch: string; connectionId: string; @@ -293,7 +296,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { this.selectedProtocol = accepted.selectedProtocol; this.closed = this.#transport.closed; this.#clientCapabilities = new ClientCapabilityChannel({ - write: (frame) => this.#transport.write(frame), + write: (frame) => writeClientFrame(this.#transport, frame), replace: (input, timeoutMs) => this.#requestOperation( 'client.capability.replace', @@ -389,7 +392,9 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { this.#queuedDomainFrames.push({ requestId, frame }); this.#drainDomainRequests(); } else { - void this.#transport.write(frame).catch((error: unknown) => this.#fail(asError(error))); + void writeClientFrame(this.#transport, frame).catch((error: unknown) => + this.#fail(asError(error)), + ); } return result; } @@ -405,9 +410,9 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { if (!pending || pending.domainState !== 'queued') continue; pending.domainState = 'in_flight'; this.#inFlightDomainRequests += 1; - void this.#transport - .write(queued.frame) - .catch((error: unknown) => this.#fail(asError(error))); + void writeClientFrame(this.#transport, queued.frame).catch((error: unknown) => + this.#fail(asError(error)), + ); } } @@ -549,7 +554,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { async close(): Promise { this.#clientCapabilities.close(new Error('Runtime Host connection closed by Client')); - this.#transport.destroy(); + this.#transport.abort(); await this.#transport.closed; } @@ -845,7 +850,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { this.#clientCapabilities.close(error); this.#configurationChangeListeners.clear(); this.#sessionCatalogChangeListeners.clear(); - this.#transport.destroy(); + this.#transport.abort(); } } @@ -985,7 +990,7 @@ export async function connectResolvedRuntimeHost( const handshakeDeadline = phaseDeadline(handshakeTimeoutMs, input.electionDeadline); const handshakeBudget = remainingTimeout(handshakeDeadline.at); if (handshakeBudget === undefined) { - transport.destroy(); + transport.abort(); if (handshakeDeadline.exhaustsElection) { return { kind: 'election_deadline_elapsed', endpointConnected: true }; } @@ -996,7 +1001,7 @@ export async function connectResolvedRuntimeHost( handshakeTimeoutError = handshakeDeadline.exhaustsElection ? new ElectionDeadlineElapsedError() : new Error('Timed out handshaking with Runtime Host'); - transport.destroy(handshakeTimeoutError); + transport.abort(handshakeTimeoutError); }, handshakeBudget); try { const staleCompatibility = registration.compatibilityEpoch !== RUNTIME_HOST_COMPATIBILITY_EPOCH; @@ -1006,7 +1011,7 @@ export async function connectResolvedRuntimeHost( max: Math.min(Number.MAX_SAFE_INTEGER, registration.protocolMax + 1), } : input.protocol; - await transport.write({ + await writeClientFrame(transport, { kind: 'hello', clientInstanceId: input.clientInstanceId, surface: input.surface, @@ -1031,7 +1036,7 @@ export async function connectResolvedRuntimeHost( throw new Error('Runtime Host returned a non-handshake frame before acceptance'); } if (handshake.hostEpoch !== registration.hostEpoch) { - transport.destroy(); + transport.abort(); return { kind: 'unavailable', reason: 'epoch_mismatch', registration }; } if (handshake.kind === 'accepted') { @@ -1055,11 +1060,11 @@ export async function connectResolvedRuntimeHost( }), }; } - transport.destroy(); + transport.abort(); if (handshake.kind === 'incompatible') return { kind: 'incompatible', handshake, registration }; return { kind: 'draining', registration }; } catch (error) { - transport.destroy(); + transport.abort(); const failure = handshakeTimeoutError ?? error; if (failure instanceof ElectionDeadlineElapsedError) { return { kind: 'election_deadline_elapsed', endpointConnected: true }; @@ -1160,6 +1165,17 @@ function asError(error: unknown): Error { return error instanceof Error ? error : new Error(String(error)); } +function writeClientFrame( + transport: RuntimeHostMessageTransport, + frame: ClientFrame, +): Promise { + try { + return transport.write(encodeProtocolMessage(frame)); + } catch (error) { + return Promise.reject(error); + } +} + function requestTimeoutError(operation: OperationKey): RuntimeHostTransportError { return new RuntimeHostTransportError( 'read_timeout', diff --git a/packages/runtime-host/src/client/session-subscription.ts b/packages/runtime-host/src/client/session-subscription.ts index 0b2edd5dd2..a9a825e33a 100644 --- a/packages/runtime-host/src/client/session-subscription.ts +++ b/packages/runtime-host/src/client/session-subscription.ts @@ -1,5 +1,5 @@ import { - encodeProtocolFrame, + encodeProtocolMessage, type SessionContinuitySnapshot, type SubscriptionFrame, type SubscriptionOpenResult, @@ -298,7 +298,7 @@ export class ClientSessionSubscription waiting.resolve({ done: false, value: frame }); return; } - const encodedBytes = encodeProtocolFrame(frame).byteLength; + const encodedBytes = encodeProtocolMessage(frame).byteLength; if ( this.#queue.length >= MAX_CLIENT_QUEUED_FRAMES || this.#queuedBytes + encodedBytes > MAX_CLIENT_QUEUED_BYTES diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 0922df6642..a8ca09c29c 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -1,4 +1,3 @@ -import { TextDecoder } from 'node:util'; import { requireCount, requireId, requireRecord, requireString } from './codec.js'; import { invalidProtocolFrame, RuntimeHostProtocolError } from './errors.js'; import { requireHostLifecycleState } from './host-status.js'; @@ -62,9 +61,15 @@ export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 10 as const; // envelope and independently bounded justification are added. Keep transport // capacity large enough to represent that domain value; narrower surfaces such // as Session continuity retain their own limits. -export const RUNTIME_HOST_MAX_FRAME_BYTES = 96 * 1024; +export const RUNTIME_HOST_MAX_MESSAGE_BYTES = 96 * 1024; export const RUNTIME_HOST_MAX_IN_FLIGHT_DOMAIN_REQUESTS = 64; +declare const encodedProtocolMessageBrand: unique symbol; + +export type EncodedProtocolMessage = Buffer & { + readonly [encodedProtocolMessageBrand]: true; +}; + export type ClientSurface = 'desktop' | 'tui' | 'run' | 'activation' | 'bot' | 'inspect'; export interface ProtocolRange { @@ -251,73 +256,15 @@ export function decodeHostRegistration(value: unknown): HostRegistration { }; } -export function encodeProtocolFrame(value: ClientFrame | HostFrame): Buffer { - const encoded = Buffer.from(`${JSON.stringify(value)}\n`, 'utf8'); - if (encoded.byteLength > RUNTIME_HOST_MAX_FRAME_BYTES) { +export function encodeProtocolMessage(value: ClientFrame | HostFrame): EncodedProtocolMessage { + const encoded = Buffer.from(JSON.stringify(value), 'utf8'); + if (encoded.byteLength > RUNTIME_HOST_MAX_MESSAGE_BYTES) { throw new RuntimeHostProtocolError( 'frame_too_large', - 'Runtime Host frame exceeds the byte limit', + 'Runtime Host message exceeds the byte limit', ); } - return encoded; -} - -export class ProtocolFrameDecoder { - readonly #decoder = new TextDecoder('utf-8', { fatal: true }); - #pending = Buffer.alloc(0); - - push(chunk: Uint8Array): unknown[] { - const frames: unknown[] = []; - let offset = 0; - while (offset < chunk.byteLength) { - const newline = chunk.indexOf(0x0a, offset); - const end = newline === -1 ? chunk.byteLength : newline; - const segment = Buffer.from(chunk.subarray(offset, end)); - const delimiterBytes = newline === -1 ? 0 : 1; - if ( - this.#pending.byteLength + segment.byteLength + delimiterBytes > - RUNTIME_HOST_MAX_FRAME_BYTES - ) { - throw new RuntimeHostProtocolError( - 'frame_too_large', - 'Runtime Host frame exceeds the byte limit', - ); - } - if (segment.byteLength > 0) this.#pending = Buffer.concat([this.#pending, segment]); - if (newline === -1) break; - frames.push(this.#decodePending()); - this.#pending = Buffer.alloc(0); - offset = newline + 1; - } - return frames; - } - - end(): void { - if (this.#pending.byteLength !== 0) { - throw new RuntimeHostProtocolError( - 'invalid_frame', - 'Runtime Host stream ended with a partial frame', - ); - } - } - - #decodePending(): unknown { - if (this.#pending.byteLength === 0) { - throw invalidProtocolFrame('Runtime Host frame is empty'); - } - let text: string; - try { - const bytes = this.#pending.at(-1) === 0x0d ? this.#pending.subarray(0, -1) : this.#pending; - text = this.#decoder.decode(bytes); - } catch { - throw new RuntimeHostProtocolError('invalid_utf8', 'Runtime Host frame is not valid UTF-8'); - } - try { - return JSON.parse(text) as unknown; - } catch { - throw new RuntimeHostProtocolError('invalid_json', 'Runtime Host frame is not valid JSON'); - } - } + return encoded as EncodedProtocolMessage; } function requireProtocolVersion(value: unknown, label: string): number { diff --git a/packages/runtime-host/src/server/candidate.ts b/packages/runtime-host/src/server/candidate.ts index 05e03686b7..bd0dba91bd 100644 --- a/packages/runtime-host/src/server/candidate.ts +++ b/packages/runtime-host/src/server/candidate.ts @@ -28,6 +28,7 @@ export async function startRuntimeHostCandidate( if (!owner) return { kind: 'loser' }; const host = await RuntimeHostKernel.start({ owner, + lifecycleMode: 'ephemeral', idleGraceMs: options.idleGraceMs, handshakeTimeoutMs: options.handshakeTimeoutMs, }); diff --git a/packages/runtime-host/src/server/connection-authority.ts b/packages/runtime-host/src/server/connection-authority.ts new file mode 100644 index 0000000000..8dfd17baf0 --- /dev/null +++ b/packages/runtime-host/src/server/connection-authority.ts @@ -0,0 +1,37 @@ +import { HOST_OPERATION_SPECS, type OperationKey } from '../protocol/index.js'; + +export interface RuntimeHostConnectionAuthority { + readonly principalKind: 'local_owner' | 'access_credential'; + readonly principalId: string; + readonly operationGrants: 'all' | readonly OperationKey[]; + readonly canPublishClientCapabilities: boolean; + readonly canUseHostPaths: boolean; +} + +export const LOCAL_OWNER_CONNECTION_AUTHORITY = createRuntimeHostConnectionAuthority({ + principalKind: 'local_owner', + principalId: 'local_os_user', + operationGrants: 'all', + canPublishClientCapabilities: true, + canUseHostPaths: true, +}); + +export function createRuntimeHostConnectionAuthority( + input: RuntimeHostConnectionAuthority, +): RuntimeHostConnectionAuthority { + if (!/^[A-Za-z0-9_.:-]{1,128}$/.test(input.principalId)) { + throw new Error('Runtime Host connection principal is invalid'); + } + const operationGrants = + input.operationGrants === 'all' + ? 'all' + : Object.freeze( + [...new Set(input.operationGrants)].map((operation) => { + if (!Object.hasOwn(HOST_OPERATION_SPECS, operation)) { + throw new Error(`Unknown Runtime Host operation grant: ${operation}`); + } + return operation; + }), + ); + return Object.freeze({ ...input, operationGrants }); +} diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts index cba4da8427..00a6f5e212 100644 --- a/packages/runtime-host/src/server/connection-session.ts +++ b/packages/runtime-host/src/server/connection-session.ts @@ -6,7 +6,7 @@ import { type HostOperationErrorCode, type RequestFrame, } from '../protocol/index.js'; -import type { FramedTransport } from '../transport/framed-transport.js'; +import type { RuntimeHostMessageTransport } from '../transport/message-transport.js'; import { dispatchOperation, operationFailureResponse, @@ -32,8 +32,12 @@ import type { HostSessionCatalogChangeService, SessionCatalogChangeConnection, } from './session-catalog-change-service.js'; +import type { RuntimeHostConnectionAuthority } from './connection-authority.js'; -type AcceptedConnectionContext = Omit; +type AcceptedConnectionContext = Omit & { + readonly clientInstanceId: string; + readonly authority: RuntimeHostConnectionAuthority; +}; export interface ConnectionOperationLease { acquireResidency(): OperationResidency; @@ -42,7 +46,7 @@ export interface ConnectionOperationLease { } export interface RuntimeHostConnectionSessionOptions { - transport: FramedTransport; + transport: RuntimeHostMessageTransport; connection: AcceptedConnectionContext; resolveHandlers(): OperationHandlerMap; resolveContinuity(): SessionContinuityService | undefined; @@ -109,7 +113,7 @@ export class RuntimeHostConnectionSession { if (this.#closed) return; this.#closed = true; this.#writer.close(); - this.#options.transport.destroyAfterFlush(); + this.#options.transport.closeAfterFlush(); this.#options.onTeardown(); } @@ -180,6 +184,7 @@ export class RuntimeHostConnectionSession { } const response = await dispatchOperation(frame, this.#options.resolveHandlers(), { ...this.#options.connection, + principal: this.#options.connection.authority.principalId, acquireResidency: () => admission.acquireResidency(), }); admission.seal(); @@ -309,7 +314,7 @@ export class RuntimeHostConnectionSession { this.#detachConfigurationChanges(); this.#detachSessionCatalogChanges(); this.#writer.close(); - this.#options.transport.destroy(); + this.#options.transport.abort(); this.#options.onTeardown(); } } diff --git a/packages/runtime-host/src/server/execution-candidate.ts b/packages/runtime-host/src/server/execution-candidate.ts index 1f9e41d222..916443c793 100644 --- a/packages/runtime-host/src/server/execution-candidate.ts +++ b/packages/runtime-host/src/server/execution-candidate.ts @@ -4,13 +4,10 @@ import { } from '@maka/storage/root-authority'; import type { RuntimeHostCandidateOptions } from './candidate.js'; import type { VerifiedGitRuntimeInput } from '@maka/storage/managed-workspace-owner'; -import { resolveBundledGitRuntime } from './bundled-git-runtime.js'; import { - createExecutionRuntimeHostComposition, - type CreateExecutionRuntimeHostCompositionOptions, - type ExecutionRuntimeHostComposition, -} from './execution-composition.js'; -import type { RuntimeHostCompositionContext } from './host-kernel.js'; + createExecutionRuntimeHostCompositionFactory, + type ExecutionRuntimeHostCompositionDependencies, +} from './execution-composition-factory.js'; import { RuntimeHostKernel } from './host-kernel.js'; export type ExecutionRuntimeHostCandidateResult = @@ -23,23 +20,16 @@ export interface ExecutionRuntimeHostCandidateOptions extends RuntimeHostCandida readonly bundledGitResourcesRoot?: string; } -export interface ExecutionRuntimeHostCandidateDependencies { - readonly createComposition?: ( - context: RuntimeHostCompositionContext, - options: CreateExecutionRuntimeHostCompositionOptions, - ) => Promise; -} +export type ExecutionRuntimeHostCandidateDependencies = ExecutionRuntimeHostCompositionDependencies; export async function startExecutionRuntimeHostCandidate( options: ExecutionRuntimeHostCandidateOptions, dependencies: ExecutionRuntimeHostCandidateDependencies = {}, ): Promise { - if (options.managedWorkspaceGitRuntime && options.bundledGitResourcesRoot) { - throw new Error('Managed workspace Git runtime must have exactly one authority'); - } - const managedWorkspaceGitRuntime = options.bundledGitResourcesRoot - ? await resolveBundledGitRuntime({ resourcesRoot: options.bundledGitResourcesRoot }) - : options.managedWorkspaceGitRuntime; + const compositionFactory = await createExecutionRuntimeHostCompositionFactory( + options, + dependencies, + ); const capability = await resolveExistingStorageRoot({ path: options.rootPath, kind: 'interactive', @@ -49,15 +39,10 @@ export async function startExecutionRuntimeHostCandidate( if (!owner) return { kind: 'loser' }; const host = await RuntimeHostKernel.start({ owner, + lifecycleMode: 'ephemeral', idleGraceMs: options.idleGraceMs, handshakeTimeoutMs: options.handshakeTimeoutMs, - compositionFactory: (context) => - (dependencies.createComposition ?? createExecutionRuntimeHostComposition)(context, { - ...(managedWorkspaceGitRuntime ? { managedWorkspaceGitRuntime } : {}), - ...(options.legacyConfigurationRoot - ? { legacyConfigurationRoot: options.legacyConfigurationRoot } - : {}), - }), + compositionFactory, }); return { kind: 'winner', host }; } diff --git a/packages/runtime-host/src/server/execution-composition-factory.ts b/packages/runtime-host/src/server/execution-composition-factory.ts new file mode 100644 index 0000000000..0d953dcec6 --- /dev/null +++ b/packages/runtime-host/src/server/execution-composition-factory.ts @@ -0,0 +1,43 @@ +import type { VerifiedGitRuntimeInput } from '@maka/storage/managed-workspace-owner'; +import { resolveBundledGitRuntime } from './bundled-git-runtime.js'; +import { + createExecutionRuntimeHostComposition, + type ExecutionRuntimeHostComposition, +} from './execution-composition.js'; +import type { + RuntimeHostCompositionContext, + RuntimeHostCompositionFactory, +} from './host-kernel.js'; + +export interface ExecutionRuntimeHostCompositionSourceOptions { + readonly managedWorkspaceGitRuntime?: VerifiedGitRuntimeInput; + readonly bundledGitResourcesRoot?: string; + readonly legacyConfigurationRoot?: string; +} + +export interface ExecutionRuntimeHostCompositionDependencies { + readonly createComposition?: ( + context: RuntimeHostCompositionContext, + options: Parameters[1], + ) => Promise; +} + +export async function createExecutionRuntimeHostCompositionFactory( + options: ExecutionRuntimeHostCompositionSourceOptions, + dependencies: ExecutionRuntimeHostCompositionDependencies = {}, +): Promise { + if (options.managedWorkspaceGitRuntime && options.bundledGitResourcesRoot) { + throw new Error('Managed workspace Git runtime must have exactly one authority'); + } + const managedWorkspaceGitRuntime = options.bundledGitResourcesRoot + ? await resolveBundledGitRuntime({ resourcesRoot: options.bundledGitResourcesRoot }) + : options.managedWorkspaceGitRuntime; + const compositionOptions = { + ...(managedWorkspaceGitRuntime ? { managedWorkspaceGitRuntime } : {}), + ...(options.legacyConfigurationRoot + ? { legacyConfigurationRoot: options.legacyConfigurationRoot } + : {}), + }; + const createComposition = dependencies.createComposition ?? createExecutionRuntimeHostComposition; + return (context) => createComposition(context, compositionOptions); +} diff --git a/packages/runtime-host/src/server/execution-service.ts b/packages/runtime-host/src/server/execution-service.ts new file mode 100644 index 0000000000..f6b305a2cb --- /dev/null +++ b/packages/runtime-host/src/server/execution-service.ts @@ -0,0 +1,47 @@ +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import type { VerifiedGitRuntimeInput } from '@maka/storage/managed-workspace-owner'; +import { + createExecutionRuntimeHostCompositionFactory, + type ExecutionRuntimeHostCompositionDependencies, +} from './execution-composition-factory.js'; +import { RuntimeHostKernel } from './host-kernel.js'; + +export interface ExecutionRuntimeHostServiceOptions { + readonly rootPath: string; + readonly legacyConfigurationRoot?: string; + readonly managedWorkspaceGitRuntime?: VerifiedGitRuntimeInput; + readonly bundledGitResourcesRoot?: string; + readonly handshakeTimeoutMs?: number; + readonly shutdownGraceMs?: number; +} + +export type ExecutionRuntimeHostServiceDependencies = ExecutionRuntimeHostCompositionDependencies; + +export class RuntimeHostRootAlreadyOwnedError extends Error { + readonly code = 'root_already_owned'; + + constructor(readonly rootPath: string) { + super(`Runtime Host root is already owned: ${rootPath}`); + this.name = 'RuntimeHostRootAlreadyOwnedError'; + } +} + +export async function startExecutionRuntimeHostService( + options: ExecutionRuntimeHostServiceOptions, + dependencies: ExecutionRuntimeHostServiceDependencies = {}, +): Promise { + const compositionFactory = await createExecutionRuntimeHostCompositionFactory( + options, + dependencies, + ); + const capability = await resolveStorageRoot({ path: options.rootPath, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + if (!owner) throw new RuntimeHostRootAlreadyOwnedError(capability.canonicalPath); + return RuntimeHostKernel.start({ + owner, + lifecycleMode: 'service', + handshakeTimeoutMs: options.handshakeTimeoutMs, + shutdownGraceMs: options.shutdownGraceMs, + compositionFactory, + }); +} diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index 84566ca2d8..3f76813a1c 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -1,15 +1,14 @@ import { randomUUID } from 'node:crypto'; -import { createServer, type Server, type Socket } from 'node:net'; import { arch as osArch, release as osRelease } from 'node:os'; import { assertInteractiveRootOwner, authenticateInteractiveRootOwner, type InteractiveRootOwner, } from '@maka/storage/root-authority'; -import { prepareRuntimeHostEndpoint, type RuntimeHostEndpoint } from '../control/endpoint.js'; import { removeHostRegistration, writeHostRegistration } from '../control/registration.js'; import { decodeClientFrame, + encodeProtocolMessage, HOST_OPERATION_SPECS, negotiateProtocol, RUNTIME_HOST_COMPATIBILITY_EPOCH, @@ -23,7 +22,7 @@ import { type HostStatusResult, type RequestFrame, } from '../protocol/index.js'; -import { FramedTransport } from '../transport/framed-transport.js'; +import type { RuntimeHostMessageTransport } from '../transport/message-transport.js'; import { RuntimeHostConnectionSession, type ConnectionOperationLease, @@ -40,6 +39,12 @@ import type { ClientCapabilityService } from './client-capability-service.js'; import type { HostConfigurationChangeService } from './configuration-change-service.js'; import { runtimeHostLogBuffer } from '../process-diagnostics.js'; import type { HostSessionCatalogChangeService } from './session-catalog-change-service.js'; +import { + startLocalRuntimeHostListenerSet, + type RuntimeHostListenerConnection, + type RuntimeHostListenerSet, + type RuntimeHostListenerSetFactory, +} from './listener-set.js'; const DEFAULT_IDLE_GRACE_MS = 30_000; const DEFAULT_HANDSHAKE_TIMEOUT_MS = 5_000; @@ -87,29 +92,40 @@ export type RuntimeHostCompositionFactory = ( context: RuntimeHostCompositionContext, ) => Promise; -export interface RuntimeHostKernelOptions { +interface RuntimeHostKernelCommonOptions { owner: InteractiveRootOwner; - idleGraceMs?: number; handshakeTimeoutMs?: number; shutdownGraceMs?: number; compositionFactory?: RuntimeHostCompositionFactory; + listenerSetFactory?: RuntimeHostListenerSetFactory; } +export type RuntimeHostLifecycleMode = 'ephemeral' | 'service'; + +export type RuntimeHostKernelOptions = RuntimeHostKernelCommonOptions & + ( + | { lifecycleMode?: 'ephemeral'; idleGraceMs?: number } + | { lifecycleMode: 'service'; idleGraceMs?: never } + ); + +type RuntimeHostLifecycle = + | { readonly kind: 'ephemeral'; readonly idleGraceMs: number } + | { readonly kind: 'service' }; + export class RuntimeHostKernel { readonly hostEpoch = randomUUID(); readonly closed: Promise; readonly #options: RuntimeHostKernelOptions; readonly #createdAt = new Date().toISOString(); - readonly #server: Server; - readonly #handshakingTransports = new Set(); - readonly #acceptedTransports = new Set(); + readonly #handshakingTransports = new Set(); + readonly #acceptedTransports = new Set(); readonly #connectionSessions = new Set(); readonly #operationDrainWaiters = new Set<() => void>(); readonly #residencyDrainWaiters = new Set<() => void>(); - readonly #idleGraceMs: number; + readonly #lifecycle: RuntimeHostLifecycle; readonly #handshakeTimeoutMs: number; readonly #shutdownGraceMs: number; - #endpoint: RuntimeHostEndpoint | undefined; + #listeners: RuntimeHostListenerSet | undefined; #state: HostLifecycleState = 'starting'; #activeOperations = 0; #activeCommandOperations = 0; @@ -128,14 +144,13 @@ export class RuntimeHostKernel { #rejectClosed!: (error: unknown) => void; private constructor(options: RuntimeHostKernelOptions) { - assertDuration(options.idleGraceMs ?? DEFAULT_IDLE_GRACE_MS, 'idleGraceMs', 0); + this.#lifecycle = normalizeLifecycle(options); assertDuration( options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS, 'handshakeTimeoutMs', 1, ); assertDuration(options.shutdownGraceMs ?? DEFAULT_SHUTDOWN_GRACE_MS, 'shutdownGraceMs', 1); - this.#idleGraceMs = options.idleGraceMs ?? DEFAULT_IDLE_GRACE_MS; this.#handshakeTimeoutMs = options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS; this.#shutdownGraceMs = options.shutdownGraceMs ?? DEFAULT_SHUTDOWN_GRACE_MS; this.#options = options; @@ -146,25 +161,18 @@ export class RuntimeHostKernel { this.#resolveClosed = resolve; this.#rejectClosed = reject; }); - this.#server = createServer({ allowHalfOpen: true }, (socket) => this.#accept(socket)); } static async start(options: RuntimeHostKernelOptions): Promise { const owner = authenticateInteractiveRootOwner(options.owner); let host: RuntimeHostKernel | undefined; try { - host = new RuntimeHostKernel({ - owner, - idleGraceMs: options.idleGraceMs, - handshakeTimeoutMs: options.handshakeTimeoutMs, - shutdownGraceMs: options.shutdownGraceMs, - compositionFactory: options.compositionFactory, - }); + host = new RuntimeHostKernel({ ...options, owner }); await host.#start(); return host; } catch (error) { if (host) { - if (host.#endpoint) { + if (host.#listeners) { host.#requestDrain(); try { await host.closed; @@ -186,8 +194,8 @@ export class RuntimeHostKernel { } get endpoint(): string { - if (!this.#endpoint) throw new Error('Runtime Host has not started listening'); - return this.#endpoint.path; + if (!this.#listeners) throw new Error('Runtime Host has not started listening'); + return this.#listeners.localEndpoint; } get connectionCount(): number { @@ -211,12 +219,11 @@ export class RuntimeHostKernel { async #start(): Promise { await assertInteractiveRootOwner(this.#options.owner); - this.#endpoint = await prepareRuntimeHostEndpoint({ + this.#listeners = await (this.#options.listenerSetFactory ?? startLocalRuntimeHostListenerSet)({ rootId: this.#options.owner.capability.rootId, hostEpoch: this.hostEpoch, + accept: (connection) => this.#accept(connection), }); - await listen(this.#server, this.#endpoint.path); - await this.#endpoint.prepareAfterListen(); await this.#publishRegistration(); const compositionFactory = this.#options.compositionFactory; if (compositionFactory) { @@ -254,15 +261,16 @@ export class RuntimeHostKernel { this.#scheduleIdleIfNeeded(); } - #accept(socket: Socket): void { - const transport = new FramedTransport(socket); + #accept(connection: RuntimeHostListenerConnection): void { + const { transport } = connection; this.#handshakingTransports.add(transport); - void this.#serveConnection(transport).finally(() => { + void this.#serveConnection(connection).finally(() => { this.#handshakingTransports.delete(transport); }); } - async #serveConnection(transport: FramedTransport): Promise { + async #serveConnection(connection: RuntimeHostListenerConnection): Promise { + const { authority, transport } = connection; let transportReleased = false; let connectionId: string | undefined; const releaseTransport = () => { @@ -277,9 +285,9 @@ export class RuntimeHostKernel { } const result = await this.#admitHandshake(frame, transport); connectionId = result.kind === 'accepted' ? result.connectionId : undefined; - await transport.write(result); + await transport.write(encodeProtocolMessage(result)); if (result.kind !== 'accepted') { - transport.destroyAfterFlush(); + transport.closeAfterFlush(); return; } const session = new RuntimeHostConnectionSession({ @@ -287,8 +295,9 @@ export class RuntimeHostKernel { connection: { hostEpoch: this.hostEpoch, connectionId: result.connectionId, + clientInstanceId: frame.clientInstanceId, surface: frame.surface, - principal: 'local_os_user', + authority, }, resolveHandlers: () => this.#operationHandlers, resolveContinuity: () => this.#composition?.continuity, @@ -305,7 +314,7 @@ export class RuntimeHostKernel { this.#connectionSessions.delete(session); } } catch { - transport.destroy(); + transport.abort(); } finally { try { if (connectionId) this.#composition?.releaseConnection?.(connectionId); @@ -317,7 +326,7 @@ export class RuntimeHostKernel { async #admitHandshake( hello: ClientHello, - transport: FramedTransport, + transport: RuntimeHostMessageTransport, ): Promise { const admittedState = await this.#readAdmissionState(); if (!admittedState) { @@ -338,7 +347,10 @@ export class RuntimeHostKernel { protocolMax: HOST_PROTOCOL.max, compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH, state: admittedState, - replacement: this.#isTrueIdle() ? 'wait_for_idle_exit' : 'blocked_by_residency', + replacement: + this.#lifecycle.kind === 'ephemeral' && this.#isTrueIdle() + ? 'wait_for_idle_exit' + : 'blocked_by_residency', }; } this.#acceptedTransports.add(transport); @@ -354,7 +366,7 @@ export class RuntimeHostKernel { }; } - #releaseConnection(transport: FramedTransport): void { + #releaseConnection(transport: RuntimeHostMessageTransport): void { if (!this.#acceptedTransports.delete(transport)) { throw new Error('Runtime Host connection residency underflow'); } @@ -515,13 +527,14 @@ export class RuntimeHostKernel { } #scheduleIdleIfNeeded(): void { + if (this.#lifecycle.kind === 'service') return; if (this.#shutdownRequested) return; if (!this.#isTrueIdle() || this.#idleTimer) return; this.#idleTimer = setTimeout(() => { this.#idleTimer = undefined; if (!this.#isTrueIdle()) return; void this.#commitShutdown().catch(() => undefined); - }, this.#idleGraceMs); + }, this.#lifecycle.idleGraceMs); } #isTrueIdle(): boolean { @@ -603,7 +616,9 @@ export class RuntimeHostKernel { // shutdown deadline may expire while publishing the draining registration; // leaving the listener open in that case strands an unreachable, ref'ed // server until the process is forcibly terminated. - const serverClosed = closeServer(this.#server).catch((error: unknown) => errors.push(error)); + const listenerClosed = this.#listeners + ?.closeAdmission() + .catch((error: unknown) => errors.push(error)); await this.#publishRegistration().catch((error: unknown) => errors.push(error)); this.#assertShutdownCanContinue(); const accepted = [...this.#acceptedTransports]; @@ -615,9 +630,9 @@ export class RuntimeHostKernel { ]); this.#assertShutdownCanContinue(); if (!operationsDrained) { - for (const transport of accepted) transport.destroy(); + for (const transport of accepted) transport.abort(); } - for (const transport of handshaking) transport.destroy(); + for (const transport of handshaking) transport.abort(); await operationDrain; this.#assertShutdownCanContinue(); await this.#compositionStartup; @@ -626,10 +641,10 @@ export class RuntimeHostKernel { this.#assertShutdownCanContinue(); await this.#waitForResidencies(); this.#assertShutdownCanContinue(); - for (const transport of accepted) transport.destroy(); - await serverClosed; + for (const transport of accepted) transport.abort(); + await listenerClosed; this.#assertShutdownCanContinue(); - await this.#endpoint?.cleanup().catch((error: unknown) => errors.push(error)); + await this.#listeners?.cleanup().catch((error: unknown) => errors.push(error)); this.#assertShutdownCanContinue(); await removeHostRegistration(this.#options.owner.controlDirectory, this.hostEpoch).catch( (error: unknown) => errors.push(error), @@ -647,10 +662,10 @@ export class RuntimeHostKernel { async #abortStartup(): Promise { this.#state = 'draining'; - for (const transport of this.#handshakingTransports) transport.destroy(); - for (const transport of this.#acceptedTransports) transport.destroy(); - await closeServer(this.#server).catch(() => undefined); - await this.#endpoint?.cleanup().catch(() => undefined); + for (const transport of this.#handshakingTransports) transport.abort(); + for (const transport of this.#acceptedTransports) transport.abort(); + await this.#listeners?.closeAdmission().catch(() => undefined); + await this.#listeners?.cleanup().catch(() => undefined); await removeHostRegistration(this.#options.owner.controlDirectory, this.hostEpoch).catch( () => undefined, ); @@ -676,34 +691,8 @@ export class RuntimeHostKernel { } } -function listen(server: Server, path: string): Promise { - return new Promise((resolve, reject) => { - const onError = (error: Error) => { - server.off('listening', onListening); - reject(error); - }; - const onListening = () => { - server.off('error', onError); - resolve(); - }; - server.once('error', onError); - server.once('listening', onListening); - server.listen(path); - }); -} - -function closeServer(server: Server): Promise { - if (!server.listening) return Promise.resolve(); - return new Promise((resolve, reject) => { - server.close((error) => { - if (error) reject(error); - else resolve(); - }); - }); -} - async function waitForTransportClose( - transports: readonly FramedTransport[], + transports: readonly RuntimeHostMessageTransport[], timeoutMs: number, ): Promise { if (transports.length === 0) return; @@ -735,3 +724,19 @@ function assertDuration(value: number, label: string, minimum: 0 | 1): void { throw new RangeError(`${label} must be an integer between ${minimum} and 120000`); } } + +function normalizeLifecycle(options: RuntimeHostKernelOptions): RuntimeHostLifecycle { + const lifecycleMode: unknown = options.lifecycleMode; + if (lifecycleMode === 'service') { + if (Object.hasOwn(options, 'idleGraceMs')) { + throw new TypeError('Runtime Host service lifecycle does not accept idleGraceMs'); + } + return { kind: 'service' }; + } + if (lifecycleMode !== undefined && lifecycleMode !== 'ephemeral') { + throw new TypeError('Runtime Host lifecycleMode must be ephemeral or service'); + } + const idleGraceMs = options.idleGraceMs ?? DEFAULT_IDLE_GRACE_MS; + assertDuration(idleGraceMs, 'idleGraceMs', 0); + return { kind: 'ephemeral', idleGraceMs }; +} diff --git a/packages/runtime-host/src/server/index.ts b/packages/runtime-host/src/server/index.ts index 39fb708d7d..519f06eeea 100644 --- a/packages/runtime-host/src/server/index.ts +++ b/packages/runtime-host/src/server/index.ts @@ -5,6 +5,7 @@ export { type RuntimeHostCompositionContext, type RuntimeHostCompositionFactory, type RuntimeHostKernelOptions, + type RuntimeHostLifecycleMode, type RuntimeHostResidency, } from './host-kernel.js'; export { @@ -13,3 +14,30 @@ export { type RuntimeHostCandidateResult, } from './candidate.js'; export { createUnavailableDomainOperationHandlers } from './operation-dispatcher.js'; +export { + RuntimeHostRootAlreadyOwnedError, + startExecutionRuntimeHostService, + type ExecutionRuntimeHostServiceDependencies, + type ExecutionRuntimeHostServiceOptions, +} from './execution-service.js'; +export { + runRuntimeHostProcessLifecycle, + type RuntimeHostProcessLifecycleOptions, +} from './process-lifecycle.js'; +export { installRuntimeHostLogCapture } from '../process-diagnostics.js'; +export { + createRuntimeHostListenerSet, + startLocalRuntimeHostListenerSet, + type RuntimeHostListenerSet, + type RuntimeHostListener, + type RuntimeHostListenerConnection, + type RuntimeHostListenerKind, + type RuntimeHostListenerSetFactory, + type RuntimeHostListenerSetFactoryInput, +} from './listener-set.js'; +export { + createRuntimeHostConnectionAuthority, + LOCAL_OWNER_CONNECTION_AUTHORITY, + type RuntimeHostConnectionAuthority, +} from './connection-authority.js'; +export type { RuntimeHostMessageTransport } from '../transport/message-transport.js'; diff --git a/packages/runtime-host/src/server/listener-set.ts b/packages/runtime-host/src/server/listener-set.ts new file mode 100644 index 0000000000..c386896af7 --- /dev/null +++ b/packages/runtime-host/src/server/listener-set.ts @@ -0,0 +1,67 @@ +import { startLocalIpcRuntimeHostListener } from './local-ipc-listener.js'; +import type { RuntimeHostMessageTransport } from '../transport/message-transport.js'; +import type { RuntimeHostConnectionAuthority } from './connection-authority.js'; + +export interface RuntimeHostListenerConnection { + readonly transport: RuntimeHostMessageTransport; + readonly authority: RuntimeHostConnectionAuthority; +} + +export interface RuntimeHostListener { + readonly kind: RuntimeHostListenerKind; + readonly endpoint: string; + closeAdmission(): Promise; + cleanup(): Promise; +} + +export type RuntimeHostListenerKind = 'local_ipc' | 'websocket'; + +export interface RuntimeHostListenerSet { + readonly listeners: readonly RuntimeHostListener[]; + readonly localEndpoint: string; + closeAdmission(): Promise; + cleanup(): Promise; +} + +export interface RuntimeHostListenerSetFactoryInput { + readonly rootId: string; + readonly hostEpoch: string; + readonly accept: (connection: RuntimeHostListenerConnection) => void; +} + +export type RuntimeHostListenerSetFactory = ( + input: RuntimeHostListenerSetFactoryInput, +) => Promise; + +export async function startLocalRuntimeHostListenerSet( + input: RuntimeHostListenerSetFactoryInput, +): Promise { + const local = await startLocalIpcRuntimeHostListener(input); + return createRuntimeHostListenerSet(local); +} + +export function createRuntimeHostListenerSet( + local: RuntimeHostListener & { readonly kind: 'local_ipc' }, + additional: readonly RuntimeHostListener[] = [], +): RuntimeHostListenerSet { + const listeners = Object.freeze([local, ...additional]); + return { + listeners, + localEndpoint: local.endpoint, + closeAdmission: () => settleListeners(listeners, (listener) => listener.closeAdmission()), + cleanup: () => settleListeners([...listeners].reverse(), (listener) => listener.cleanup()), + }; +} + +async function settleListeners( + listeners: readonly RuntimeHostListener[], + operation: (listener: RuntimeHostListener) => Promise, +): Promise { + const outcomes = await Promise.allSettled(listeners.map(operation)); + const errors = outcomes.flatMap((outcome) => + outcome.status === 'rejected' ? [outcome.reason] : [], + ); + if (errors.length > 0) { + throw new AggregateError(errors, 'Runtime Host listener set operation failed'); + } +} diff --git a/packages/runtime-host/src/server/local-ipc-listener.ts b/packages/runtime-host/src/server/local-ipc-listener.ts new file mode 100644 index 0000000000..972f991fa4 --- /dev/null +++ b/packages/runtime-host/src/server/local-ipc-listener.ts @@ -0,0 +1,99 @@ +import { createServer, type Server } from 'node:net'; +import { prepareRuntimeHostEndpoint, type RuntimeHostEndpoint } from '../control/endpoint.js'; +import { FramedTransport } from '../transport/framed-transport.js'; +import { LOCAL_OWNER_CONNECTION_AUTHORITY } from './connection-authority.js'; +import type { RuntimeHostListener, RuntimeHostListenerConnection } from './listener-set.js'; + +export interface StartLocalIpcRuntimeHostListenerOptions { + readonly rootId: string; + readonly hostEpoch: string; + readonly accept: (connection: RuntimeHostListenerConnection) => void; +} + +export async function startLocalIpcRuntimeHostListener( + options: StartLocalIpcRuntimeHostListenerOptions, +): Promise { + const endpoint = await prepareRuntimeHostEndpoint({ + rootId: options.rootId, + hostEpoch: options.hostEpoch, + }); + const startupTransports = new Set(); + let starting = true; + const server = createServer({ allowHalfOpen: true }, (socket) => { + const transport = new FramedTransport(socket); + if (starting) { + startupTransports.add(transport); + void transport.closed.then(() => startupTransports.delete(transport)); + } + try { + options.accept({ transport, authority: LOCAL_OWNER_CONNECTION_AUTHORITY }); + } catch (error) { + transport.abort(asError(error)); + } + }); + try { + await listen(server, endpoint.path); + await endpoint.prepareAfterListen(); + starting = false; + startupTransports.clear(); + return new LocalIpcRuntimeHostListener(server, endpoint); + } catch (error) { + for (const transport of startupTransports) transport.abort(); + await closeServer(server).catch(() => undefined); + await endpoint.cleanup().catch(() => undefined); + throw error; + } +} + +class LocalIpcRuntimeHostListener implements RuntimeHostListener { + readonly kind = 'local_ipc' as const; + readonly endpoint: string; + readonly #server: Server; + readonly #runtimeEndpoint: RuntimeHostEndpoint; + #closeTask: Promise | undefined; + + constructor(server: Server, endpoint: RuntimeHostEndpoint) { + this.#server = server; + this.#runtimeEndpoint = endpoint; + this.endpoint = endpoint.path; + } + + closeAdmission(): Promise { + this.#closeTask ??= closeServer(this.#server); + return this.#closeTask; + } + + cleanup(): Promise { + return this.#runtimeEndpoint.cleanup(); + } +} + +function listen(server: Server, path: string): Promise { + return new Promise((resolve, reject) => { + const onError = (error: Error) => { + server.off('listening', onListening); + reject(error); + }; + const onListening = () => { + server.off('error', onError); + resolve(); + }; + server.once('error', onError); + server.once('listening', onListening); + server.listen(path); + }); +} + +function closeServer(server: Server): Promise { + if (!server.listening) return Promise.resolve(); + return new Promise((resolve, reject) => { + server.close((error) => { + if (error) reject(error); + else resolve(); + }); + }); +} + +function asError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index 10d9c32c07..ffd4f2ccfb 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -17,7 +17,7 @@ export interface ConnectionContext { hostEpoch: string; connectionId: string; surface: ClientSurface; - principal: 'local_os_user'; + principal: string; acquireResidency(): OperationResidency; } diff --git a/packages/runtime-host/src/server/serial-outbound-writer.ts b/packages/runtime-host/src/server/serial-outbound-writer.ts index b0a5e9c547..a2891a77fc 100644 --- a/packages/runtime-host/src/server/serial-outbound-writer.ts +++ b/packages/runtime-host/src/server/serial-outbound-writer.ts @@ -1,11 +1,15 @@ -import { encodeProtocolFrame, type HostFrame } from '../protocol/index.js'; -import type { FramedTransport } from '../transport/framed-transport.js'; +import { + encodeProtocolMessage, + type EncodedProtocolMessage, + type HostFrame, +} from '../protocol/index.js'; +import type { RuntimeHostMessageTransport } from '../transport/message-transport.js'; const MAX_QUEUED_FRAMES = 64; const MAX_QUEUED_BYTES = 2 * 1024 * 1024; interface QueuedFrame { - encoded: Buffer; + message: EncodedProtocolMessage; resolve(): void; reject(error: Error): void; } @@ -26,7 +30,7 @@ export class RuntimeHostOutboundQueueError extends Error { } export class BoundedSerialOutboundWriter { - readonly #transport: FramedTransport; + readonly #transport: RuntimeHostMessageTransport; readonly #onFailure: () => void; readonly #queue: QueuedFrame[] = []; #queuedBytes = 0; @@ -34,7 +38,7 @@ export class BoundedSerialOutboundWriter { #drainTask: Promise | undefined; #closed = false; - constructor(transport: FramedTransport, onFailure: () => void) { + constructor(transport: RuntimeHostMessageTransport, onFailure: () => void) { this.#transport = transport; this.#onFailure = onFailure; } @@ -44,9 +48,9 @@ export class BoundedSerialOutboundWriter { throw new Error('Runtime Host outbound writer is closed'); } - let encoded: Buffer; + let message: EncodedProtocolMessage; try { - encoded = encodeProtocolFrame(frame); + message = encodeProtocolMessage(frame); } catch (error) { const failure = asError(error); this.#fail(failure); @@ -57,15 +61,15 @@ export class BoundedSerialOutboundWriter { this.#fail(failure); throw failure; } - if (this.#queuedBytes + encoded.byteLength > MAX_QUEUED_BYTES) { + if (this.#queuedBytes + message.byteLength > MAX_QUEUED_BYTES) { const failure = new RuntimeHostOutboundQueueError('byte_limit'); this.#fail(failure); throw failure; } const flushed = new Promise((resolve, reject) => { - this.#queue.push({ encoded, resolve, reject }); - this.#queuedBytes += encoded.byteLength; + this.#queue.push({ message, resolve, reject }); + this.#queuedBytes += message.byteLength; if (!this.#writing) { this.#writing = true; this.#drainTask = this.#drain(); @@ -93,14 +97,14 @@ export class BoundedSerialOutboundWriter { const queued = this.#queue[0]; if (!queued) return; try { - await this.#transport.writeEncoded(queued.encoded); + await this.#transport.write(queued.message); } catch (error) { this.#fail(asError(error)); return; } if (this.#closed) return; this.#queue.shift(); - this.#queuedBytes -= queued.encoded.byteLength; + this.#queuedBytes -= queued.message.byteLength; queued.resolve(); } } catch (error) { diff --git a/packages/runtime-host/src/server/session-continuity-coordinator.ts b/packages/runtime-host/src/server/session-continuity-coordinator.ts index 3729455157..17d4e4727a 100644 --- a/packages/runtime-host/src/server/session-continuity-coordinator.ts +++ b/packages/runtime-host/src/server/session-continuity-coordinator.ts @@ -3,8 +3,8 @@ import { isDeepStrictEqual } from 'node:util'; import type { SessionEvent, ShellRunUpdate } from '@maka/core/events'; import type { StoredMessage } from '@maka/core/session'; import { - encodeProtocolFrame, - RUNTIME_HOST_MAX_FRAME_BYTES, + encodeProtocolMessage, + RUNTIME_HOST_MAX_MESSAGE_BYTES, SESSION_LIVE_DELTA_MAX_BYTES, SESSION_RUNTIME_RESOURCE_PTY_DATA_MAX_BYTES, SESSION_RUNTIME_RESOURCE_CHANGES_MAX, @@ -806,7 +806,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { if (subscriber.phase !== 'open' || subscriber.terminalQueued) return; let encodedBytes: number; try { - encodedBytes = encodeProtocolFrame(frame).byteLength; + encodedBytes = encodeProtocolMessage(frame).byteLength; } catch { this.#evictSlowSubscriber(subscriber); return; @@ -841,7 +841,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { }; subscriber.nextSequence += 1; subscriber.terminalQueued = true; - const encodedBytes = encodeProtocolFrame(frame).byteLength; + const encodedBytes = encodeProtocolMessage(frame).byteLength; if (inFlight) { subscriber.queue.push(inFlight); subscriber.queuedBytes += inFlight.encodedBytes; @@ -914,7 +914,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { sequence: subscriber.nextSequence, reason: 'session_removed', }; - const encodedBytes = encodeProtocolFrame(frame).byteLength; + const encodedBytes = encodeProtocolMessage(frame).byteLength; if ( subscriber.queue.length >= MAX_SUBSCRIBER_QUEUED_FRAMES || subscriber.queuedBytes + encodedBytes > MAX_SUBSCRIBER_QUEUED_BYTES @@ -1072,7 +1072,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { } function slowConsumerFrameBytes(subscriber: Subscriber, hostEpoch: string): number { - return encodeProtocolFrame({ + return encodeProtocolMessage({ kind: 'subscription.closed', hostEpoch, subscriptionId: subscriber.subscriptionId, @@ -1137,7 +1137,7 @@ function transcriptSubscriptionNotFound(): OperationOutcome<'session.transcript. function terminalFrameByteBudget(subscriber: Subscriber, hostEpoch: string): number { return Math.max( slowConsumerFrameBytes(subscriber, hostEpoch), - encodeProtocolFrame({ + encodeProtocolMessage({ kind: 'subscription.closed', hostEpoch, subscriptionId: subscriber.subscriptionId, @@ -1180,7 +1180,7 @@ function isTerminalTurn(turn: TurnSnapshot): boolean { } function wireTextByteLimit(frame: SessionDeltaFrame): number { - return RUNTIME_HOST_MAX_FRAME_BYTES - encodeProtocolFrame(frame).byteLength; + return RUNTIME_HOST_MAX_MESSAGE_BYTES - encodeProtocolMessage(frame).byteLength; } function jsonStringContentBytes(value: string): number { diff --git a/packages/runtime-host/src/transport/framed-transport.ts b/packages/runtime-host/src/transport/framed-transport.ts index 47bde21a6e..fec2b44a47 100644 --- a/packages/runtime-host/src/transport/framed-transport.ts +++ b/packages/runtime-host/src/transport/framed-transport.ts @@ -1,12 +1,11 @@ import type { Socket } from 'node:net'; import { - encodeProtocolFrame, - ProtocolFrameDecoder, - RUNTIME_HOST_MAX_FRAME_BYTES, + RUNTIME_HOST_MAX_MESSAGE_BYTES, RuntimeHostProtocolError, - type ClientFrame, - type HostFrame, + type EncodedProtocolMessage, } from '../protocol/index.js'; +import { frameLocalIpcProtocolMessage, LocalIpcProtocolFrameDecoder } from './local-ipc-framing.js'; +import type { RuntimeHostMessageTransport } from './message-transport.js'; const MAX_QUEUED_FRAMES = 64; const MAX_QUEUED_BYTES = 2 * 1024 * 1024; @@ -39,9 +38,9 @@ export class RuntimeHostTransportError extends Error { } } -export class FramedTransport { +export class FramedTransport implements RuntimeHostMessageTransport { readonly closed: Promise; - readonly #decoder = new ProtocolFrameDecoder(); + readonly #decoder = new LocalIpcProtocolFrameDecoder(); readonly #queue: QueuedFrame[] = []; #queuedBytes = 0; #buffered = Buffer.alloc(0); @@ -106,29 +105,22 @@ export class FramedTransport { }); } - write(frame: ClientFrame | HostFrame): Promise { - try { - return this.writeEncoded(encodeProtocolFrame(frame)); - } catch (error) { - return Promise.reject(error); - } - } - - writeEncoded(encoded: Uint8Array): Promise { + write(message: EncodedProtocolMessage): Promise { if (this.#failure) return Promise.reject(this.#failure); + const frame = frameLocalIpcProtocolMessage(message); return new Promise((resolve, reject) => { - this.socket.write(encoded, (error) => { + this.socket.write(frame, (error) => { if (error) reject(error); else resolve(); }); }); } - destroyAfterFlush(): void { + closeAfterFlush(): void { this.socket.destroySoon(); } - destroy(error?: Error): void { + abort(error?: Error): void { this.socket.destroy(error); } @@ -149,10 +141,10 @@ export class FramedTransport { while (true) { const newline = this.#buffered.indexOf(0x0a); if (newline === -1) { - if (this.#buffered.byteLength > RUNTIME_HOST_MAX_FRAME_BYTES) { + if (this.#buffered.byteLength > RUNTIME_HOST_MAX_MESSAGE_BYTES) { throw new RuntimeHostProtocolError( 'frame_too_large', - 'Runtime Host frame exceeds the byte limit', + 'Runtime Host message exceeds the byte limit', ); } break; diff --git a/packages/runtime-host/src/transport/local-ipc-framing.ts b/packages/runtime-host/src/transport/local-ipc-framing.ts new file mode 100644 index 0000000000..269c7b74cf --- /dev/null +++ b/packages/runtime-host/src/transport/local-ipc-framing.ts @@ -0,0 +1,64 @@ +import { TextDecoder } from 'node:util'; +import { + RUNTIME_HOST_MAX_MESSAGE_BYTES, + RuntimeHostProtocolError, + type EncodedProtocolMessage, +} from '../protocol/index.js'; + +export function frameLocalIpcProtocolMessage(message: EncodedProtocolMessage): Buffer { + return Buffer.concat([message, Buffer.from('\n')]); +} + +export class LocalIpcProtocolFrameDecoder { + readonly #decoder = new TextDecoder('utf-8', { fatal: true }); + #pending = Buffer.alloc(0); + + push(chunk: Uint8Array): unknown[] { + const frames: unknown[] = []; + let offset = 0; + while (offset < chunk.byteLength) { + const newline = chunk.indexOf(0x0a, offset); + const end = newline === -1 ? chunk.byteLength : newline; + const segment = Buffer.from(chunk.subarray(offset, end)); + if (this.#pending.byteLength + segment.byteLength > RUNTIME_HOST_MAX_MESSAGE_BYTES) { + throw new RuntimeHostProtocolError( + 'frame_too_large', + 'Runtime Host message exceeds the byte limit', + ); + } + if (segment.byteLength > 0) this.#pending = Buffer.concat([this.#pending, segment]); + if (newline === -1) break; + frames.push(this.#decodePending()); + this.#pending = Buffer.alloc(0); + offset = newline + 1; + } + return frames; + } + + end(): void { + if (this.#pending.byteLength !== 0) { + throw new RuntimeHostProtocolError( + 'invalid_frame', + 'Runtime Host stream ended with a partial frame', + ); + } + } + + #decodePending(): unknown { + if (this.#pending.byteLength === 0) { + throw new RuntimeHostProtocolError('invalid_frame', 'Runtime Host frame is empty'); + } + let text: string; + try { + const bytes = this.#pending.at(-1) === 0x0d ? this.#pending.subarray(0, -1) : this.#pending; + text = this.#decoder.decode(bytes); + } catch { + throw new RuntimeHostProtocolError('invalid_utf8', 'Runtime Host frame is not valid UTF-8'); + } + try { + return JSON.parse(text) as unknown; + } catch { + throw new RuntimeHostProtocolError('invalid_json', 'Runtime Host frame is not valid JSON'); + } + } +} diff --git a/packages/runtime-host/src/transport/message-transport.ts b/packages/runtime-host/src/transport/message-transport.ts new file mode 100644 index 0000000000..06aa336c2d --- /dev/null +++ b/packages/runtime-host/src/transport/message-transport.ts @@ -0,0 +1,9 @@ +import type { EncodedProtocolMessage } from '../protocol/index.js'; + +export interface RuntimeHostMessageTransport { + readonly closed: Promise; + read(timeoutMs: number): Promise; + write(message: EncodedProtocolMessage): Promise; + closeAfterFlush(): void; + abort(error?: Error): void; +}