Track terminal listening ports in activity events and thread state

apps/server/src/process/Layers/TerminalProcessInspector.test.ts

@@ -0,0 +1,162 @@
+import { spawn, type ChildProcess, type ChildProcessByStdio } from "node:child_process";
+import type { Readable } from "node:stream";
+
+import * as NodeChildProcessSpawner from "@effect/platform-node/NodeChildProcessSpawner";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { assert, it } from "@effect/vitest";
+import { Effect, Layer } from "effect";
+
+import { TerminalProcessInspector } from "../Services/TerminalProcessInspector.ts";
+import { TerminalProcessInspectorLive } from "./TerminalProcessInspector.ts";
+
+type ListenerProcess = ChildProcessByStdio<null, Readable, Readable>;
+
+interface StartedProcess {
+  readonly process: ListenerProcess;
+  readonly port: number;
+}
+
+const stopProcess = (child: ChildProcess) =>
+  Effect.callback<void>((resume) => {
+    if (child.exitCode !== null || child.signalCode !== null) {
+      resume(Effect.void);
+      return;
+    }
+
+    child.kill("SIGTERM");
+
+    const timeout = setTimeout(() => {
+      if (child.exitCode === null && child.signalCode === null) {
+        child.kill("SIGKILL");
+      }
+    }, 1_000);
+
+    child.once("exit", () => {
+      clearTimeout(timeout);
+      resume(Effect.void);
+    });
+  });
+
+const waitForPort = (child: ListenerProcess) =>
+  Effect.callback<number, Error>((resume) => {
+    const timeout = setTimeout(() => {
+      cleanup();
+      resume(Effect.fail(new Error("Timed out waiting for listener port")));
+    }, 3_000);
+
+    let stdout = "";
+    let stderr = "";
+
+    const cleanup = () => {
+      clearTimeout(timeout);
+      child.stdout.off("data", onStdout);
+      child.stderr.off("data", onStderr);
+      child.off("exit", onExit);
+    };
+
+    const onStdout = (chunk: Buffer) => {
+      stdout += chunk.toString("utf8");
+      const match = stdout.match(/PORT:(\d+)/);
+      if (!match?.[1]) return;
+      const port = Number(match[1]);
+      if (!Number.isInteger(port) || port <= 0) return;
+      cleanup();
+      resume(Effect.succeed(port));
+    };
+
+    const onStderr = (chunk: Buffer) => {
+      stderr += chunk.toString("utf8");
+    };
+
+    const onExit = (code: number | null) => {
+      cleanup();
+      resume(
+        Effect.fail(
+          new Error(
+            `Listener process exited before reporting port (code=${String(code)}): ${stderr.trim()}`,
+          ),
+        ),
+      );
+    };
+
+    child.stdout.on("data", onStdout);
+    child.stderr.on("data", onStderr);
+    child.on("exit", onExit);
+
+    return Effect.sync(cleanup);
+  });
+
+const startListenerProcess = Effect.gen(function* () {
+  const script = [
+    "const { createServer } = require('node:http');",
+    "const server = createServer((_req, res) => {",
+    "  res.statusCode = 200;",
+    "  res.end('ok');",
+    "});",
+    "server.listen(0, '127.0.0.1', () => {",
+    "  const address = server.address();",
+    "  if (typeof address !== 'object' || !address) process.exit(1);",
+    "  console.log(`PORT:${address.port}`);",
+    "});",
+    "const shutdown = () => server.close(() => process.exit(0));",
+    "process.on('SIGTERM', shutdown);",
+    "process.on('SIGINT', shutdown);",
+    "setInterval(() => {}, 10_000);",
+  ].join("");
+
+  const process = spawn("node", ["-e", script], {
+    stdio: ["ignore", "pipe", "pipe"],
+  });
+  const port = yield* waitForPort(process);
+  return { process, port } satisfies StartedProcess;
+});
+
+const nodeChildProcessLayer = NodeChildProcessSpawner.layer.pipe(Layer.provide(NodeServices.layer));
+
+const testLayer = TerminalProcessInspectorLive.pipe(Layer.provide(nodeChildProcessLayer));
+
+it.layer(testLayer)("TerminalProcessInspectorLive", (it) => {
+  it.effect("detects listening ports when the terminal root pid is the listener", () =>
+    Effect.acquireUseRelease(
+      startListenerProcess,
+      ({ process, port }) =>
+        Effect.gen(function* () {
+          const listenerPid = process.pid;
+          if (!listenerPid) {
+            return yield* Effect.fail(new Error("Listener process pid missing"));
+          }
+
+          const inspector = yield* TerminalProcessInspector;
+          const activity = yield* inspector.inspect(listenerPid);
+
+          assert.equal(activity.hasRunningSubprocess, true);
+          assert.deepStrictEqual(activity.runningPorts.includes(port), true);
+        }),
+      ({ process }) => stopProcess(process),
+    ),
+  );
+
+  it.effect("returns idle activity when root process has no children and no listening ports", () =>
+    Effect.acquireUseRelease(
+      Effect.sync(() =>
+        spawn("node", ["-e", "setInterval(() => {}, 10_000)"], {
+          stdio: ["ignore", "ignore", "ignore"],
+        }),
+      ),
+      (process) =>
+        Effect.gen(function* () {
+          const idlePid = process.pid;
+          if (!idlePid) {
+            return yield* Effect.fail(new Error("Idle process pid missing"));
+          }
+
+          const inspector = yield* TerminalProcessInspector;
+          const activity = yield* inspector.inspect(idlePid);
+
+          assert.equal(activity.hasRunningSubprocess, false);
+          assert.deepStrictEqual(activity.runningPorts, []);
+        }),
+      stopProcess,
+    ),
+  );
+});

