fix: harden Node runtime cleanup (#457)

* fix: harden node runtime cleanup

* fix: propagate command cancellation

* fix: address runtime cleanup review

* fix: close partial artifact writes before cleanup

Author: thymikee

.github/actions/setup-node-pnpm/action.yml

@@ -5,11 +5,11 @@ inputs:
   node-version:
     description: 'Node.js version'
     required: false
-    default: '22'
+    default: '22.19'
   pnpm-version:
     description: 'pnpm version'
     required: false
-    default: '9.12.2'
+    default: '10.33.2'
   install-deps:
     description: 'Whether to install dependencies with pnpm install --frozen-lockfile'
     required: false

package.json

@@ -13,6 +13,7 @@
     "url": "https://github.com/callstackincubator/agent-device/issues"
   },
   "type": "module",
+  "packageManager": "pnpm@10.33.2",
   "main": "dist/src/index.js",
   "types": "dist/src/index.d.ts",
   "exports": {
@@ -62,7 +63,7 @@
     }
   },
   "engines": {
-    "node": ">=22"
+    "node": ">=22.19"
   },
   "bin": {
     "agent-device": "bin/agent-device.mjs"
@@ -82,7 +83,7 @@
     "build:all": "pnpm build:node && pnpm build:xcuitest",
     "ad": "node bin/agent-device.mjs",
     "lint": "oxlint . --deny-warnings",
-    "format": "oxfmt --write src test skills package.json tsconfig.json .oxlintrc.json .oxfmtrc.json '!test/skillgym/.skillgym-results/**'",
+    "format": "oxfmt --write src test skills package.json tsconfig.json tsconfig.lib.json rslib.config.ts vitest.config.ts .github/actions/setup-node-pnpm/action.yml .oxlintrc.json .oxfmtrc.json '!test/skillgym/.skillgym-results/**'",
     "fallow": "fallow --summary",
     "fallow:baseline": "(fallow dead-code --save-baseline fallow-baselines/dead-code.json --summary || true) && (fallow dupes --save-baseline fallow-baselines/dupes.json --summary || true) && (fallow health --save-baseline fallow-baselines/health.json --summary || true)",
     "check:fallow": "fallow audit",

src/__tests__/client-companion-tunnel-worker.test.ts

@@ -38,12 +38,15 @@ function createDeferred<T>(): Deferred<T> {
 }
 
 function waitFor<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
-  return Promise.race([
-    promise,
-    delay(timeoutMs).then(() => {
-      throw new Error(`Timed out waiting for ${label}.`);
-    }),
-  ]);
+  let timeout: NodeJS.Timeout | null = null;
+  const timeoutPromise = new Promise<never>((_, reject) => {
+    timeout = setTimeout(() => {
+      reject(new Error(`Timed out waiting for ${label}.`));
+    }, timeoutMs);
+  });
+  return Promise.race([promise, timeoutPromise]).finally(() => {
+    if (timeout) clearTimeout(timeout);
+  });
 }
 
 function encodeTextFrame(text: string): Buffer {

src/__tests__/client-metro.test.ts

@@ -10,6 +10,7 @@ import os from 'node:os';
 import path from 'node:path';
 import { prepareMetroRuntime, reloadMetro } from '../client-metro.ts';
 import { AppError } from '../utils/errors.ts';
+import { isProcessAlive, waitForProcessExit } from '../utils/process-identity.ts';
 
 const TEST_TOKEN = 'agent-device-proxy-test-token';
 
@@ -163,11 +164,7 @@ test('prepareMetroRuntime starts Metro, bridges through proxy, and writes runtim
       socket.destroy();
     }
     await closeServer(proxyServer);
-    if (pid) {
-      try {
-        process.kill(pid);
-      } catch {}
-    }
+    await stopProcess(pid);
     rmSync(tempRoot, { recursive: true, force: true });
   }
 });
@@ -326,3 +323,19 @@ async function closeServer(server: ReturnType<typeof createServer>): Promise<voi
     });
   });
 }
+
+async function stopProcess(pid: number): Promise<void> {
+  if (!pid || !isProcessAlive(pid)) return;
+  try {
+    process.kill(pid, 'SIGTERM');
+  } catch {
+    return;
+  }
+  if (await waitForProcessExit(pid, 1_500)) return;
+  try {
+    process.kill(pid, 'SIGKILL');
+  } catch {
+    return;
+  }
+  await waitForProcessExit(pid, 1_500);
+}

src/daemon-client.ts

@@ -3,6 +3,7 @@ import http from 'node:http';
 import https from 'node:https';
 import fs from 'node:fs';
 import path from 'node:path';
+import { pipeline } from 'node:stream/promises';
 import { sleep } from './utils/timeouts.ts';
 import { AppError, toAppErrorCode } from './utils/errors.ts';
 import type {
@@ -1334,13 +1335,6 @@ export async function downloadRemoteArtifact(params: {
           });
           return;
         }
