feat(runtime-host): establish protocol and admission foundations

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}

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?.();
  }
 }

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();
  },
  };

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<TurnSnapshot> => 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;

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<void>((resolve) => {
+ markFactoryEntered = resolve;
+ });
+ const factoryReleased = new Promise<void>((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, {

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<OperationHandlerMap>),
+ /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<OperationHandlerMap>),
+ /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 <K extends OperationKey>(): Promise<OperationOutcome<K>> =>
+ ({
+ ok: false,
+ error: {
+ code: 'internal_failure',
+ message: 'not used',
+ },
+ }) as OperationOutcome<K>;
+ 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' },
+ };
+}