From 8b3c9ab31b3a049e13bb046f91a2b31f2185f66b Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 23 Jul 2026 02:08:25 +0800 Subject: [PATCH 1/4] feat(runtime-host): establish protocol and admission foundations --- .../src/__tests__/execution-host.test.ts | 85 +++- .../src/__tests__/host-kernel.test.ts | 82 ++++ .../__tests__/operation-dispatcher.test.ts | 150 +++++++ .../src/__tests__/protocol.test.ts | 24 + .../__tests__/root-admission-owner.test.ts | 112 +++++ .../__tests__/session-admission-gate.test.ts | 50 +++ packages/runtime-host/src/protocol/codec.ts | 59 +++ .../runtime-host/src/protocol/host-status.ts | 59 +++ packages/runtime-host/src/protocol/index.ts | 72 +-- .../src/protocol/operation-spec.ts | 57 +++ .../runtime-host/src/protocol/operations.ts | 412 +++--------------- packages/runtime-host/src/protocol/turn.ts | 203 +++++++++ .../src/server/execution-composition.ts | 6 + .../runtime-host/src/server/host-kernel.ts | 31 +- .../src/server/operation-dispatcher.ts | 29 +- .../src/server/root-admission-owner.ts | 92 ++++ .../src/server/root-turn-coordinator.ts | 41 +- .../src/server/session-admission-gate.ts | 20 + .../src/__tests__/execution-stores.test.ts | 113 ++++- packages/storage/src/agent-run-store.ts | 70 ++- packages/storage/src/execution-stores.ts | 3 + 21 files changed, 1312 insertions(+), 458 deletions(-) create mode 100644 packages/runtime-host/src/__tests__/operation-dispatcher.test.ts create mode 100644 packages/runtime-host/src/__tests__/root-admission-owner.test.ts create mode 100644 packages/runtime-host/src/__tests__/session-admission-gate.test.ts create mode 100644 packages/runtime-host/src/protocol/codec.ts create mode 100644 packages/runtime-host/src/protocol/host-status.ts create mode 100644 packages/runtime-host/src/protocol/operation-spec.ts create mode 100644 packages/runtime-host/src/protocol/turn.ts create mode 100644 packages/runtime-host/src/server/root-admission-owner.ts create mode 100644 packages/runtime-host/src/server/session-admission-gate.ts diff --git a/packages/runtime-host/src/__tests__/execution-host.test.ts b/packages/runtime-host/src/__tests__/execution-host.test.ts index ba172668ad..34cfad5a4b 100644 --- a/packages/runtime-host/src/__tests__/execution-host.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host.test.ts @@ -131,6 +131,54 @@ test('two Clients share one execution after the starting Client disconnects', as }); }); +test('concurrent root admission for one Session has a single winner', async () => { + await withExecutionRoot(async (fixture) => { + const host = await fixture.startHost(); + const first = await connectClient(fixture.root, 'desktop'); + const second = await connectClient(fixture.root, 'tui'); + const turnIds = [randomUUID(), randomUUID()] as const; + + const outcomes = await Promise.allSettled([ + first.startTurn({ + sessionId: fixture.sessionId, + turnId: turnIds[0], + text: FAKE_ASK_USER_QUESTION_PROMPT, + }), + second.startTurn({ + sessionId: fixture.sessionId, + turnId: turnIds[1], + text: FAKE_ASK_USER_QUESTION_PROMPT, + }), + ]); + const winners = outcomes.filter( + (outcome): outcome is PromiseFulfilledResult => outcome.status === 'fulfilled', + ); + const rejected = outcomes.filter( + (outcome): outcome is PromiseRejectedResult => outcome.status === 'rejected', + ); + assert.equal(winners.length, 1); + assert.equal(rejected.length, 1); + assert.ok(rejected[0]?.reason instanceof RuntimeHostOperationError); + assert.equal(rejected[0]?.reason.code, 'session_busy'); + + const winner = winners[0]?.value; + assert.ok(winner); + await first.stopTurn({ + sessionId: fixture.sessionId, + turnId: winner.turnId, + runId: winner.runId, + }); + await first.close(); + await second.close(); + await fixture.stopHost(host); + + const chain = await fixture.readAdmissionChain(); + assert.equal(chain.length, 1); + assert.equal(chain[0]?.turnId, winner.turnId); + assert.equal(chain[0]?.previousRootTurnId, null); + }); +}); + test('an archived Session rejects a new Turn before durable admission', async () => { await withExecutionRoot(async (fixture) => { await fixture.archiveSession(); @@ -472,12 +520,34 @@ test('retry after a discarded turn.start response reuses the durable semantic ad const terminal = await waitForTerminalTurn(observer, fixture.sessionId, turnId); assert.equal(terminal.status, 'completed'); await observer.close(); - await fixture.stopHost(host); + + await fixture.killHost(host); + const successorHost = await fixture.startHost(); + const successorClient = await connectClient(fixture.root, 'run'); + assert.deepEqual( + await successorClient.startTurn({ sessionId: fixture.sessionId, turnId, text }), + terminal, + ); + const successorTurnId = randomUUID(); + await successorClient.startTurn({ + sessionId: fixture.sessionId, + turnId: successorTurnId, + text: 'successor must extend the recovered durable tip', + }); + await waitForTerminalTurn(successorClient, fixture.sessionId, successorTurnId); + await successorClient.close(); + await fixture.stopHost(successorHost); const ledger = await fixture.readTurn(turnId); assert.equal(ledger.runs.length, 1); assert.equal(ledger.userMessages.length, 1); assert.equal(ledger.terminalEvents.length, 1); + const chain = await fixture.readAdmissionChain(); + assert.deepEqual( + chain.map((admission) => admission.turnId), + [turnId, successorTurnId], + ); + assert.equal(chain[1]?.previousRootTurnId, turnId); }); }); @@ -555,6 +625,7 @@ class ExecutionFixture { turnId, proposedRunId: randomUUID(), proposedUserMessageId: randomUUID(), + previousRootTurnId: null, normalizedInput: { text }, admittedAt, }); @@ -665,6 +736,18 @@ class ExecutionFixture { } } + async readAdmissionChain() { + const owner = await tryAcquireInteractiveRootOwner(this.capability); + assert.ok(owner); + if (!owner) throw new Error('Unable to acquire execution root for admission inspection'); + try { + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + return stores.agentRunStore.listRootTurnAdmissionsForRecovery(this.sessionId); + } finally { + await owner.close(); + } + } + async readTurnFootprint(turnId: string): Promise<{ admitted: boolean; runCount: number; diff --git a/packages/runtime-host/src/__tests__/host-kernel.test.ts b/packages/runtime-host/src/__tests__/host-kernel.test.ts index 6ea10a2248..062cab613d 100644 --- a/packages/runtime-host/src/__tests__/host-kernel.test.ts +++ b/packages/runtime-host/src/__tests__/host-kernel.test.ts @@ -102,6 +102,88 @@ describe('non-serving Runtime Host kernel', () => { }); }); + test('serves bootstrap operations during recovery and rejects ready-only operations', async () => { + await withHostPaths(async (paths) => { + const capability = await resolveStorageRoot({ path: paths.root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + let releaseFactory = () => {}; + let markFactoryEntered!: () => void; + const factoryEntered = new Promise((resolve) => { + markFactoryEntered = resolve; + }); + const factoryReleased = new Promise((resolve) => { + releaseFactory = resolve; + }); + const unavailable = async () => + ({ + ok: false, + error: { + code: 'operation_unavailable', + message: 'not available in this composition', + }, + }) as const; + const hostTask = RuntimeHostKernel.start({ + owner, + idleGraceMs: 10_000, + compositionFactory: async () => { + markFactoryEntered(); + await factoryReleased; + return { + handlers: { + 'turn.start': unavailable, + 'turn.query': unavailable, + 'turn.stop': unavailable, + }, + async recover() {}, + async close() {}, + }; + }, + }); + let host: RuntimeHostKernel | undefined; + let transport: FramedTransport | undefined; + try { + await withTimeout(factoryEntered, 1_000, 'Runtime Host did not enter composition'); + const registration = await readHostRegistration(owner.controlDirectory); + assert.ok(registration); + assert.equal(registration.state, 'recovering'); + transport = new FramedTransport(await openSocket(registration.endpoint)); + await transport.write({ + kind: 'hello', + clientInstanceId: 'lifecycle-test', + surface: 'inspect', + protocolMin: CURRENT_PROTOCOL.min, + protocolMax: CURRENT_PROTOCOL.max, + }); + 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: {} }); + 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({ + requestId: 'query', + operation: 'turn.query', + input: { sessionId: 'session', turnId: 'turn' }, + }); + const query = decodeHostFrame(await transport.read(1_000)); + assert.ok(!('kind' in query) && query.operation === 'turn.query' && !query.ok); + if (!('kind' in query) && query.operation === 'turn.query' && !query.ok) { + assert.equal(query.error.code, 'host_not_ready'); + } + } finally { + releaseFactory(); + transport?.destroy(); + host = await hostTask.catch(() => undefined); + await host?.close().catch(() => undefined); + } + }); + }); + test('blocks incompatible replacement while resident and permits it only after true idle', async () => { await withHostPaths(async (paths) => { const candidate = await startTestRuntimeHostCandidate(paths, { diff --git a/packages/runtime-host/src/__tests__/operation-dispatcher.test.ts b/packages/runtime-host/src/__tests__/operation-dispatcher.test.ts new file mode 100644 index 0000000000..95e0f3f820 --- /dev/null +++ b/packages/runtime-host/src/__tests__/operation-dispatcher.test.ts @@ -0,0 +1,150 @@ +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import type { OperationKey, OperationOutcome, RequestFrame } from '../protocol/index.js'; +import { + composeOperationHandlers, + dispatchOperation, + type ConnectionContext, + type OperationHandlerMap, +} from '../server/operation-dispatcher.js'; + +const context: ConnectionContext = { + hostEpoch: 'epoch-1', + connectionId: 'connection-1', + surface: 'tui', + principal: 'local_os_user', + acquireResidency: () => ({ release() {} }), +}; + +const request = { + requestId: 'request-1', + operation: 'turn.query', + input: { sessionId: 'session-1', turnId: 'turn-1' }, +} satisfies RequestFrame; + +describe('Runtime Host operation dispatcher', () => { + test('rejects malformed handler composition', () => { + const handlers = validHandlers(); + assert.throws( + () => + composeOperationHandlers({ + ...handlers, + 'unknown.operation': handlers['host.status'], + } as unknown as Partial), + /Unknown Runtime Host operation handler: unknown\.operation/, + ); + assert.throws( + () => composeOperationHandlers(handlers, { 'host.status': handlers['host.status'] }), + /Duplicate Runtime Host operation handler: host\.status/, + ); + assert.throws( + () => composeOperationHandlers({ 'host.status': handlers['host.status'] }), + /Missing Runtime Host operation handlers:/, + ); + assert.throws( + () => + composeOperationHandlers({ + ...handlers, + 'turn.query': undefined, + } as unknown as Partial), + /Invalid Runtime Host operation handler: turn\.query/, + ); + }); + + test('converts handler throws and malformed outcomes to declared internal_failure', async () => { + const malformedOutcomes: unknown[] = [ + { ok: true, result: { sessionId: 'session-1', turnId: 'turn-1' } }, + { + ok: true, + result: runningSnapshot(), + privateState: true, + }, + { ok: false, error: { code: 'session_busy', message: 'not declared for query' } }, + { ok: false, error: { code: 'not_found', message: 'missing', details: {} } }, + { ok: false, error: 'missing' }, + ]; + + for (const outcome of malformedOutcomes) { + const response = await dispatchOperation( + request, + handlersWithQuery(async () => outcome as OperationOutcome<'turn.query'>), + context, + ); + assert.deepEqual(response, internalFailure()); + } + + const thrown = await dispatchOperation( + request, + handlersWithQuery(async () => { + throw new Error('private failure'); + }), + context, + ); + assert.deepEqual(thrown, internalFailure()); + }); + + test('passes only decoded valid success and declared exact failure outcomes', async () => { + const success = await dispatchOperation( + request, + handlersWithQuery(async () => ({ ok: true, result: runningSnapshot() })), + context, + ); + assert.deepEqual(success, { ...requestIdentity(), ok: true, result: runningSnapshot() }); + + const failure = await dispatchOperation( + request, + handlersWithQuery(async () => ({ + ok: false, + error: { code: 'not_found', message: 'Turn does not exist' }, + })), + context, + ); + assert.deepEqual(failure, { + ...requestIdentity(), + ok: false, + error: { code: 'not_found', message: 'Turn does not exist' }, + }); + }); +}); + +function validHandlers(): OperationHandlerMap { + const unavailable = async (): Promise> => + ({ + ok: false, + error: { + code: 'internal_failure', + message: 'not used', + }, + }) as OperationOutcome; + return { + 'host.status': unavailable, + 'turn.start': unavailable, + 'turn.query': unavailable, + 'turn.stop': unavailable, + }; +} + +function handlersWithQuery(query: OperationHandlerMap['turn.query']): OperationHandlerMap { + return { ...validHandlers(), 'turn.query': query }; +} + +function runningSnapshot() { + return { + sessionId: 'session-1', + turnId: 'turn-1', + runId: 'run-1', + status: 'running' as const, + }; +} + +function requestIdentity() { + return { requestId: request.requestId, operation: request.operation }; +} + +function internalFailure() { + return { + ...requestIdentity(), + ok: false, + error: { code: 'internal_failure', message: 'Runtime Host operation failed' }, + }; +} diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index fc8a60509d..88dafc99fd 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -8,6 +8,8 @@ import { RUNTIME_HOST_MAX_FRAME_BYTES, RuntimeHostProtocolError, } from '../protocol/index.js'; +import { HOST_STATUS_OPERATION_SPECS } from '../protocol/host-status.js'; +import { composeOperationSpecMaps } from '../protocol/operation-spec.js'; describe('Runtime Host bootstrap protocol', () => { test('selects the highest mutually supported protocol and rejects a gap', () => { @@ -64,6 +66,28 @@ describe('Runtime Host bootstrap protocol', () => { }), isInvalidFrame, ); + assert.throws( + () => + decodeHostFrame({ + requestId: 'request-unknown-field', + operation: 'host.status', + ok: false, + error: { code: 'host_draining', message: 'draining' }, + trace: 'private', + }), + isInvalidFrame, + ); + }); + + test('rejects duplicate operation keys while composing domain registries', () => { + const composeUnchecked = composeOperationSpecMaps as ( + left: typeof HOST_STATUS_OPERATION_SPECS, + right: typeof HOST_STATUS_OPERATION_SPECS, + ) => unknown; + assert.throws( + () => composeUnchecked(HOST_STATUS_OPERATION_SPECS, HOST_STATUS_OPERATION_SPECS), + /Duplicate Runtime Host operation key: host\.status/, + ); }); test('rejects terminal snapshots with fields from another terminal variant', () => { diff --git a/packages/runtime-host/src/__tests__/root-admission-owner.test.ts b/packages/runtime-host/src/__tests__/root-admission-owner.test.ts new file mode 100644 index 0000000000..0b42c27327 --- /dev/null +++ b/packages/runtime-host/src/__tests__/root-admission-owner.test.ts @@ -0,0 +1,112 @@ +import assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { createAgentRunStore, type RootTurnAdmissionStore } from '@maka/storage'; +import { RootAdmissionOwner } from '../server/root-admission-owner.js'; +import { SessionAdmissionGate } from '../server/session-admission-gate.js'; + +test('poisons a Session after an ambiguous durable admission failure', async () => { + await withStore(async (durableStore) => { + let failAfterCommit = true; + const store: RootTurnAdmissionStore = { + admitRootTurn: async (input) => { + const result = await durableStore.admitRootTurn(input); + if (failAfterCommit) { + failAfterCommit = false; + throw new Error('post-commit durability failure'); + } + return result; + }, + readRootTurnAdmission: (sessionId, turnId) => + durableStore.readRootTurnAdmission(sessionId, turnId), + listRootTurnAdmissionsForRecovery: (sessionId) => + durableStore.listRootTurnAdmissionsForRecovery(sessionId), + }; + const owner = new RootAdmissionOwner(store); + await owner.recoverSession('session'); + const gate = new SessionAdmissionGate(); + + const outcomes = await Promise.allSettled([ + gate.run('session', () => owner.admitRootTurn(admitInput('session', 'turn-1', 10))), + gate.run('session', () => owner.admitRootTurn(admitInput('session', 'turn-2', 20))), + ]); + assert.equal(outcomes[0]?.status, 'rejected'); + assert.equal(outcomes[1]?.status, 'rejected'); + if (outcomes[0]?.status === 'rejected') { + assert.match(String(outcomes[0].reason), /post-commit durability failure/); + } + if (outcomes[1]?.status === 'rejected') { + assert.match(String(outcomes[1].reason), /admission state is uncertain/); + } + + const chain = await durableStore.listRootTurnAdmissionsForRecovery('session'); + assert.deepEqual( + chain.map((admission) => admission.turnId), + ['turn-1'], + ); + }); +}); + +test('recovery installs the validated tip and the successor extends it', async () => { + await withStore(async (store) => { + await store.admitRootTurn({ + ...admitInput('session', 'turn-1', 100), + previousRootTurnId: null, + }); + await store.admitRootTurn({ + ...admitInput('session', 'turn-2', 100), + previousRootTurnId: 'turn-1', + }); + const owner = new RootAdmissionOwner(store); + const chain = await owner.recoverSession('session'); + assert.deepEqual( + chain.map((admission) => admission.turnId), + ['turn-1', 'turn-2'], + ); + + const successor = await owner.admitRootTurn(admitInput('session', 'turn-3', 100)); + assert.equal(successor.admission.previousRootTurnId, 'turn-2'); + assert.doesNotThrow(() => owner.assertKnownAdmission(successor.admission)); + }); +}); + +test('fails closed when a known durable admission identity drifts', async () => { + await withStore(async (store) => { + const first = await store.admitRootTurn({ + ...admitInput('session', 'turn-1', 10), + previousRootTurnId: null, + }); + const owner = new RootAdmissionOwner(store); + await owner.recoverSession('session'); + owner.assertKnownAdmission(first.admission); + assert.throws( + () => owner.assertKnownAdmission({ ...first.admission, runId: 'run-drifted' }), + /identity changed/, + ); + await assert.rejects(() => owner.recoverSession('session'), /already installed/); + }); +}); + +function admitInput(sessionId: string, turnId: string, admittedAt: number) { + return { + sessionId, + turnId, + proposedRunId: `run-${turnId}`, + proposedUserMessageId: `message-${turnId}`, + normalizedInput: { text: `text-${turnId}` }, + admittedAt, + }; +} + +async function withStore( + run: (store: ReturnType) => Promise, +): Promise { + const root = await mkdtemp(join(tmpdir(), 'maka-root-admission-owner-')); + try { + await run(createAgentRunStore(root)); + } finally { + await rm(root, { recursive: true, force: true }); + } +} diff --git a/packages/runtime-host/src/__tests__/session-admission-gate.test.ts b/packages/runtime-host/src/__tests__/session-admission-gate.test.ts new file mode 100644 index 0000000000..bd8d9efd69 --- /dev/null +++ b/packages/runtime-host/src/__tests__/session-admission-gate.test.ts @@ -0,0 +1,50 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { SessionAdmissionGate } from '../server/session-admission-gate.js'; + +test('serializes operations for one Session', async () => { + const gate = new SessionAdmissionGate(); + const entered = deferred(); + const release = deferred(); + const order: string[] = []; + + const first = gate.run('session', async () => { + order.push('first:start'); + entered.resolve(); + await release.promise; + order.push('first:end'); + }); + await entered.promise; + const second = gate.run('session', () => { + order.push('second'); + }); + + await Promise.resolve(); + assert.deepEqual(order, ['first:start']); + release.resolve(); + await Promise.all([first, second]); + assert.deepEqual(order, ['first:start', 'first:end', 'second']); +}); + +test('does not serialize operations for different Sessions', async () => { + const gate = new SessionAdmissionGate(); + const entered = deferred(); + const release = deferred(); + const first = gate.run('first', async () => { + entered.resolve(); + await release.promise; + }); + await entered.promise; + + assert.equal(await gate.run('second', () => 'completed'), 'completed'); + release.resolve(); + await first; +}); + +function deferred(): { promise: Promise; resolve(): void } { + let resolve!: () => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} diff --git a/packages/runtime-host/src/protocol/codec.ts b/packages/runtime-host/src/protocol/codec.ts new file mode 100644 index 0000000000..203e5a24d7 --- /dev/null +++ b/packages/runtime-host/src/protocol/codec.ts @@ -0,0 +1,59 @@ +import { invalidProtocolFrame } from './errors.js'; + +export function requireRecord(value: unknown, label: string): Record { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw invalidProtocolFrame(`Invalid ${label}`); + } + return value as Record; +} + +export function requireExactRecord( + value: unknown, + label: string, + keys: readonly string[], +): Record { + const record = requireRecord(value, label); + assertExactKeys(record, label, keys); + return record; +} + +export function assertExactKeys( + record: Record, + label: string, + keys: readonly string[], +): void { + const allowed = new Set(keys); + if (Object.keys(record).some((key) => !allowed.has(key))) { + throw invalidProtocolFrame(`Unknown ${label} field`); + } + if ( + Object.keys(record).length !== keys.length || + keys.some((key) => !Object.hasOwn(record, key)) + ) { + throw invalidProtocolFrame(`Invalid ${label} fields`); + } +} + +export function requireString(value: unknown, label: string, maxLength: number): string { + if (typeof value !== 'string' || value.length === 0 || value.length > maxLength) { + throw invalidProtocolFrame(`Invalid ${label}`); + } + return value; +} + +export function requireId(value: unknown, label: string): string { + return requireString(value, label, 128); +} + +export function requireEntityId(value: unknown, label: string): string { + const id = requireId(value, label); + if (!/^[A-Za-z0-9_-]{1,128}$/.test(id)) throw invalidProtocolFrame(`Invalid ${label}`); + return id; +} + +export function requireCount(value: unknown, label: string): number { + if (!Number.isSafeInteger(value) || (value as number) < 0) { + throw invalidProtocolFrame(`Invalid ${label}`); + } + return value as number; +} diff --git a/packages/runtime-host/src/protocol/host-status.ts b/packages/runtime-host/src/protocol/host-status.ts new file mode 100644 index 0000000000..51c9b3fb97 --- /dev/null +++ b/packages/runtime-host/src/protocol/host-status.ts @@ -0,0 +1,59 @@ +import { invalidProtocolFrame } from './errors.js'; +import { requireCount, requireExactRecord, requireId } from './codec.js'; +import { defineOperation } from './operation-spec.js'; + +export type HostLifecycleState = 'starting' | 'containing' | 'recovering' | 'ready' | 'draining'; +export type HostStatusInput = Record; + +export interface HostStatusResult { + hostEpoch: string; + state: HostLifecycleState; + connections: number; + activeOperations: number; + activeResidencies: number; +} + +export const HOST_STATUS_OPERATION_SPECS = { + 'host.status': defineOperation({ + mode: 'query', + availability: 'bootstrap', + errors: ['host_draining', 'internal_failure'] as const, + decodeInput: decodeHostStatusInput, + decodeOutput: decodeHostStatusResult, + }), +} as const; + +function decodeHostStatusInput(value: unknown): HostStatusInput { + requireExactRecord(value, 'host.status input', []); + return {}; +} + +function decodeHostStatusResult(value: unknown): HostStatusResult { + const record = requireExactRecord(value, 'host.status result', [ + 'hostEpoch', + 'state', + 'connections', + 'activeOperations', + 'activeResidencies', + ]); + return { + hostEpoch: requireId(record.hostEpoch, 'hostEpoch'), + state: requireHostLifecycleState(record.state), + connections: requireCount(record.connections, 'connections'), + activeOperations: requireCount(record.activeOperations, 'activeOperations'), + activeResidencies: requireCount(record.activeResidencies, 'activeResidencies'), + }; +} + +export function requireHostLifecycleState(value: unknown): HostLifecycleState { + if ( + value === 'starting' || + value === 'containing' || + value === 'recovering' || + value === 'ready' || + value === 'draining' + ) { + return value; + } + throw invalidProtocolFrame('Invalid Host state'); +} diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 171d71fc5f..cb263774e2 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -1,5 +1,7 @@ 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'; import { decodeRequestFrame, decodeResponseFrame, @@ -84,7 +86,7 @@ export function validateProtocolRange(range: ProtocolRange): void { range.min < 1 || range.max < range.min ) { - throw invalidFrame('Invalid protocol range'); + throw invalidProtocolFrame('Invalid protocol range'); } } @@ -129,7 +131,7 @@ export function decodeHostFrame(value: unknown): HostFrame { hostEpoch: requireId(frame.hostEpoch, 'hostEpoch'), protocolMin, protocolMax, - state: requireHostState(frame.state), + state: requireHostLifecycleState(frame.state), replacement: requireReplacement(frame.replacement), } satisfies HostIncompatible; } @@ -141,17 +143,19 @@ export function decodeHostFrame(value: unknown): HostFrame { export function decodeHostRegistration(value: unknown): HostRegistration { const registration = requireRecord(value, 'host registration'); - if (registration.kind !== 'maka-runtime-host') throw invalidFrame('Invalid registration kind'); + if (registration.kind !== 'maka-runtime-host') { + throw invalidProtocolFrame('Invalid registration kind'); + } if (registration.schemaVersion !== RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION) { - throw invalidFrame('Unsupported registration schema'); + throw invalidProtocolFrame('Unsupported registration schema'); } const protocolMin = requireProtocolVersion(registration.protocolMin, 'protocolMin'); const protocolMax = requireProtocolVersion(registration.protocolMax, 'protocolMax'); validateProtocolRange({ min: protocolMin, max: protocolMax }); const rootId = requireString(registration.rootId, 'rootId', 128); - if (!/^[a-f0-9]{64}$/.test(rootId)) throw invalidFrame('Invalid rootId'); + if (!/^[a-f0-9]{64}$/.test(rootId)) throw invalidProtocolFrame('Invalid rootId'); const pid = requireCount(registration.pid, 'pid'); - if (pid === 0) throw invalidFrame('Invalid pid'); + if (pid === 0) throw invalidProtocolFrame('Invalid pid'); return { kind: 'maka-runtime-host', schemaVersion: RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION, @@ -160,7 +164,7 @@ export function decodeHostRegistration(value: unknown): HostRegistration { endpoint: requireString(registration.endpoint, 'endpoint', 512), protocolMin, protocolMax, - state: requireHostState(registration.state), + state: requireHostLifecycleState(registration.state), pid, createdAt: requireString(registration.createdAt, 'createdAt', 64), }; @@ -217,7 +221,9 @@ export class ProtocolFrameDecoder { } #decodePending(): unknown { - if (this.#pending.byteLength === 0) throw invalidFrame('Runtime Host frame is empty'); + 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; @@ -233,30 +239,10 @@ export class ProtocolFrameDecoder { } } -function requireRecord(value: unknown, label: string): Record { - if (!value || typeof value !== 'object' || Array.isArray(value)) - throw invalidFrame(`Invalid ${label}`); - return value as Record; -} - -function requireString(value: unknown, label: string, maxLength: number): string { - if (typeof value !== 'string' || value.length === 0 || value.length > maxLength) { - throw invalidFrame(`Invalid ${label}`); - } - return value; -} - -function requireId(value: unknown, label: string): string { - return requireString(value, label, 128); -} - function requireProtocolVersion(value: unknown, label: string): number { - if (!Number.isSafeInteger(value) || (value as number) < 1) throw invalidFrame(`Invalid ${label}`); - return value as number; -} - -function requireCount(value: unknown, label: string): number { - if (!Number.isSafeInteger(value) || (value as number) < 0) throw invalidFrame(`Invalid ${label}`); + if (!Number.isSafeInteger(value) || (value as number) < 1) { + throw invalidProtocolFrame(`Invalid ${label}`); + } return value as number; } @@ -270,32 +256,16 @@ function requireSurface(value: unknown): ClientSurface { value === 'inspect' ) return value; - throw invalidFrame('Invalid surface'); -} - -function requireHostState(value: unknown): HostLifecycleState { - if ( - value === 'starting' || - value === 'containing' || - value === 'recovering' || - value === 'ready' || - value === 'draining' - ) - return value; - throw invalidFrame('Invalid Host state'); + throw invalidProtocolFrame('Invalid surface'); } function requireAcceptedState(value: unknown): Exclude { - const state = requireHostState(value); - if (state === 'draining') throw invalidFrame('Accepted Host cannot be draining'); + const state = requireHostLifecycleState(value); + if (state === 'draining') throw invalidProtocolFrame('Accepted Host cannot be draining'); return state; } function requireReplacement(value: unknown): HostIncompatible['replacement'] { if (value === 'blocked_by_residency' || value === 'wait_for_idle_exit') return value; - throw invalidFrame('Invalid replacement disposition'); -} - -function invalidFrame(message: string): RuntimeHostProtocolError { - return invalidProtocolFrame(message); + throw invalidProtocolFrame('Invalid replacement disposition'); } diff --git a/packages/runtime-host/src/protocol/operation-spec.ts b/packages/runtime-host/src/protocol/operation-spec.ts new file mode 100644 index 0000000000..8e7db454ae --- /dev/null +++ b/packages/runtime-host/src/protocol/operation-spec.ts @@ -0,0 +1,57 @@ +export type OperationMode = 'command' | 'query' | 'control'; +export type OperationAvailability = 'bootstrap' | 'ready'; + +export type HostOperationErrorCode = + | 'host_not_ready' + | 'host_draining' + | 'operation_unavailable' + | 'not_found' + | 'session_archived' + | 'session_busy' + | 'operation_conflict' + | 'internal_failure'; + +export interface HostOperationError { + code: C; + message: string; +} + +export interface OperationSpec { + mode: OperationMode; + availability: OperationAvailability; + errors: readonly ErrorCode[]; + decodeInput(value: unknown): Input; + decodeOutput(value: unknown): Output; +} + +type AnyOperationSpec = OperationSpec; +export type OperationSpecMap = Readonly>; +type DuplicateOperationKeys = Extract; +type RequireDisjointOperationKeys = [DuplicateOperationKeys] extends [ + never, +] + ? unknown + : { readonly duplicateOperationKeys: DuplicateOperationKeys }; + +export function defineOperation( + spec: OperationSpec, +): OperationSpec { + if (!(spec.errors as readonly HostOperationErrorCode[]).includes('internal_failure')) { + throw new Error('Every Runtime Host operation must declare internal_failure'); + } + return spec; +} + +export function composeOperationSpecMaps< + const Left extends OperationSpecMap, + const Right extends OperationSpecMap, +>(left: Left, right: Right & RequireDisjointOperationKeys): Left & Right { + const combined: Record = { ...left }; + for (const [key, spec] of Object.entries(right)) { + if (Object.hasOwn(combined, key)) { + throw new Error(`Duplicate Runtime Host operation key: ${key}`); + } + combined[key] = spec; + } + return combined as Left & Right; +} diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 6682a50d26..1531503484 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -1,154 +1,28 @@ +import { requireExactRecord, requireId, requireRecord, requireString } from './codec.js'; import { invalidProtocolFrame } from './errors.js'; - -export type HostLifecycleState = 'starting' | 'containing' | 'recovering' | 'ready' | 'draining'; -export type OperationMode = 'command' | 'query' | 'control'; -export type RetryPolicy = 'none' | 'safe' | 'semantic'; -export type AdmissionClass = 'bootstrap' | 'ready' | 'session'; - -export type HostOperationErrorCode = - | 'host_not_ready' - | 'host_draining' - | 'operation_unavailable' - | 'not_found' - | 'session_archived' - | 'session_busy' - | 'operation_conflict' - | 'internal_failure'; - -export interface HostOperationError { - code: C; - message: string; -} - -export interface OperationSpec { - mode: OperationMode; - decodeInput(value: unknown): Input; - decodeOutput(value: unknown): Output; - errors: readonly ErrorCode[]; - retry: RetryPolicy; - admission: AdmissionClass; -} - -export type HostStatusInput = Record; - -export interface HostStatusResult { - hostEpoch: string; - state: HostLifecycleState; - connections: number; - activeOperations: number; - activeResidencies: number; -} - -export interface TurnStartInput { - sessionId: string; - turnId: string; - text: string; -} - -export interface TurnQueryInput { - sessionId: string; - turnId: string; -} - -export interface TurnStopInput { - sessionId: string; - turnId: string; - runId: string; -} - -export type TurnRunStatus = - | 'admitted' - | 'created' - | 'running' - | 'waiting_permission' - | 'completed' - | 'failed' - | 'cancelled'; - -interface TurnSnapshotBase { - sessionId: string; - turnId: string; - runId: string; -} - -export type TurnSnapshot = - | (TurnSnapshotBase & { - status: Exclude; - }) - | (TurnSnapshotBase & { status: 'completed'; terminalEventId: string }) - | (TurnSnapshotBase & { - status: 'failed'; - terminalEventId: string; - failureClass: string; - }) - | (TurnSnapshotBase & { - status: 'cancelled'; - terminalEventId: string; - abortSource: string; - }); - -function defineOperation( - spec: OperationSpec, -): OperationSpec { - return spec; -} - -export const HOST_OPERATION_SPECS = { - 'host.status': defineOperation({ - mode: 'query', - decodeInput: decodeHostStatusInput, - decodeOutput: decodeHostStatusResult, - errors: ['host_draining', 'internal_failure'] as const, - retry: 'safe', - admission: 'bootstrap', - }), - 'turn.start': defineOperation({ - mode: 'command', - decodeInput: decodeTurnStartInput, - decodeOutput: decodeTurnSnapshot, - errors: [ - 'host_not_ready', - 'host_draining', - 'operation_unavailable', - 'not_found', - 'session_archived', - 'session_busy', - 'operation_conflict', - 'internal_failure', - ] as const, - retry: 'semantic', - admission: 'session', - }), - 'turn.query': defineOperation({ - mode: 'query', - decodeInput: decodeTurnQueryInput, - decodeOutput: decodeTurnSnapshot, - errors: [ - 'host_not_ready', - 'host_draining', - 'operation_unavailable', - 'not_found', - 'internal_failure', - ] as const, - retry: 'safe', - admission: 'ready', - }), - 'turn.stop': defineOperation({ - mode: 'control', - decodeInput: decodeTurnStopInput, - decodeOutput: decodeTurnSnapshot, - errors: [ - 'host_not_ready', - 'host_draining', - 'operation_unavailable', - 'not_found', - 'operation_conflict', - 'internal_failure', - ] as const, - retry: 'semantic', - admission: 'session', - }), -} as const; +import { HOST_STATUS_OPERATION_SPECS } from './host-status.js'; +import { + composeOperationSpecMaps, + type HostOperationError, + type HostOperationErrorCode, + type OperationSpec, +} from './operation-spec.js'; +import { TURN_OPERATION_SPECS } from './turn.js'; + +export type { HostLifecycleState, HostStatusInput, HostStatusResult } from './host-status.js'; +export type { HostOperationError, HostOperationErrorCode } from './operation-spec.js'; +export type { + TurnQueryInput, + TurnRunStatus, + TurnSnapshot, + TurnStartInput, + TurnStopInput, +} from './turn.js'; + +export const HOST_OPERATION_SPECS = composeOperationSpecMaps( + HOST_STATUS_OPERATION_SPECS, + TURN_OPERATION_SPECS, +); export type OperationSpecMap = typeof HOST_OPERATION_SPECS; export type OperationKey = keyof OperationSpecMap; @@ -191,138 +65,54 @@ export function decodeRequestFrame(value: unknown): RequestFrame { const frame = requireExactRecord(value, 'operation request', ['requestId', 'operation', 'input']); const requestId = requireId(frame.requestId, 'requestId'); const operation = requireOperationKey(frame.operation); - const spec = HOST_OPERATION_SPECS[operation]; - const input = spec.decodeInput(frame.input); + const input = HOST_OPERATION_SPECS[operation].decodeInput(frame.input); return { requestId, operation, input } as RequestFrame; } export function decodeResponseFrame(value: unknown): ResponseFrame { const record = requireRecord(value, 'operation response'); + const requestId = requireId(record.requestId, 'requestId'); + const operation = requireOperationKey(record.operation); + const outcome = decodeOperationOutcome(operation, omitResponseIdentity(record)); + return { requestId, operation, ...outcome } as ResponseFrame; +} + +export function decodeOperationOutcome( + operation: K, + value: unknown, +): OperationOutcome { + const record = requireRecord(value, 'operation outcome'); if (record.ok === true) { - assertExactKeys(record, 'operation response', ['requestId', 'operation', 'ok', 'result']); - const requestId = requireId(record.requestId, 'requestId'); - const operation = requireOperationKey(record.operation); - const result = HOST_OPERATION_SPECS[operation].decodeOutput(record.result); - return { requestId, operation, ok: true, result } as ResponseFrame; + const exact = requireExactRecord(record, 'operation outcome', ['ok', 'result']); + return { + ok: true, + result: HOST_OPERATION_SPECS[operation].decodeOutput(exact.result), + } as OperationOutcome; } if (record.ok === false) { - assertExactKeys(record, 'operation response', ['requestId', 'operation', 'ok', 'error']); - const requestId = requireId(record.requestId, 'requestId'); - const operation = requireOperationKey(record.operation); - const error = decodeOperationError(record.error, HOST_OPERATION_SPECS[operation].errors); - return { requestId, operation, ok: false, error } as ResponseFrame; + const exact = requireExactRecord(record, 'operation outcome', ['ok', 'error']); + return { + ok: false, + error: decodeOperationError(exact.error, HOST_OPERATION_SPECS[operation].errors), + } as OperationOutcome; } - throw invalidProtocolFrame('Invalid operation response outcome'); + throw invalidProtocolFrame('Invalid operation outcome'); } export function isOperationKey(value: unknown): value is OperationKey { return typeof value === 'string' && Object.hasOwn(HOST_OPERATION_SPECS, value); } -function decodeHostStatusInput(value: unknown): HostStatusInput { - requireExactRecord(value, 'host.status input', []); - return {}; -} - -function decodeHostStatusResult(value: unknown): HostStatusResult { - const record = requireExactRecord(value, 'host.status result', [ - 'hostEpoch', - 'state', - 'connections', - 'activeOperations', - 'activeResidencies', - ]); - return { - hostEpoch: requireId(record.hostEpoch, 'hostEpoch'), - state: requireHostState(record.state), - connections: requireCount(record.connections, 'connections'), - activeOperations: requireCount(record.activeOperations, 'activeOperations'), - activeResidencies: requireCount(record.activeResidencies, 'activeResidencies'), - }; -} - -function decodeTurnStartInput(value: unknown): TurnStartInput { - const record = requireExactRecord(value, 'turn.start input', ['sessionId', 'turnId', 'text']); - return { - sessionId: requireEntityId(record.sessionId, 'sessionId'), - turnId: requireEntityId(record.turnId, 'turnId'), - text: requireString(record.text, 'text', 48 * 1024), - }; -} - -function decodeTurnQueryInput(value: unknown): TurnQueryInput { - const record = requireExactRecord(value, 'turn.query input', ['sessionId', 'turnId']); - return { - sessionId: requireEntityId(record.sessionId, 'sessionId'), - turnId: requireEntityId(record.turnId, 'turnId'), - }; -} - -function decodeTurnStopInput(value: unknown): TurnStopInput { - const record = requireExactRecord(value, 'turn.stop input', ['sessionId', 'turnId', 'runId']); - return { - sessionId: requireEntityId(record.sessionId, 'sessionId'), - turnId: requireEntityId(record.turnId, 'turnId'), - runId: requireEntityId(record.runId, 'runId'), - }; -} - -function decodeTurnSnapshot(value: unknown): TurnSnapshot { - const record = requireRecord(value, 'Turn snapshot'); - const base = { - sessionId: requireEntityId(record.sessionId, 'sessionId'), - turnId: requireEntityId(record.turnId, 'turnId'), - runId: requireEntityId(record.runId, 'runId'), - }; - const status = requireTurnRunStatus(record.status); - if (status === 'completed') { - assertExactKeys(record, 'completed Turn snapshot', [ - 'sessionId', - 'turnId', - 'runId', - 'status', - 'terminalEventId', - ]); - return { - ...base, - status, - terminalEventId: requireId(record.terminalEventId, 'terminalEventId'), - }; - } - if (status === 'failed') { - assertExactKeys(record, 'failed Turn snapshot', [ - 'sessionId', - 'turnId', - 'runId', - 'status', - 'terminalEventId', - 'failureClass', - ]); - return { - ...base, - status, - terminalEventId: requireId(record.terminalEventId, 'terminalEventId'), - failureClass: requireString(record.failureClass, 'failureClass', 128), - }; +function omitResponseIdentity(record: Record): Record { + if (record.ok === true) { + requireExactRecord(record, 'operation response', ['requestId', 'operation', 'ok', 'result']); + return { ok: true, result: record.result }; } - if (status === 'cancelled') { - assertExactKeys(record, 'cancelled Turn snapshot', [ - 'sessionId', - 'turnId', - 'runId', - 'status', - 'terminalEventId', - 'abortSource', - ]); - return { - ...base, - status, - terminalEventId: requireId(record.terminalEventId, 'terminalEventId'), - abortSource: requireString(record.abortSource, 'abortSource', 128), - }; + if (record.ok === false) { + requireExactRecord(record, 'operation response', ['requestId', 'operation', 'ok', 'error']); + return { ok: false, error: record.error }; } - assertExactKeys(record, 'non-terminal Turn snapshot', ['sessionId', 'turnId', 'runId', 'status']); - return { ...base, status }; + throw invalidProtocolFrame('Invalid operation response outcome'); } function decodeOperationError( @@ -343,95 +133,3 @@ function requireOperationKey(value: unknown): OperationKey { if (!isOperationKey(value)) throw invalidProtocolFrame('Unknown operation key'); return value; } - -function requireHostState(value: unknown): HostLifecycleState { - if ( - value === 'starting' || - value === 'containing' || - value === 'recovering' || - value === 'ready' || - value === 'draining' - ) - return value; - throw invalidProtocolFrame('Invalid Host state'); -} - -function requireTurnRunStatus(value: unknown): TurnRunStatus { - if ( - value === 'admitted' || - value === 'created' || - value === 'running' || - value === 'waiting_permission' || - value === 'completed' || - value === 'failed' || - value === 'cancelled' - ) - return value; - throw invalidProtocolFrame('Invalid Turn run status'); -} - -function requireRecord(value: unknown, label: string): Record { - if (!value || typeof value !== 'object' || Array.isArray(value)) { - throw invalidProtocolFrame(`Invalid ${label}`); - } - return value as Record; -} - -function requireExactRecord( - value: unknown, - label: string, - keys: readonly string[], -): Record { - const record = requireRecord(value, label); - assertExactKeys(record, label, keys); - return record; -} - -function assertExactKeys( - record: Record, - label: string, - keys: readonly string[], -): void { - assertAllowedKeys(record, label, keys); - if ( - Object.keys(record).length !== keys.length || - keys.some((key) => !Object.hasOwn(record, key)) - ) { - throw invalidProtocolFrame(`Invalid ${label} fields`); - } -} - -function assertAllowedKeys( - record: Record, - label: string, - keys: readonly string[], -): void { - const allowed = new Set(keys); - if (Object.keys(record).some((key) => !allowed.has(key))) { - throw invalidProtocolFrame(`Unknown ${label} field`); - } -} - -function requireString(value: unknown, label: string, maxLength: number): string { - if (typeof value !== 'string' || value.length === 0 || value.length > maxLength) { - throw invalidProtocolFrame(`Invalid ${label}`); - } - return value; -} - -function requireId(value: unknown, label: string): string { - return requireString(value, label, 128); -} - -function requireEntityId(value: unknown, label: string): string { - const id = requireId(value, label); - if (!/^[A-Za-z0-9_-]{1,128}$/.test(id)) throw invalidProtocolFrame(`Invalid ${label}`); - return id; -} - -function requireCount(value: unknown, label: string): number { - if (!Number.isSafeInteger(value) || (value as number) < 0) { - throw invalidProtocolFrame(`Invalid ${label}`); - } - return value as number; -} diff --git a/packages/runtime-host/src/protocol/turn.ts b/packages/runtime-host/src/protocol/turn.ts new file mode 100644 index 0000000000..00b69fc2eb --- /dev/null +++ b/packages/runtime-host/src/protocol/turn.ts @@ -0,0 +1,203 @@ +import { invalidProtocolFrame } from './errors.js'; +import { + assertExactKeys, + requireEntityId, + requireExactRecord, + requireId, + requireRecord, + requireString, +} from './codec.js'; +import { defineOperation } from './operation-spec.js'; + +export interface TurnStartInput { + sessionId: string; + turnId: string; + text: string; +} + +export interface TurnQueryInput { + sessionId: string; + turnId: string; +} + +export interface TurnStopInput { + sessionId: string; + turnId: string; + runId: string; +} + +export type TurnRunStatus = + | 'admitted' + | 'created' + | 'running' + | 'waiting_permission' + | 'completed' + | 'failed' + | 'cancelled'; + +interface TurnSnapshotBase { + sessionId: string; + turnId: string; + runId: string; +} + +export type TurnSnapshot = + | (TurnSnapshotBase & { + status: Exclude; + }) + | (TurnSnapshotBase & { status: 'completed'; terminalEventId: string }) + | (TurnSnapshotBase & { + status: 'failed'; + terminalEventId: string; + failureClass: string; + }) + | (TurnSnapshotBase & { + status: 'cancelled'; + terminalEventId: string; + abortSource: string; + }); + +export const TURN_OPERATION_SPECS = { + 'turn.start': defineOperation({ + mode: 'command', + availability: 'ready', + errors: [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'not_found', + 'session_archived', + 'session_busy', + 'operation_conflict', + 'internal_failure', + ] as const, + decodeInput: decodeTurnStartInput, + decodeOutput: decodeTurnSnapshot, + }), + 'turn.query': defineOperation({ + mode: 'query', + availability: 'ready', + errors: [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'not_found', + 'internal_failure', + ] as const, + decodeInput: decodeTurnQueryInput, + decodeOutput: decodeTurnSnapshot, + }), + 'turn.stop': defineOperation({ + mode: 'control', + availability: 'ready', + errors: [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'not_found', + 'operation_conflict', + 'internal_failure', + ] as const, + decodeInput: decodeTurnStopInput, + decodeOutput: decodeTurnSnapshot, + }), +} as const; + +function decodeTurnStartInput(value: unknown): TurnStartInput { + const record = requireExactRecord(value, 'turn.start input', ['sessionId', 'turnId', 'text']); + return { + sessionId: requireEntityId(record.sessionId, 'sessionId'), + turnId: requireEntityId(record.turnId, 'turnId'), + text: requireString(record.text, 'text', 48 * 1024), + }; +} + +function decodeTurnQueryInput(value: unknown): TurnQueryInput { + const record = requireExactRecord(value, 'turn.query input', ['sessionId', 'turnId']); + return { + sessionId: requireEntityId(record.sessionId, 'sessionId'), + turnId: requireEntityId(record.turnId, 'turnId'), + }; +} + +function decodeTurnStopInput(value: unknown): TurnStopInput { + const record = requireExactRecord(value, 'turn.stop input', ['sessionId', 'turnId', 'runId']); + return { + sessionId: requireEntityId(record.sessionId, 'sessionId'), + turnId: requireEntityId(record.turnId, 'turnId'), + runId: requireEntityId(record.runId, 'runId'), + }; +} + +function decodeTurnSnapshot(value: unknown): TurnSnapshot { + const record = requireRecord(value, 'Turn snapshot'); + const base = { + sessionId: requireEntityId(record.sessionId, 'sessionId'), + turnId: requireEntityId(record.turnId, 'turnId'), + runId: requireEntityId(record.runId, 'runId'), + }; + const status = requireTurnRunStatus(record.status); + if (status === 'completed') { + assertExactKeys(record, 'completed Turn snapshot', [ + 'sessionId', + 'turnId', + 'runId', + 'status', + 'terminalEventId', + ]); + return { + ...base, + status, + terminalEventId: requireId(record.terminalEventId, 'terminalEventId'), + }; + } + if (status === 'failed') { + assertExactKeys(record, 'failed Turn snapshot', [ + 'sessionId', + 'turnId', + 'runId', + 'status', + 'terminalEventId', + 'failureClass', + ]); + return { + ...base, + status, + terminalEventId: requireId(record.terminalEventId, 'terminalEventId'), + failureClass: requireString(record.failureClass, 'failureClass', 128), + }; + } + if (status === 'cancelled') { + assertExactKeys(record, 'cancelled Turn snapshot', [ + 'sessionId', + 'turnId', + 'runId', + 'status', + 'terminalEventId', + 'abortSource', + ]); + return { + ...base, + status, + terminalEventId: requireId(record.terminalEventId, 'terminalEventId'), + abortSource: requireString(record.abortSource, 'abortSource', 128), + }; + } + assertExactKeys(record, 'non-terminal Turn snapshot', ['sessionId', 'turnId', 'runId', 'status']); + return { ...base, status }; +} + +function requireTurnRunStatus(value: unknown): TurnRunStatus { + if ( + value === 'admitted' || + value === 'created' || + value === 'running' || + value === 'waiting_permission' || + value === 'completed' || + value === 'failed' || + value === 'cancelled' + ) { + return value; + } + throw invalidProtocolFrame('Invalid Turn run status'); +} diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index e73786a33c..995dc19c1c 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -2,7 +2,9 @@ import { randomUUID } from 'node:crypto'; import { BackendRegistry, FakeBackend, SessionManager } from '@maka/runtime'; import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; import type { RuntimeHostComposition, RuntimeHostCompositionContext } from './host-kernel.js'; +import { RootAdmissionOwner } from './root-admission-owner.js'; import { RootTurnCoordinator } from './root-turn-coordinator.js'; +import { SessionAdmissionGate } from './session-admission-gate.js'; export async function createExecutionRuntimeHostComposition( context: RuntimeHostCompositionContext, @@ -18,9 +20,13 @@ export async function createExecutionRuntimeHostComposition( newId: randomUUID, now: Date.now, }); + const sessionAdmission = new SessionAdmissionGate(); + const rootAdmissionOwner = new RootAdmissionOwner(stores.agentRunStore); const coordinator = new RootTurnCoordinator( manager, stores, + sessionAdmission, + rootAdmissionOwner, context.acquireResidency, context.requestDrain, ); diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index 7fdb332c32..984410c029 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -26,6 +26,7 @@ import { type ConnectionOperationLease, } from './connection-session.js'; import { + composeOperationHandlers, type DomainOperationHandlerMap, type OperationResidency, type OperationHandlerMap, @@ -290,7 +291,7 @@ export class RuntimeHostKernel { ): Promise { if (!(await this.#readAdmissionState())) return 'host_draining'; if ( - HOST_OPERATION_SPECS[frame.operation].admission !== 'bootstrap' && + HOST_OPERATION_SPECS[frame.operation].availability !== 'bootstrap' && this.#state !== 'ready' ) { return 'host_not_ready'; @@ -379,19 +380,21 @@ export class RuntimeHostKernel { } #createOperationHandlers(domainHandlers: DomainOperationHandlerMap): OperationHandlerMap { - return { - 'host.status': async () => ({ - ok: true, - result: { - hostEpoch: this.hostEpoch, - state: this.#state, - connections: this.#acceptedTransports.size, - activeOperations: this.#activeOperations, - activeResidencies: this.#activeResidencies, - }, - }), - ...domainHandlers, - }; + return composeOperationHandlers( + { + 'host.status': async () => ({ + ok: true, + result: { + hostEpoch: this.hostEpoch, + state: this.#state, + connections: this.#acceptedTransports.size, + activeOperations: this.#activeOperations, + activeResidencies: this.#activeResidencies, + }, + }), + }, + domainHandlers, + ); } #waitForOperations(): Promise { diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index fb240dd5c2..fe24c860af 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -1,6 +1,7 @@ import { HOST_OPERATION_SPECS, type ClientSurface, + decodeOperationOutcome, type HostOperationErrorCode, type OperationInput, type OperationKey, @@ -35,6 +36,32 @@ export type OperationHandlerMap = { export type DomainOperationKey = Exclude; export type DomainOperationHandlerMap = Pick; +export function composeOperationHandlers( + ...handlerMaps: readonly Partial[] +): OperationHandlerMap { + const combined: Partial = {}; + for (const handlers of handlerMaps) { + for (const key of Object.keys(handlers)) { + if (!Object.hasOwn(HOST_OPERATION_SPECS, key)) { + throw new Error(`Unknown Runtime Host operation handler: ${key}`); + } + if (Object.hasOwn(combined, key)) { + throw new Error(`Duplicate Runtime Host operation handler: ${key}`); + } + const handler = handlers[key as OperationKey]; + if (typeof handler !== 'function') { + throw new Error(`Invalid Runtime Host operation handler: ${key}`); + } + Object.assign(combined, { [key]: handler }); + } + } + const missing = Object.keys(HOST_OPERATION_SPECS).filter((key) => !Object.hasOwn(combined, key)); + if (missing.length > 0) { + throw new Error(`Missing Runtime Host operation handlers: ${missing.join(', ')}`); + } + return combined as OperationHandlerMap; +} + export async function dispatchOperation( request: RequestFrame, handlers: OperationHandlerMap, @@ -73,7 +100,7 @@ async function dispatchTypedOperation( const handler = handlers[request.operation] as OperationHandler; let outcome: OperationOutcome; try { - outcome = await handler(request.input, context); + outcome = decodeOperationOutcome(request.operation, await handler(request.input, context)); } catch { return operationFailureResponse( request as RequestFrame, diff --git a/packages/runtime-host/src/server/root-admission-owner.ts b/packages/runtime-host/src/server/root-admission-owner.ts new file mode 100644 index 0000000000..39d1706af1 --- /dev/null +++ b/packages/runtime-host/src/server/root-admission-owner.ts @@ -0,0 +1,92 @@ +import type { + AdmitRootTurnInput, + AdmitRootTurnResult, + RootTurnAdmission, + RootTurnAdmissionStore, +} from '@maka/storage/execution-stores'; + +type OwnedAdmitRootTurnInput = Omit; + +export class RootAdmissionOwner { + readonly #admissionsBySession = new Map>(); + readonly #tips = new Map(); + readonly #poisonedSessions = new Set(); + + constructor(private readonly store: RootTurnAdmissionStore) {} + + assertKnownAdmission(admission: RootTurnAdmission): void { + const known = this.#admissionsBySession.get(admission.sessionId)?.get(admission.turnId); + if (!known || !sameRootAdmission(known, admission)) { + throw new Error('Root Turn admission identity changed within one Host Epoch'); + } + } + + async recoverSession(sessionId: string): Promise { + if (this.#admissionsBySession.has(sessionId)) { + throw new Error(`Root Turn recovery chain was already installed for Session ${sessionId}`); + } + const admissions = await this.store.listRootTurnAdmissionsForRecovery(sessionId); + const snapshots = admissions.map(snapshotAdmission); + const byTurnId = new Map(); + for (const admission of snapshots) byTurnId.set(admission.turnId, admission); + this.#admissionsBySession.set(sessionId, byTurnId); + const tip = snapshots.at(-1); + if (tip) this.#tips.set(sessionId, tip); + return snapshots; + } + + async admitRootTurn(input: OwnedAdmitRootTurnInput): Promise { + if (this.#poisonedSessions.has(input.sessionId)) { + throw new Error(`Root Turn admission state is uncertain for Session ${input.sessionId}`); + } + const current = this.#tips.get(input.sessionId); + try { + const result = await this.store.admitRootTurn({ + ...input, + previousRootTurnId: current?.turnId ?? null, + }); + const admission = result.admission; + if ( + admission.sessionId !== input.sessionId || + admission.turnId !== input.turnId || + admission.previousRootTurnId !== (current?.turnId ?? null) + ) { + throw new Error('Durable Root Turn admission does not extend the owned chain'); + } + + const byTurnId = this.#admissionsBySession.get(input.sessionId) ?? new Map(); + const known = byTurnId.get(admission.turnId); + if (known && !sameRootAdmission(known, admission)) { + throw new Error('Root Turn admission identity changed within one Host Epoch'); + } + const snapshot = snapshotAdmission(admission); + byTurnId.set(admission.turnId, snapshot); + this.#admissionsBySession.set(input.sessionId, byTurnId); + this.#tips.set(input.sessionId, snapshot); + return result; + } catch (error) { + this.#poisonedSessions.add(input.sessionId); + throw error; + } + } +} + +function sameRootAdmission(left: RootTurnAdmission, right: RootTurnAdmission): boolean { + return ( + left.schemaVersion === right.schemaVersion && + left.sessionId === right.sessionId && + left.turnId === right.turnId && + left.runId === right.runId && + left.userMessageId === right.userMessageId && + left.previousRootTurnId === right.previousRootTurnId && + left.normalizedInput.text === right.normalizedInput.text && + left.admittedAt === right.admittedAt + ); +} + +function snapshotAdmission(admission: RootTurnAdmission): RootTurnAdmission { + return Object.freeze({ + ...admission, + normalizedInput: Object.freeze({ ...admission.normalizedInput }), + }); +} diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index 680d9578c8..67609f07da 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -16,6 +16,8 @@ import type { } from '../protocol/index.js'; import type { RuntimeHostResidency } from './host-kernel.js'; import type { ConnectionContext, DomainOperationHandlerMap } from './operation-dispatcher.js'; +import { RootAdmissionOwner } from './root-admission-owner.js'; +import { SessionAdmissionGate } from './session-admission-gate.js'; interface ActiveRootTurn { turnId: string; @@ -53,13 +55,14 @@ export class RootTurnCoordinator { }; readonly #activeBySession = new Map(); - readonly #sessionGateTails = new Map>(); readonly #recoveryAdmissionsBySession = new Map(); private readonly stores: ExecutionStoresWriter<'interactive'>; constructor( private readonly manager: SessionManager, stores: ExecutionStoresWriter<'interactive'>, + private readonly sessionAdmission: SessionAdmissionGate, + private readonly rootAdmissionOwner: RootAdmissionOwner, private readonly acquireRecoveryResidency: () => RuntimeHostResidency, private readonly requestHostDrain: () => void, ) { @@ -70,9 +73,7 @@ export class RootTurnCoordinator { const sessions = await this.stores.sessionStore.listForRecovery(); const plans: RecoverySessionPlan[] = []; for (const session of sessions) { - const admissions = await this.stores.agentRunStore.listRootTurnAdmissionsForRecovery( - session.id, - ); + const admissions = await this.rootAdmissionOwner.recoverSession(session.id); const messages = await this.stores.sessionStore.readMessagesForRecovery(session.id); const runs = await this.stores.agentRunStore.listSessionRunsForRecovery(session.id); const runsById = new Map(runs.map((run) => [run.runId, run])); @@ -180,7 +181,7 @@ export class RootTurnCoordinator { turnId: admission.turnId, text: admission.normalizedInput.text, }; - const disposition = await this.withSessionGate(sessionId, () => + const disposition = await this.sessionAdmission.run(sessionId, () => this.prepareAdmittedTurn(input, admission, this.acquireRecoveryResidency), ); const outcome = await this.resolveStartDisposition(input, disposition); @@ -211,12 +212,13 @@ export class RootTurnCoordinator { private startTurn(input: TurnStartInput, context: ConnectionContext): Promise { return this.runCommand(async () => { - const disposition = await this.withSessionGate(input.sessionId, async () => { + const disposition = await this.sessionAdmission.run(input.sessionId, async () => { const existing = await this.stores.agentRunStore.readRootTurnAdmission( input.sessionId, input.turnId, ); if (existing) { + this.rootAdmissionOwner.assertKnownAdmission(existing); if (existing.normalizedInput.text !== input.text) { return completedStart( operationConflict('Turn identity was already admitted with a different payload'), @@ -240,7 +242,7 @@ export class RootTurnCoordinator { return completedStart(sessionBusy('Session already has an active root Turn')); } - const admission = await this.stores.agentRunStore.admitRootTurn({ + const admission = await this.rootAdmissionOwner.admitRootTurn({ sessionId: input.sessionId, turnId: input.turnId, proposedRunId: randomUUID(), @@ -260,12 +262,13 @@ export class RootTurnCoordinator { } private queryTurn(input: TurnQueryInput): Promise> { - return this.withSessionGate(input.sessionId, async () => { + return this.sessionAdmission.run(input.sessionId, async () => { const admission = await this.stores.agentRunStore.readRootTurnAdmission( input.sessionId, input.turnId, ); if (!admission) return notFound('Turn was not admitted'); + this.rootAdmissionOwner.assertKnownAdmission(admission); return { ok: true, result: await this.readCanonicalSnapshot(input.sessionId, input.turnId, admission.runId), @@ -275,12 +278,13 @@ export class RootTurnCoordinator { private stopTurn(input: TurnStopInput): Promise> { return this.runCommand(() => - this.withSessionGate(input.sessionId, async () => { + this.sessionAdmission.run(input.sessionId, async () => { const admission = await this.stores.agentRunStore.readRootTurnAdmission( input.sessionId, input.turnId, ); if (!admission) return notFound('Turn was not admitted'); + this.rootAdmissionOwner.assertKnownAdmission(admission); if (admission.runId !== input.runId) { return operationConflict('Run identity does not match the admitted Turn'); } @@ -503,25 +507,6 @@ export class RootTurnCoordinator { } } - private async withSessionGate(sessionId: string, operation: () => Promise): Promise { - const previous = this.#sessionGateTails.get(sessionId) ?? Promise.resolve(); - let release!: () => void; - const current = new Promise((resolve) => { - release = resolve; - }); - const tail = previous.then(() => current); - this.#sessionGateTails.set(sessionId, tail); - await previous; - try { - return await operation(); - } finally { - release(); - if (this.#sessionGateTails.get(sessionId) === tail) { - this.#sessionGateTails.delete(sessionId); - } - } - } - private async runCommand(operation: () => Promise): Promise { try { return await operation(); diff --git a/packages/runtime-host/src/server/session-admission-gate.ts b/packages/runtime-host/src/server/session-admission-gate.ts new file mode 100644 index 0000000000..d69a4ec644 --- /dev/null +++ b/packages/runtime-host/src/server/session-admission-gate.ts @@ -0,0 +1,20 @@ +export class SessionAdmissionGate { + readonly #tails = new Map>(); + + async run(sessionId: string, operation: () => Promise | T): Promise { + const previous = this.#tails.get(sessionId) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise((resolve) => { + release = resolve; + }); + const tail = previous.then(() => current); + this.#tails.set(sessionId, tail); + await previous; + try { + return await operation(); + } finally { + release(); + if (this.#tails.get(sessionId) === tail) this.#tails.delete(sessionId); + } + } +} diff --git a/packages/storage/src/__tests__/execution-stores.test.ts b/packages/storage/src/__tests__/execution-stores.test.ts index 890e50bb1b..21393b452a 100644 --- a/packages/storage/src/__tests__/execution-stores.test.ts +++ b/packages/storage/src/__tests__/execution-stores.test.ts @@ -13,7 +13,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, test } from 'node:test'; import type { AgentRunEvent, AgentRunHeader, RuntimeEvent } from '@maka/core'; -import { createAgentRunStore } from '../agent-run-store.js'; +import { createAgentRunStore, ROOT_TURN_ADMISSION_SCHEMA_VERSION } from '../agent-run-store.js'; import { authenticateExecutionStoresReader, authenticateExecutionStoresWriter, @@ -150,6 +150,7 @@ describe('execution stores', () => { turnId: 'turn-1', proposedRunId: 'run-1', proposedUserMessageId: 'message-1', + previousRootTurnId: null, normalizedInput: { text: 'hello' }, admittedAt: 10, }); @@ -161,6 +162,7 @@ describe('execution stores', () => { turnId: 'turn-1', proposedRunId: 'run-never-used', proposedUserMessageId: 'message-never-used', + previousRootTurnId: null, normalizedInput: { text: 'hello' }, admittedAt: 20, }); @@ -175,12 +177,25 @@ describe('execution stores', () => { turnId: 'turn-1', proposedRunId: 'run-never-used', proposedUserMessageId: 'message-never-used', + previousRootTurnId: null, normalizedInput: { text: 'changed' }, admittedAt: 30, }); assert.equal(conflict.kind, 'conflict'); assert.equal(conflict.admission.runId, 'run-1'); + const lineageConflict = await stores.agentRunStore.admitRootTurn({ + sessionId: session.id, + turnId: 'turn-1', + proposedRunId: 'run-never-used', + proposedUserMessageId: 'message-never-used', + previousRootTurnId: 'different-predecessor', + normalizedInput: { text: 'hello' }, + admittedAt: 40, + }); + assert.equal(lineageConflict.kind, 'conflict'); + assert.equal(lineageConflict.admission.previousRootTurnId, null); + const header = runHeader(session.id, first.admission.runId); await stores.agentRunStore.createRun(header); const bytes = await readFile( @@ -225,6 +240,7 @@ describe('execution stores', () => { turnId: 'turn-1', proposedRunId: 'run-1', proposedUserMessageId: 'message-1', + previousRootTurnId: null, normalizedInput: { text: 'hello' }, admittedAt: 9, }); @@ -432,6 +448,7 @@ describe('execution stores', () => { turnId: 'turn-1', proposedRunId: 'run-1', proposedUserMessageId: 'message-1', + previousRootTurnId: null, normalizedInput: { text: 'hello' }, admittedAt: 10, }); @@ -458,6 +475,86 @@ describe('execution stores', () => { }); }); + test('strict recovery orders same-millisecond admissions by predecessor lineage', async () => { + await withRoot(async ({ root }) => { + const store = createAgentRunStore(root); + await store.admitRootTurn({ + sessionId: 'session', + turnId: 'z-root', + proposedRunId: 'run-root', + proposedUserMessageId: 'message-root', + previousRootTurnId: null, + normalizedInput: { text: 'root' }, + admittedAt: 100, + }); + await store.admitRootTurn({ + sessionId: 'session', + turnId: 'a-successor', + proposedRunId: 'run-successor', + proposedUserMessageId: 'message-successor', + previousRootTurnId: 'z-root', + normalizedInput: { text: 'successor' }, + admittedAt: 100, + }); + + const chain = await store.listRootTurnAdmissionsForRecovery('session'); + assert.deepEqual( + chain.map((admission) => admission.turnId), + ['z-root', 'a-successor'], + ); + }); + }); + + test('strict recovery rejects malformed predecessor graphs', async () => { + await withRoot(async ({ root }) => { + const store = createAgentRunStore(root); + const admissionsRoot = join(root, 'sessions', 'session', 'turn-admissions'); + const install = async ( + records: readonly ReturnType[], + ): Promise => { + await rm(admissionsRoot, { recursive: true, force: true }); + await mkdir(admissionsRoot, { recursive: true }); + await Promise.all( + records.map((record) => + writeFile( + join(admissionsRoot, `${record.turnId}.json`), + `${JSON.stringify(record)}\n`, + 'utf8', + ), + ), + ); + }; + + await install([rootAdmissionRecord('root', null), rootAdmissionRecord('missing', 'absent')]); + await assert.rejects( + () => store.listRootTurnAdmissionsForRecovery('session'), + /missing predecessor/, + ); + + await install([rootAdmissionRecord('root-a', null), rootAdmissionRecord('root-b', null)]); + await assert.rejects( + () => store.listRootTurnAdmissionsForRecovery('session'), + /exactly one root/, + ); + + await install([ + rootAdmissionRecord('root', null), + rootAdmissionRecord('left', 'root'), + rootAdmissionRecord('right', 'root'), + ]); + await assert.rejects(() => store.listRootTurnAdmissionsForRecovery('session'), /branches/); + + await install([ + rootAdmissionRecord('cycle-a', 'cycle-b'), + rootAdmissionRecord('cycle-b', 'cycle-a'), + ]); + await assert.rejects( + () => store.listRootTurnAdmissionsForRecovery('session'), + /exactly one root/, + ); + }); + }); + test('strict recovery enumeration fails on malformed durable entities', async () => { await withRoot(async ({ root }) => { const capability = await resolveStorageRoot({ @@ -475,6 +572,7 @@ describe('execution stores', () => { turnId: 'turn-1', proposedRunId: 'run-1', proposedUserMessageId: 'message-1', + previousRootTurnId: null, normalizedInput: { text: 'hello' }, admittedAt: 10, }); @@ -508,6 +606,19 @@ describe('execution stores', () => { }); }); +function rootAdmissionRecord(turnId: string, previousRootTurnId: string | null) { + return { + schemaVersion: ROOT_TURN_ADMISSION_SCHEMA_VERSION, + sessionId: 'session', + turnId, + runId: `run-${turnId}`, + userMessageId: `message-${turnId}`, + previousRootTurnId, + normalizedInput: { text: turnId }, + admittedAt: 100, + }; +} + async function withRoot( run: (paths: { base: string; root: string }) => Promise, ): Promise { diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index 728533e541..36979ea221 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -37,7 +37,7 @@ const SAFE_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/; const EXCLUSIVE_TEMP_SUFFIX_PATTERN = /^\d+\.[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.tmp$/; -export const ROOT_TURN_ADMISSION_SCHEMA_VERSION = 1 as const; +export const ROOT_TURN_ADMISSION_SCHEMA_VERSION = 2 as const; export interface RootTurnAdmissionInput { text: string; @@ -49,6 +49,7 @@ export interface RootTurnAdmission { turnId: string; runId: string; userMessageId: string; + previousRootTurnId: string | null; normalizedInput: RootTurnAdmissionInput; admittedAt: number; } @@ -58,6 +59,7 @@ export interface AdmitRootTurnInput { turnId: string; proposedRunId: string; proposedUserMessageId: string; + previousRootTurnId: string | null; normalizedInput: RootTurnAdmissionInput; admittedAt: number; } @@ -153,6 +155,12 @@ class FileAgentRunStore implements DurableAgentRunStore { assertSafeId(input.turnId, 'Invalid turn id'); assertSafeId(input.proposedRunId, 'Invalid run id'); assertSafeId(input.proposedUserMessageId, 'Invalid user message id'); + if (input.previousRootTurnId !== null) { + assertSafeId(input.previousRootTurnId, 'Invalid previous root turn id'); + if (input.previousRootTurnId === input.turnId) { + throw new Error('Root turn admission cannot reference itself'); + } + } const normalizedInput = normalizeRootTurnAdmissionInput(input.normalizedInput); if (!Number.isSafeInteger(input.admittedAt) || input.admittedAt < 0) { throw new Error('Invalid root turn admission timestamp'); @@ -163,6 +171,7 @@ class FileAgentRunStore implements DurableAgentRunStore { turnId: input.turnId, runId: input.proposedRunId, userMessageId: input.proposedUserMessageId, + previousRootTurnId: input.previousRootTurnId, normalizedInput, admittedAt: input.admittedAt, }; @@ -176,7 +185,8 @@ class FileAgentRunStore implements DurableAgentRunStore { if (created) return { kind: 'admitted', admission }; const existing = await this.readRootTurnAdmission(input.sessionId, input.turnId); if (!existing) throw new Error(`Root turn admission disappeared: ${input.turnId}`); - return existing.normalizedInput.text === normalizedInput.text + return existing.previousRootTurnId === input.previousRootTurnId && + existing.normalizedInput.text === normalizedInput.text ? { kind: 'existing', admission: existing } : { kind: 'conflict', admission: existing }; } @@ -226,9 +236,7 @@ class FileAgentRunStore implements DurableAgentRunStore { throw new Error(`Invalid root turn admission entry: ${entry.name}`); } if (removedStagingFile) await syncDirectory(admissionsRoot); - return admissions.sort( - (a, b) => a.admittedAt - b.admittedAt || a.turnId.localeCompare(b.turnId), - ); + return orderRootTurnAdmissionChain(sessionId, admissions); } async updateRun( @@ -1231,6 +1239,10 @@ function normalizeRootTurnAdmission( isSafeId(record.runId) && typeof record.userMessageId === 'string' && isSafeId(record.userMessageId) && + (record.previousRootTurnId === null || + (typeof record.previousRootTurnId === 'string' && + isSafeId(record.previousRootTurnId) && + record.previousRootTurnId !== turnId)) && Number.isSafeInteger(record.admittedAt) && (record.admittedAt as number) >= 0 && hasExactKeys(record, [ @@ -1239,6 +1251,7 @@ function normalizeRootTurnAdmission( 'turnId', 'runId', 'userMessageId', + 'previousRootTurnId', 'normalizedInput', 'admittedAt', ]); @@ -1251,11 +1264,58 @@ function normalizeRootTurnAdmission( turnId, runId: record.runId as string, userMessageId: record.userMessageId as string, + previousRootTurnId: record.previousRootTurnId as string | null, normalizedInput: normalizeRootTurnAdmissionInput(record.normalizedInput), admittedAt: record.admittedAt as number, }; } +function orderRootTurnAdmissionChain( + sessionId: string, + admissions: readonly RootTurnAdmission[], +): RootTurnAdmission[] { + if (admissions.length === 0) return []; + const byTurnId = new Map(admissions.map((admission) => [admission.turnId, admission])); + if (byTurnId.size !== admissions.length) { + throw new Error(`Session ${sessionId} has duplicate root turn admissions`); + } + for (const admission of admissions) { + const predecessor = admission.previousRootTurnId; + if (predecessor !== null && !byTurnId.has(predecessor)) { + throw new Error( + `Root turn admission ${admission.turnId} has missing predecessor ${predecessor}`, + ); + } + } + const roots = admissions.filter((admission) => admission.previousRootTurnId === null); + if (roots.length !== 1) { + throw new Error(`Session ${sessionId} must have exactly one root turn admission root`); + } + const childByTurnId = new Map(); + for (const admission of admissions) { + const predecessor = admission.previousRootTurnId; + if (predecessor === null) continue; + const existing = childByTurnId.get(predecessor); + if (existing) { + throw new Error( + `Root turn admission ${predecessor} branches to ${existing.turnId} and ${admission.turnId}`, + ); + } + childByTurnId.set(predecessor, admission); + } + + const ordered: RootTurnAdmission[] = []; + let current: RootTurnAdmission | undefined = roots[0]; + while (current) { + ordered.push(current); + current = childByTurnId.get(current.turnId); + } + if (ordered.length !== admissions.length) { + throw new Error(`Session ${sessionId} root turn admissions do not form one linear chain`); + } + return ordered; +} + function normalizeRootTurnAdmissionInput(value: unknown): RootTurnAdmissionInput { if (!value || typeof value !== 'object' || Array.isArray(value)) { throw new Error('Invalid root turn normalized input: expected an object'); diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 2c6f0110ed..1a6efdd522 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -33,8 +33,11 @@ const executionStoresWriterKinds = new WeakMap(); const executionStoresReaderKinds = new WeakMap(); export type { + AdmitRootTurnInput, + AdmitRootTurnResult, RootTurnAdmission, RootTurnAdmissionInput, + RootTurnAdmissionStore, } from './agent-run-store.js'; export type ExecutionSessionWriter = SessionStore; From 3af657f5d5d54603e0505967c87845a7d052cddc Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 23 Jul 2026 15:39:43 +0800 Subject: [PATCH 2/4] fix(runtime-host): keep experimental protocol at v0 --- packages/runtime-host/src/__tests__/protocol.test.ts | 9 +++++---- packages/runtime-host/src/protocol/index.ts | 6 +++--- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 88dafc99fd..cf581f0a35 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -13,14 +13,15 @@ import { composeOperationSpecMaps } from '../protocol/operation-spec.js'; describe('Runtime Host bootstrap protocol', () => { test('selects the highest mutually supported protocol and rejects a gap', () => { + assert.equal(negotiateProtocol({ min: 0, max: 0 }, { min: 0, max: 0 }), 0); assert.equal(negotiateProtocol({ min: 1, max: 3 }, { min: 2, max: 4 }), 3); - assert.equal(negotiateProtocol({ min: 1, max: 1 }, { min: 2, max: 2 }), undefined); + assert.equal(negotiateProtocol({ min: 0, max: 0 }, { min: 1, max: 1 }), undefined); }); 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: 1, protocolMax: 1 })}\n` + + `${JSON.stringify({ kind: 'hello', clientInstanceId: '客户端', surface: 'tui', protocolMin: 0, protocolMax: 0 })}\n` + `${JSON.stringify({ requestId: 'status-1', operation: 'host.status', input: {} })}\n`, ); const split = wire.indexOf(Buffer.from('端')) + 1; @@ -31,8 +32,8 @@ describe('Runtime Host bootstrap protocol', () => { kind: 'hello', clientInstanceId: '客户端', surface: 'tui', - protocolMin: 1, - protocolMax: 1, + protocolMin: 0, + protocolMax: 0, }); assert.deepEqual(decodeClientFrame(frames[1]), { requestId: 'status-1', diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index cb263774e2..a218160218 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -14,7 +14,7 @@ export { RuntimeHostProtocolError } from './errors.js'; export * from './operations.js'; export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; -export const RUNTIME_HOST_PROTOCOL_VERSION = 2 as const; +export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; export const RUNTIME_HOST_MAX_FRAME_BYTES = 64 * 1024; export type ClientSurface = 'desktop' | 'tui' | 'run' | 'bot' | 'open_gateway' | 'inspect'; @@ -83,7 +83,7 @@ export function validateProtocolRange(range: ProtocolRange): void { if ( !Number.isSafeInteger(range.min) || !Number.isSafeInteger(range.max) || - range.min < 1 || + range.min < 0 || range.max < range.min ) { throw invalidProtocolFrame('Invalid protocol range'); @@ -240,7 +240,7 @@ export class ProtocolFrameDecoder { } function requireProtocolVersion(value: unknown, label: string): number { - if (!Number.isSafeInteger(value) || (value as number) < 1) { + if (!Number.isSafeInteger(value) || (value as number) < 0) { throw invalidProtocolFrame(`Invalid ${label}`); } return value as number; From 414be18f82f23ff3a8b1949d1e4143bda0cf65f6 Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 23 Jul 2026 15:40:07 +0800 Subject: [PATCH 3/4] fix(runtime-host): close sqlite session metadata --- packages/runtime-host/src/server/execution-composition.ts | 8 +++++++- packages/storage/src/execution-stores.ts | 1 + 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 995dc19c1c..d06518b4e0 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -37,6 +37,12 @@ export async function createExecutionRuntimeHostComposition( await manager.recoverInterruptedSessionsStrict(stores); await coordinator.recover(); }, - close: () => coordinator.close(), + close: async () => { + try { + await coordinator.close(); + } finally { + stores.sessionStore.close?.(); + } + }, }; } diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 1a6efdd522..674bb10bed 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -157,6 +157,7 @@ async function openExecutionStoresForWrite( setGeneratedTitleIfAbsent: (sessionId, title) => run(() => sessionStore.setGeneratedTitleIfAbsent(sessionId, title)), remove: (sessionId) => run(() => sessionStore.remove(sessionId)), + close: () => sessionStore.close?.(), }, agentRunStore: { createRun: (header, options) => run(() => agentRunStore.createRun(header, options)), From f8a993507fcff53151a0907210f19dc606d07a90 Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 23 Jul 2026 17:01:30 +0800 Subject: [PATCH 4/4] fix(storage): await session metadata initialization on close --- apps/desktop/src/main/app-lifecycle.ts | 2 +- packages/cli/src/runtime-bootstrap.ts | 2 +- .../src/server/execution-composition.ts | 2 +- .../session-metadata-maintenance.test.ts | 4 +- .../__tests__/settings-store-usage.test.ts | 2 +- .../__tests__/sqlite-session-store.test.ts | 39 ++++++++++++++----- packages/storage/src/execution-stores.ts | 4 +- packages/storage/src/session-store.ts | 12 +++++- 8 files changed, 49 insertions(+), 18 deletions(-) diff --git a/apps/desktop/src/main/app-lifecycle.ts b/apps/desktop/src/main/app-lifecycle.ts index 2ff1fc92c3..371172deed 100644 --- a/apps/desktop/src/main/app-lifecycle.ts +++ b/apps/desktop/src/main/app-lifecycle.ts @@ -325,6 +325,6 @@ export function wireAppLifecycle(deps: AppLifecycleDeps): void { if (result.status === 'rejected') console.error('[shutdown] cleanup failed:', result.reason); } runtimePersistence.close(); - sessionStore.close?.(); + await sessionStore.close?.(); } } diff --git a/packages/cli/src/runtime-bootstrap.ts b/packages/cli/src/runtime-bootstrap.ts index 595e2defb4..6388bd776f 100644 --- a/packages/cli/src/runtime-bootstrap.ts +++ b/packages/cli/src/runtime-bootstrap.ts @@ -889,7 +889,7 @@ export async function createMakaCliRuntimeContext( goalManager.dispose(); await shellRuns.terminateAll(); shellRunListeners.clear(); - store.close?.(); + await store.close?.(); runtimePersistence.close(); }, }; diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index d06518b4e0..b7fab924eb 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -41,7 +41,7 @@ export async function createExecutionRuntimeHostComposition( try { await coordinator.close(); } finally { - stores.sessionStore.close?.(); + await stores.sessionStore.close?.(); } }, }; diff --git a/packages/storage/src/__tests__/session-metadata-maintenance.test.ts b/packages/storage/src/__tests__/session-metadata-maintenance.test.ts index cb2f420430..e22d57add7 100644 --- a/packages/storage/src/__tests__/session-metadata-maintenance.test.ts +++ b/packages/storage/src/__tests__/session-metadata-maintenance.test.ts @@ -85,7 +85,7 @@ describe('session metadata migration maintenance', () => { /already exists/, ); } finally { - store.close?.(); + await store.close?.(); await rm(container, { recursive: true, force: true }); } }); @@ -108,7 +108,7 @@ describe('session metadata migration maintenance', () => { /differ from the source/, ); } finally { - store.close?.(); + await store.close?.(); await rm(root, { recursive: true, force: true }); } }); diff --git a/packages/storage/src/__tests__/settings-store-usage.test.ts b/packages/storage/src/__tests__/settings-store-usage.test.ts index e7cd3b8251..e9fa39763e 100644 --- a/packages/storage/src/__tests__/settings-store-usage.test.ts +++ b/packages/storage/src/__tests__/settings-store-usage.test.ts @@ -81,7 +81,7 @@ describe('SettingsStore.usageStats request logs', () => { assert.equal(stats.logs[0]?.provider, 'sqlite-provider'); assert.equal(stats.logs[0]?.model, 'sqlite-runtime-model'); } finally { - sessions.close?.(); + await sessions.close?.(); await rm(workspaceRoot, { recursive: true, force: true }); } }); diff --git a/packages/storage/src/__tests__/sqlite-session-store.test.ts b/packages/storage/src/__tests__/sqlite-session-store.test.ts index 8891e0c892..573987dfde 100644 --- a/packages/storage/src/__tests__/sqlite-session-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-store.test.ts @@ -12,6 +12,27 @@ import { import { createSqliteSessionMetadataStore } from '../sqlite-session-metadata-store.js'; describe('default SQLite session metadata store', () => { + test('waits for in-flight legacy metadata import before closing SQLite', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-store-close-ready-')); + const legacy = createLegacyFileSessionStore(root); + const created = await legacy.create(makeInput({ name: 'Imported before close' })); + const store = createSessionStore(root); + + try { + await store.close?.(); + const metadata = createSqliteSessionMetadataStore( + join(root, SQLITE_SESSION_METADATA_DATABASE_NAME), + ); + try { + assert.equal((await metadata.read(created.id)).header.name, 'Imported before close'); + } finally { + metadata.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + test('uses SQLite as canonical metadata while keeping transcript bodies in JSONL', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-default-session-store-')); const store = createSessionStore(root); @@ -49,16 +70,16 @@ describe('default SQLite session metadata store', () => { await stat(join(root, SQLITE_SESSION_METADATA_DATABASE_NAME)); await assert.rejects(() => stat(join(root, 'runtime.sqlite')), { code: 'ENOENT' }); - store.close?.(); + await store.close?.(); const reopened = createSessionStore(root); try { assert.equal((await reopened.readHeader(created.id)).name, 'SQLite title'); assert.equal((await reopened.readMessages(created.id)).length, 1); } finally { - reopened.close?.(); + await reopened.close?.(); } } finally { - store.close?.(); + await store.close?.(); await rm(root, { recursive: true, force: true }); } }); @@ -93,7 +114,7 @@ describe('default SQLite session metadata store', () => { modelId: 'fake-model', }); } finally { - first.close?.(); + await first.close?.(); } const reopened = createSessionStore(root); @@ -105,7 +126,7 @@ describe('default SQLite session metadata store', () => { [created.id], ); } finally { - reopened.close?.(); + await reopened.close?.(); await rm(root, { recursive: true, force: true }); } }); @@ -118,7 +139,7 @@ describe('default SQLite session metadata store', () => { header = await store.create(makeInput({ name: 'Delete me' })); await store.remove(header.id); } finally { - store.close?.(); + await store.close?.(); } const sessionDir = join(root, 'sessions', header.id); @@ -130,7 +151,7 @@ describe('default SQLite session metadata store', () => { assert.deepEqual(await reopened.list(), []); await assert.rejects(() => reopened.readHeader(header.id), /not found/); } finally { - reopened.close?.(); + await reopened.close?.(); await rm(root, { recursive: true, force: true }); } }); @@ -149,7 +170,7 @@ describe('default SQLite session metadata store', () => { try { await assert.rejects(() => store.list(), /Invalid legacy session header/); } finally { - store.close?.(); + await store.close?.(); } const metadata = createSqliteSessionMetadataStore( @@ -183,7 +204,7 @@ describe('default SQLite session metadata store', () => { try { await assert.rejects(() => store.list(), /has no SQLite metadata/); } finally { - store.close?.(); + await store.close?.(); await rm(root, { recursive: true, force: true }); } }); diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 674bb10bed..d74756e652 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -157,7 +157,9 @@ async function openExecutionStoresForWrite( setGeneratedTitleIfAbsent: (sessionId, title) => run(() => sessionStore.setGeneratedTitleIfAbsent(sessionId, title)), remove: (sessionId) => run(() => sessionStore.remove(sessionId)), - close: () => sessionStore.close?.(), + close: async () => { + await sessionStore.close?.(); + }, }, agentRunStore: { createRun: (header, options) => run(() => agentRunStore.createRun(header, options)), diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 186fcd0c57..6835d1716c 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -67,7 +67,7 @@ export interface SessionStore { rename(sessionId: string, name: string): Promise; setGeneratedTitleIfAbsent(sessionId: string, title: string): Promise; remove(sessionId: string): Promise; - close?(): void; + close?(): Promise; } export function createSessionStore(workspaceRoot: string): SessionStore { @@ -83,6 +83,7 @@ class SqliteSessionStore implements SessionStore { private readonly files: FileSessionStore; private readonly metadata: SqliteSessionMetadataStore; private readonly ready: Promise; + private closePromise: Promise | null = null; constructor(workspaceRoot: string) { this.files = new FileSessionStore(workspaceRoot); @@ -93,6 +94,7 @@ class SqliteSessionStore implements SessionStore { workspaceRoot, destination: this.metadata, }).then(() => {}); + void this.ready.catch(() => {}); } async create(input: CreateSessionInput): Promise { @@ -278,7 +280,13 @@ class SqliteSessionStore implements SessionStore { await this.files.remove(sessionId); } - close(): void { + close(): Promise { + this.closePromise ??= this.closeAfterReady(); + return this.closePromise; + } + + private async closeAfterReady(): Promise { + await this.ready.catch(() => {}); this.metadata.close(); }