feat(runtime-host): bound shutdown with process fail-stop

packages/runtime-host/src/__tests__/execution-host.test.ts

@@ -216,7 +216,8 @@ test('graceful Host shutdown stops and drains an active Turn before releasing ow
       text: FAKE_ASK_USER_QUESTION_PROMPT,
     });
 
-    await fixture.stopHost(host);
+    const exit = await fixture.stopHost(host);
+    assert.deepEqual(exit, { code: 0, signal: null });
     await client.closed;
 
     const successor = await fixture.startHost();
@@ -602,12 +603,19 @@ class ExecutionFixture {
     await owner?.close();
   }
 
-  async stopHost(host: ExecutionHostHandle): Promise<void> {
+  async stopHost(
+    host: ExecutionHostHandle,
+  ): Promise<{ code: number | null; signal: NodeJS.Signals | null }> {
     if (host.child.exitCode === null && host.child.signalCode === null) {
       host.child.kill('SIGTERM');
     }
-    await withTimeout(waitForExit(host.child), PROCESS_TIMEOUT_MS, 'execution Host did not stop');
+    const exit = await withTimeout(
+      waitForExitResult(host.child),
+      PROCESS_TIMEOUT_MS,
+      'execution Host did not stop',
+    );
     this.#children.delete(host.child);
+    return exit;
   }
 
   async killHost(host: ExecutionHostHandle): Promise<void> {
@@ -889,6 +897,30 @@ function waitForExit(child: ChildProcess): Promise<void> {
   });
 }
 