apps/server/src/process/Layers/TerminalProcessInspector.ts

@@ -0,0 +1,232 @@
+import { Buffer } from "node:buffer";
+
+import { Effect, Layer, Option, PlatformError, Stream } from "effect";
+import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
+
+import { collectPosixProcessFamilyPids, checkPosixListeningPorts } from "../posix.ts";
+import type {
+  TerminalProcessInspectorShape,
+  TerminalSubprocessActivity,
+} from "../Services/TerminalProcessInspector.ts";
+import {
+  TerminalProcessInspector,
+  TerminalProcessInspectionError,
+} from "../Services/TerminalProcessInspector.ts";
+import { checkWindowsListeningPorts, collectWindowsChildPids } from "../win32.ts";
+
+const DEFAULT_COMMAND_KILL_GRACE_MS = 1_000;
+
+interface InspectorCommandResult {
+  readonly stdout: string;
+  readonly stderr: string;
+  readonly exitCode: number;
+}
+
+interface CollectOutputResult {
+  readonly text: string;
+  readonly truncated: boolean;
+}
+
+interface RunInspectorCommandInput {
+  readonly operation: string;
+  readonly terminalPid: number;
+  readonly command: string;
+  readonly args: ReadonlyArray<string>;
+  readonly timeoutMs: number;
+  readonly maxOutputBytes: number;
+}
+
+function commandLabel(command: string, args: ReadonlyArray<string>): string {
+  return [command, ...args].join(" ");
+}
+
+const collectOutput = Effect.fn("process.collectOutput")(function* (
+  stream: Stream.Stream<Uint8Array, PlatformError.PlatformError>,
+  maxOutputBytes: number,
+): Effect.fn.Return<CollectOutputResult, PlatformError.PlatformError> {
+  return yield* stream.pipe(
+    Stream.decodeText(),
+    Stream.runFold(
+      () => ({
+        text: "",
+        bytes: 0,
+        truncated: false,
+      }),
+      (state, chunk) => {
+        if (state.bytes >= maxOutputBytes) {
+          return {
+            ...state,
+            truncated: true,
+          };
+        }
+
+        const chunkBytes = Buffer.byteLength(chunk);
+        const remainingBytes = maxOutputBytes - state.bytes;
+        if (chunkBytes <= remainingBytes) {
+          return {
+            text: `${state.text}${chunk}`,
+            bytes: state.bytes + chunkBytes,
+            truncated: state.truncated,
+          };
+        }
+
+        const truncatedChunk = Buffer.from(chunk).subarray(0, remainingBytes).toString("utf8");
+        return {
+          text: `${state.text}${truncatedChunk}`,
+          bytes: state.bytes + remainingBytes,
+          truncated: true,
+        };
+      },
+    ),
+    Effect.map(({ text, truncated }) => ({ text, truncated })),
+  );
+});
+
+const makeTerminalProcessInspector = Effect.gen(function* () {
+  const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
+
+  const runInspectorCommand = Effect.fn("process.runInspectorCommand")(function* (
+    input: RunInspectorCommandInput,
+  ) {
+    const command = ChildProcess.make(input.command, [...input.args], {
+      killSignal: "SIGTERM",
+      forceKillAfter: DEFAULT_COMMAND_KILL_GRACE_MS,
+    });
+
+    return yield* Effect.gen(function* () {
+      const child = yield* spawner.spawn(command).pipe(
+        Effect.mapError(
+          (cause) =>
+            new TerminalProcessInspectionError({
+              operation: input.operation,
+              terminalPid: input.terminalPid,
+              command: commandLabel(input.command, input.args),
+              detail: "Failed to spawn inspector command.",
+              cause,
+            }),
+        ),
+      );
+
+      const [stdout, stderr, exitCode] = yield* Effect.all(
+        [
+          collectOutput(child.stdout, input.maxOutputBytes).pipe(
+            Effect.mapError(
+              (cause) =>
+                new TerminalProcessInspectionError({
+                  operation: input.operation,
+                  terminalPid: input.terminalPid,
+                  command: commandLabel(input.command, input.args),
+                  detail: "Failed to read stdout from inspector command.",
+                  cause,
+                }),
+            ),
+          ),
+          collectOutput(child.stderr, input.maxOutputBytes).pipe(
+            Effect.mapError(
+              (cause) =>
+                new TerminalProcessInspectionError({
+                  operation: input.operation,
+                  terminalPid: input.terminalPid,
+                  command: commandLabel(input.command, input.args),
+                  detail: "Failed to read stderr from inspector command.",
+                  cause,
+                }),
+            ),
+          ),
+          child.exitCode.pipe(
+            Effect.map(Number),
+            Effect.mapError(
+              (cause) =>
+                new TerminalProcessInspectionError({
+                  operation: input.operation,
+                  terminalPid: input.terminalPid,
+                  command: commandLabel(input.command, input.args),
+                  detail: "Failed to read inspector command exit code.",
+                  cause,
+                }),
+            ),
+          ),
+        ],
+        { concurrency: "unbounded" },
+      );
+
+      return {
+        stdout: stdout.text,
+        stderr: stderr.text,
+        exitCode,
+      } satisfies InspectorCommandResult;
+    }).pipe(
+      Effect.scoped,
+      Effect.timeoutOption(input.timeoutMs),
+      Effect.flatMap(
+        Option.match({
+          onNone: () =>
+            Effect.fail(
+              new TerminalProcessInspectionError({
+                operation: input.operation,
+                terminalPid: input.terminalPid,
+                command: commandLabel(input.command, input.args),
+                detail: "Inspector command timed out.",
+              }),
+            ),
+          onSome: Effect.succeed,
+        }),
+      ),
+    );
+  });
+
+  const inspect: TerminalProcessInspectorShape["inspect"] = Effect.fn("process.inspect")(function* (
+    terminalPid: number,
+  ) {
+    if (!Number.isInteger(terminalPid) || terminalPid <= 0) {
+      return {
+        hasRunningSubprocess: false,
+        runningPorts: [],
+      } satisfies TerminalSubprocessActivity;
+    }
+
+    if (process.platform === "win32") {
+      const childPids = yield* collectWindowsChildPids(terminalPid, runInspectorCommand);
+      const processPidsForPortScan = [terminalPid, ...childPids];
+      const runningPorts = yield* checkWindowsListeningPorts(processPidsForPortScan, {
+        terminalPid,
+        runCommand: runInspectorCommand,
+      });
+      return {
+        hasRunningSubprocess: childPids.length > 0 || runningPorts.length > 0,
+        runningPorts,
+      } satisfies TerminalSubprocessActivity;
+    }
+
+    const processFamilyPids = yield* collectPosixProcessFamilyPids(
+      terminalPid,
+      runInspectorCommand,
+    );
+    if (processFamilyPids.length === 0) {
+      return {
+        hasRunningSubprocess: false,
+        runningPorts: [],
+      } satisfies TerminalSubprocessActivity;
+    }
+
+    const subprocessPids = processFamilyPids.filter((pid: number) => pid !== terminalPid);
+    const runningPorts = yield* checkPosixListeningPorts(processFamilyPids, {
+      terminalPid,
+      runCommand: runInspectorCommand,
+      platform: process.platform,
+    });
+    return {
+      hasRunningSubprocess: subprocessPids.length > 0 || runningPorts.length > 0,
+      runningPorts,
+    } satisfies TerminalSubprocessActivity;
+  });
+
+  return {
+    inspect,
+  } satisfies TerminalProcessInspectorShape;
+});
+
+export const TerminalProcessInspectorLive = Layer.effect(
+  TerminalProcessInspector,
+  makeTerminalProcessInspector,
+);