-        const output = fs.createWriteStream(params.destinationPath);
-        output.on('error', (error) => {
-          settle(error instanceof Error ? error : new Error(String(error)));
-        });
-        res.on('error', (error) => {
-          settle(error instanceof Error ? error : new Error(String(error)));
-        });
         res.on('aborted', () => {
           settle(
             new AppError('COMMAND_FAILED', 'Remote artifact download was interrupted', {
@@ -1349,10 +1343,10 @@ export async function downloadRemoteArtifact(params: {
             }),
           );
         });
-        output.on('finish', () => {
-          output.close(() => settle());
-        });
-        res.pipe(output);
+        void pipeline(res, fs.createWriteStream(params.destinationPath)).then(
+          () => settle(),
+          (error: unknown) => settle(error instanceof Error ? error : new Error(String(error))),
+        );
       },
     );
     const timeoutHandle = setTimeout(() => {

src/daemon.ts

@@ -8,6 +8,14 @@ import { createDaemonHttpServer } from './daemon/http-server.ts';
 import { trackDownloadableArtifact } from './daemon/artifact-tracking.ts';
 import { LeaseRegistry } from './daemon/lease-registry.ts';
 import { createRequestHandler } from './daemon/request-router.ts';
+import { teardownSessionResources } from './daemon/handlers/session-close.ts';
+import { closeDaemonServers } from './daemon/server-shutdown.ts';
+import type { SessionState } from './daemon/types.ts';
+import {
+  emitDiagnostic,
+  flushDiagnosticsToSessionFile,
+  withDiagnosticsScope,
+} from './utils/diagnostics.ts';
 import {
   acquireDaemonLock,
   parseIntegerEnv,
@@ -18,7 +26,15 @@ import {
   resolveDaemonCodeSignature,
   writeInfo,
 } from './daemon/server-lifecycle.ts';
-import { createSocketServer, listenHttpServer, listenNetServer } from './daemon/transport.ts';
+import {
+  createSocketServer,
+  listenHttpServer,
+  listenNetServer,
+  type DaemonServer,
+} from './daemon/transport.ts';
+import { sleep } from './utils/timeouts.ts';
+
+const DAEMON_SESSION_TEARDOWN_TIMEOUT_MS = 5_000;
 
 const daemonPaths = resolveDaemonPaths(process.env.AGENT_DEVICE_STATE_DIR);
 const { baseDir, infoPath, lockPath, logPath, sessionsDir } = daemonPaths;
@@ -45,6 +61,90 @@ const handleRequest = createRequestHandler({
   trackDownloadableArtifact,
 });
 
+async function emitFatalDiagnostic(error: unknown): Promise<void> {
+  await withDiagnosticsScope(
+    { command: 'daemon', session: 'daemon', logPath, debug: true },
+    async () => {
+      emitDiagnostic({
+        level: 'error',
+        phase: 'daemon_fatal',
+        data: {
+          error: error instanceof Error ? error.message : String(error),
+        },
+      });
+      flushDiagnosticsToSessionFile({ force: true });
+    },
+  );
+}
+
+async function teardownDaemonSessions(): Promise<void> {
+  const sessionsToStop = sessionStore.toArray();
+  await Promise.all(sessionsToStop.map(teardownDaemonSession));
+}
+
+async function teardownDaemonSession(session: SessionState) {
+  const teardown = teardownSessionResources(session, session.name).catch((error) => {
+    process.stderr.write(
+      `Daemon session teardown error (${session.name}): ${
+        error instanceof Error ? error.message : String(error)
+      }\n`,
+    );
+  });
+  await Promise.race([
+    teardown,
+    sleep(DAEMON_SESSION_TEARDOWN_TIMEOUT_MS).then(() => {
+      process.stderr.write(`Daemon session teardown timed out (${session.name}).\n`);
+    }),
+  ]);
+  sessionStore.writeSessionLog(session);
+  sessionStore.delete(session.name);
+}
+
+async function openDaemonServers(): Promise<{
+  servers: DaemonServer[];
+  socketPort?: number;
+  httpPort?: number;
+}> {
+  const servers: DaemonServer[] = [];
+  let socketPort: number | undefined;
+  let httpPort: number | undefined;
+  const startSocketServer = daemonServerMode !== 'http';
+  const startHttpServer = daemonServerMode !== 'socket';
+  if (startSocketServer) {
+    const socketServer = createSocketServer(handleRequest);
+    servers.push(socketServer);
+    socketPort = await listenNetServer(socketServer);
+  }
+
+  if (startHttpServer) {
+    const httpServer = await createDaemonHttpServer({ handleRequest, token });
+    servers.push(httpServer);
+    httpPort = await listenHttpServer(httpServer);
+  }
+  return { servers, socketPort, httpPort };
+}
+
+function publishDaemonInfo(socketPort: number | undefined, httpPort: number | undefined): void {
+  writeInfo(baseDir, infoPath, logPath, {
+    socketPort,
+    httpPort,
+    token,
+    version,
+    codeSignature: daemonCodeSignature,
+    processStartTime: daemonProcessStartTime,
+  });
+  if (socketPort) process.stdout.write(`AGENT_DEVICE_DAEMON_PORT=${socketPort}\n`);
+  if (httpPort) process.stdout.write(`AGENT_DEVICE_DAEMON_HTTP_PORT=${httpPort}\n`);
+}
+
+function closeServersBestEffort(servers: DaemonServer[]): void {
+  for (const server of servers) {
+    try {
+      server.close(() => {});
+    } catch {}
+  }
+}
+
 async function start(): Promise<void> {
   const lockData = {
     pid: process.pid,
@@ -58,73 +158,34 @@ async function start(): Promise<void> {
     return;
   }
 
-  const servers: Array<{ close: (cb: (err?: Error) => void) => void }> = [];
-  let socketPort: number | undefined;
-  let httpPort: number | undefined;
-
+  let servers: DaemonServer[] = [];
   try {
-    if (daemonServerMode === 'socket' || daemonServerMode === 'dual') {
-      const socketServer = createSocketServer(handleRequest);
-      servers.push(socketServer);
-      socketPort = await listenNetServer(socketServer);
-    }
-
-    if (daemonServerMode === 'http' || daemonServerMode === 'dual') {
-      const httpServer = await createDaemonHttpServer({ handleRequest, token });
-      servers.push(httpServer);
-      httpPort = await listenHttpServer(httpServer);
-    }
-
-    writeInfo(baseDir, infoPath, logPath, {
-      socketPort,
-      httpPort,
-      token,
-      version,
-      codeSignature: daemonCodeSignature,
-      processStartTime: daemonProcessStartTime,
-    });
-    if (socketPort) process.stdout.write(`AGENT_DEVICE_DAEMON_PORT=${socketPort}\n`);
-    if (httpPort) process.stdout.write(`AGENT_DEVICE_DAEMON_HTTP_PORT=${httpPort}\n`);
+    const opened = await openDaemonServers();
+    servers = opened.servers;
+    publishDaemonInfo(opened.socketPort, opened.httpPort);
   } catch (error) {
     const appErr = asAppError(error);
     process.stderr.write(`Daemon error: ${appErr.message}\n`);
-    for (const server of servers) {
-      try {
-        server.close(() => {});
-      } catch {}
-    }
+    closeServersBestEffort(servers);
     removeInfo(infoPath);
     releaseDaemonLock(lockPath);
     process.exit(1);
     return;
   }
 
   let shuttingDown = false;
-  const closeServers = async (): Promise<void> => {
-    await Promise.all(
-      servers.map(async (server) => {
-        await new Promise<void>((resolve) => {
-          try {
-            server.close(() => resolve());
-          } catch {
-            resolve();
-          }
-        });
-      }),
-    );
-  };
-  const shutdown = async () => {
+  const shutdown = async (options: { exitCode?: number; cause?: unknown } = {}) => {
     if (shuttingDown) return;
     shuttingDown = true;
-    await closeServers();
-    const sessionsToStop = sessionStore.toArray();
-    for (const session of sessionsToStop) {
-      sessionStore.writeSessionLog(session);
+    if (options.cause) {
+      await emitFatalDiagnostic(options.cause);
     }
+    await closeDaemonServers(servers);
+    await teardownDaemonSessions();
     await stopAllIosRunnerSessions();
     removeInfo(infoPath);
     releaseDaemonLock(lockPath);
-    process.exit(0);
+    process.exit(options.exitCode ?? 0);
   };
 
   process.on('SIGINT', () => {
@@ -139,13 +200,13 @@ async function start(): Promise<void> {
   process.on('uncaughtException', (err) => {
     const appErr = err instanceof AppError ? err : asAppError(err);
     process.stderr.write(`Daemon error: ${appErr.message}\n`);
-    void shutdown();
+    void shutdown({ exitCode: 1, cause: err });
   });
   process.on('unhandledRejection', (reason) => {
     const err = reason instanceof Error ? reason : new Error(String(reason));
     const appErr = err instanceof AppError ? err : asAppError(err);
     process.stderr.write(`Daemon error: ${appErr.message}\n`);
-    void shutdown();
+    void shutdown({ exitCode: 1, cause: err });
   });
 }
 

src/daemon/__tests__/server-shutdown.test.ts

@@ -0,0 +1,20 @@
+import { test } from 'vitest';
+import assert from 'node:assert/strict';
+import type { DaemonServer } from '../transport.ts';
+import { closeDaemonServers } from '../server-shutdown.ts';
+
+test('closeDaemonServers forces stuck servers after timeout', async () => {
+  let destroyed = false;
+  const server = {
+    close: () => {},
+    destroyConnections: () => {
+      destroyed = true;
+    },
+  } as unknown as DaemonServer;
+
+  const startedAt = Date.now();
+  await closeDaemonServers([server], 20);
+
+  assert.equal(destroyed, true);
+  assert.ok(Date.now() - startedAt < 500);
+});