+function waitForExitResult(
+  child: ChildProcess,
+): Promise<{ code: number | null; signal: NodeJS.Signals | null }> {
+  if (child.exitCode !== null || child.signalCode !== null) {
+    return Promise.resolve({ code: child.exitCode, signal: child.signalCode });
+  }
+  return new Promise((resolve, reject) => {
+    const cleanup = () => {
+      child.off('error', onError);
+      child.off('exit', onExit);
+    };
+    const onError = (error: Error) => {
+      cleanup();
+      reject(error);
+    };
+    const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
+      cleanup();
+      resolve({ code, signal });
+    };
+    child.once('error', onError);
+    child.once('exit', onExit);
+  });
+}
+
 async function acquireReader(capability: StorageRootCapability<'interactive'>) {
   const deadline = Date.now() + PROCESS_TIMEOUT_MS;
   while (true) {

packages/runtime-host/src/__tests__/fixtures/execution-host.ts

@@ -1,4 +1,5 @@
 import { startExecutionRuntimeHostCandidate } from '../../server/execution-candidate.js';
+import { runRuntimeHostProcessLifecycle } from '../../server/process-lifecycle.js';
 
 const [rootPath, expectedRootId, idleGraceRaw] = process.argv.slice(2);
 if (!rootPath || !expectedRootId || !/^[a-f0-9]{64}$/.test(expectedRootId)) {
@@ -22,17 +23,8 @@ process.send?.({
   endpoint: result.host.endpoint,
 });
 
-let closing = false;
-const close = () => {
-  if (closing) return;
-  closing = true;
-  void result.host.close();
-};
-process.once('SIGINT', close);
-process.once('SIGTERM', close);
-process.once('disconnect', close);
 try {
-  await result.host.closed;
+  await runRuntimeHostProcessLifecycle(result.host, { closeOnDisconnect: true });
 } catch {
   process.exitCode = 1;
 } finally {

packages/runtime-host/src/__tests__/fixtures/uncooperative-host.ts

@@ -0,0 +1,66 @@
+import {
+  resolveExistingStorageRoot,
+  tryAcquireInteractiveRootOwner,
+} from '@maka/storage/root-authority';
+import { RuntimeHostKernel, type RuntimeHostComposition } from '../../server/host-kernel.js';
+import { runRuntimeHostProcessLifecycle } from '../../server/process-lifecycle.js';
+
+const [rootPath, expectedRootId, shutdownGraceRaw] = process.argv.slice(2);
+if (!rootPath || !expectedRootId || !/^[a-f0-9]{64}$/.test(expectedRootId)) {
+  throw new Error('usage: uncooperative-host <root> <expected-root-id> <shutdown-grace-ms>');
+}
+const shutdownGraceMs = Number(shutdownGraceRaw);
+if (!Number.isSafeInteger(shutdownGraceMs) || shutdownGraceMs <= 0) {
+  throw new Error('uncooperative-host requires a positive shutdown grace');
+}
+
+const capability = await resolveExistingStorageRoot({
+  path: rootPath,
+  kind: 'interactive',
+  expectedRootId,
+});
+const owner = await tryAcquireInteractiveRootOwner(capability);
+if (!owner) throw new Error('uncooperative-host could not acquire the Interactive root');
+
+const host = await RuntimeHostKernel.start({
+  owner,
+  idleGraceMs: 60_000,
+  shutdownGraceMs,
+  compositionFactory: async (context): Promise<RuntimeHostComposition> => ({
+    handlers: {
+      'turn.start': async () => {
+        context.acquireResidency();
+        process.send?.({ type: 'operation-blocked' });
+        return new Promise<never>(() => undefined);
+      },
+      'turn.query': async () => ({
+        ok: false,
+        error: { code: 'operation_unavailable', message: 'Operation unavailable in test Host' },
+      }),
+      'turn.stop': async () => ({
+        ok: false,
+        error: { code: 'operation_unavailable', message: 'Operation unavailable in test Host' },
+      }),
+    },
+    async recover() {},
+    async close() {},
+  }),
+});
+
+process.on('message', (message: unknown) => {
+  if (
+    message &&
+    typeof message === 'object' &&
+    (message as { type?: unknown }).type === 'shutdown'
+  ) {
+    void host.close();
+    process.send?.({ type: 'shutdown-requested' });
+  }
+});
+process.send?.({ type: 'ready', hostEpoch: host.hostEpoch, endpoint: host.endpoint });
+
+try {
+  await runRuntimeHostProcessLifecycle(host, { closeOnDisconnect: true });
+} catch {
+  process.exitCode = 1;
+}

packages/runtime-host/src/__tests__/host-kernel.test.ts

@@ -713,6 +713,112 @@ describe('non-serving Runtime Host kernel', () => {
     });
   });
 
+  test('forces an uncooperative command Host to exit before a successor acquires ownership', {
+    timeout: 10_000,
+  }, async () => {
+    await withHostPaths(async (paths) => {
+      const capability = await resolveStorageRoot({ path: paths.root, kind: 'interactive' });
+      const child = paths.resources.trackChild(
+        fork(
+          new URL('./fixtures/uncooperative-host.js', import.meta.url),
+          [paths.root, capability.rootId, '2000'],
+          { stdio: ['ignore', 'ignore', 'inherit', 'ipc'] },
+        ),
+      );
+      let transport: FramedTransport | undefined;
+      try {
+        const ready = await waitForUncooperativeHostMessage(child, 'ready');
+        transport = new FramedTransport(await openSocket(ready.endpoint));
+        await transport.write({
+          kind: 'hello',
+          clientInstanceId: 'bounded-shutdown-test',
+          surface: 'tui',
+          protocolMin: CURRENT_PROTOCOL.min,
+          protocolMax: CURRENT_PROTOCOL.max,
+        });
+        const handshake = decodeHostFrame(await transport.read(2_000));
+        assert.ok('kind' in handshake);
+        assert.equal(handshake.kind, 'accepted');
+
+        const blocked = waitForUncooperativeHostMessage(child, 'operation-blocked');
+        await transport.write({
+          requestId: 'blocked-turn-start',
+          operation: 'turn.start',
+          input: { sessionId: 'session', turnId: 'turn', text: 'block forever' },
+        });
+        await blocked;
+        const shutdownRequested = waitForUncooperativeHostMessage(child, 'shutdown-requested');
+        child.send({ type: 'shutdown' });
+        await shutdownRequested;
+
+        await transport.write({
+          requestId: 'post-drain-status',
+          operation: 'host.status',
+          input: {},
+        });
+        const rejectedOperation = decodeHostFrame(await transport.read(1_000));
+        assert.ok(!('kind' in rejectedOperation));
+        if (!('kind' in rejectedOperation)) {
+          assert.equal(rejectedOperation.requestId, 'post-drain-status');
+          assert.equal(rejectedOperation.operation, 'host.status');
+          assert.equal(rejectedOperation.ok, false);
+          if (!rejectedOperation.ok) assert.equal(rejectedOperation.error.code, 'host_draining');
+        }
+
+        const rejectedHandshakeTransport = new FramedTransport(await openSocket(ready.endpoint));
+        try {
+          await rejectedHandshakeTransport.write({
+            kind: 'hello',
+            clientInstanceId: 'post-drain-client',
+            surface: 'inspect',
+            protocolMin: CURRENT_PROTOCOL.min,
+            protocolMax: CURRENT_PROTOCOL.max,
+          });
+          assert.deepEqual(decodeHostFrame(await rejectedHandshakeTransport.read(1_000)), {
+            kind: 'draining',
+            hostEpoch: ready.hostEpoch,
+          });
+        } finally {
+          rejectedHandshakeTransport.destroy();
+        }
+
+        assert.equal(child.exitCode, null);
+        assert.equal(child.signalCode, null);
+        const contender = await tryAcquireInteractiveRootOwner(capability);
+        try {
+          assert.equal(contender, undefined);
+        } finally {
+          await contender?.close();
+        }
+        const exit = await withTimeout(
+          waitForChildExitResult(child),
+          5_000,
+          'uncooperative Runtime Host did not exit within its shutdown bound',
+        );
+        assert.deepEqual(exit, { code: 1, signal: null });
+
+        const successor = await startTestRuntimeHostCandidate(paths, {
+          rootPath: paths.root,
+          idleGraceMs: 10_000,
+        });
+        assert.equal(successor.kind, 'winner');
+        if (successor.kind !== 'winner') return;
+        assert.notEqual(successor.host.hostEpoch, ready.hostEpoch);
+        const connected = await retryConnect(paths, CURRENT_PROTOCOL);
+        assert.equal(connected.kind, 'connected');
+        if (connected.kind !== 'connected') return;
+        const status = await connected.connection.status();
+        assert.equal(status.hostEpoch, successor.host.hostEpoch);
+        await connected.connection.close();
+        await successor.host.close();
+      } finally {
+        transport?.destroy();
+        if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL');
+        await withTimeout(waitForExit(child), 1_000, 'uncooperative Host cleanup did not exit');
+      }
+    });
+  });
+
   test('startup rejects invalid lifecycle durations and releases the owner lock', async () => {
     await withHostPaths(async (paths) => {
       await assert.rejects(
@@ -731,6 +837,16 @@ describe('non-serving Runtime Host kernel', () => {
           }),
         RangeError,
       );
+      const capability = await resolveStorageRoot({ path: paths.root, kind: 'interactive' });
+      const owner = paths.resources.trackCloseable(
+        await tryAcquireInteractiveRootOwner(capability),
+      );
+      assert.ok(owner);
+      if (!owner) return;
+      await assert.rejects(
+        () => RuntimeHostKernel.start({ owner, shutdownGraceMs: 0 }),
+        RangeError,
+      );
       const retry = await startTestRuntimeHostCandidate(paths, {
         rootPath: paths.root,
         idleGraceMs: 0,
@@ -1199,6 +1315,80 @@ function isElectronParentLaunch(
   );
 }
 
+type UncooperativeHostMessage =
+  | { type: 'ready'; hostEpoch: string; endpoint: string }
+  | { type: 'operation-blocked' }
+  | { type: 'shutdown-requested' };
+
+function waitForUncooperativeHostMessage<T extends UncooperativeHostMessage['type']>(
+  child: ChildProcess,
+  type: T,
+): Promise<Extract<UncooperativeHostMessage, { type: T }>> {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      cleanup();
+      reject(new Error(`uncooperative Host did not report ${type}`));
+    }, 5_000);
+    const cleanup = () => {
+      clearTimeout(timer);
+      child.off('error', onError);
+      child.off('exit', onExit);
+      child.off('message', onMessage);
+    };
+    const onError = (error: Error) => {
+      cleanup();
+      reject(error);
+    };
+    const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
+      cleanup();
+      reject(new Error(`uncooperative Host exited before ${type}: ${code ?? signal}`));
+    };
+    const onMessage = (message: unknown) => {
+      if (!isUncooperativeHostMessage(message) || message.type !== type) return;
+      cleanup();
+      resolve(message as Extract<UncooperativeHostMessage, { type: T }>);
+    };
+    child.once('error', onError);
+    child.once('exit', onExit);
+    child.on('message', onMessage);
+  });
+}
+
+function isUncooperativeHostMessage(value: unknown): value is UncooperativeHostMessage {
+  if (!value || typeof value !== 'object') return false;
+  const message = value as Record<string, unknown>;
+  if (message.type === 'operation-blocked' || message.type === 'shutdown-requested') return true;
+  return (
+    message.type === 'ready' &&
+    typeof message.hostEpoch === 'string' &&
+    typeof message.endpoint === 'string'
+  );
+}
+
+function waitForChildExitResult(
+  child: ChildProcess,
+): Promise<{ code: number | null; signal: NodeJS.Signals | null }> {
+  if (child.exitCode !== null || child.signalCode !== null) {
+    return Promise.resolve({ code: child.exitCode, signal: child.signalCode });
+  }
+  return new Promise((resolve, reject) => {
+    const cleanup = () => {
+      child.off('error', onError);
+      child.off('exit', onExit);
+    };
+    const onError = (error: Error) => {
+      cleanup();
+      reject(error);
+    };
+    const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
+      cleanup();
+      resolve({ code, signal });
+    };
+    child.once('error', onError);
+    child.once('exit', onExit);
+  });
+}
+
 function waitForExit(child: ChildProcess): Promise<void> {
   if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve();
   return new Promise((resolve) => child.once('exit', () => resolve()));

packages/runtime-host/src/candidate-main.ts

@@ -1,20 +1,13 @@
 #!/usr/bin/env node
 import { startRuntimeHostCandidate, type RuntimeHostCandidateOptions } from './server/candidate.js';
+import { runRuntimeHostProcessLifecycle } from './server/process-lifecycle.js';
 
 const options = parseArguments(process.argv.slice(2));
 const result = await startRuntimeHostCandidate(options);
 if (result.kind === 'loser') process.exit(2);
 
-let closing = false;
-const close = () => {
-  if (closing) return;
-  closing = true;
-  void result.host.close();
-};
-process.once('SIGINT', close);
-process.once('SIGTERM', close);
 try {
-  await result.host.closed;
+  await runRuntimeHostProcessLifecycle(result.host);
 } catch {
   process.exitCode = 1;
 }