rea/tests/boundary/process/providerProcess.test.ts

import { access, stat } from "node:fs/promises";
import { tmpdir } from "node:os";
import { fileURLToPath } from "node:url";

import { afterEach, describe, expect, it, vi } from "vitest";

import { PendingOperations } from "../../../src/process/PendingOperations.js";
import { PrivateRuntimeRoot } from "../../../src/process/PrivateRuntimeRoot.js";
import { ProviderStartupDeadline } from "../../../src/process/ProviderDeadline.js";
import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js";
import {
  observeOwnedProcessLineage,
  systemProcessOwnershipHost,
} from "../../../src/process/ProcessOwnershipObservation.js";
import {
  ProviderProcessSupervisor,
  spawnOwnedProviderProcess,
  type ProviderProcessDiagnostic,
} from "../../../src/process/ProviderProcess.js";
import { WINDOWS_NATIVE_AUTHORITY_UNAVAILABLE_REASON } from "../../../src/process/WindowsAuthority.js";
import {
  spawnProviderProcessFixture,
  stopProviderProcessFixture,
  waitForDetachedProviderChild,
  waitForProviderProcessReady,
} from "../../fixtures/providerProcess.js";

const processFixturePath = fileURLToPath(
  new URL("../../fixtures/providerProcess.mjs", import.meta.url),
);

const waitForPidExit = async (
  pid: number,
  timeoutMs = 2_000,
): Promise<void> => {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    try {
      process.kill(pid, 0);
    } catch (cause: unknown) {
      if (cause instanceof Error && "code" in cause && cause.code === "ESRCH")
        return;
      throw cause;
    }
    await new Promise((resolve) => setTimeout(resolve, 10));
  }
  throw new Error(`Process ${String(pid)} did not exit within the test bound`);
};

afterEach(() => {
  vi.useRealTimers();
});

describe("provider process runtime and wait primitives", () => {
  it("allocates a private runtime root and removes it idempotently", async () => {
    const runtime = await PrivateRuntimeRoot.create({
      parent: tmpdir(),
      prefix: "rea-provider-test-",
    });
    expect((await stat(runtime.path)).mode & 0o777).toBe(0o700);

    const first = runtime.close();
    const second = runtime.close();
    expect(second).toBe(first);
    await Promise.all([first, second]);
    await expect(access(runtime.path)).rejects.toMatchObject({
      code: "ENOENT",
    });
  });

  it("fails closed for a Windows runtime root without native DACL authority", async () => {
    await expect(
      PrivateRuntimeRoot.create({ platform: "win32" }),
    ).rejects.toMatchObject({
      code: "private-runtime-root-authority-unavailable",
      message: WINDOWS_NATIVE_AUTHORITY_UNAVAILABLE_REASON,
    });
  });

  it("classifies external cancellation during startup", async () => {
    vi.useFakeTimers();
    const controller = new AbortController();
    const deadline = new ProviderStartupDeadline(1_000, controller.signal);

    const waiting = deadline.wait(1_000);
    controller.abort();
    await expect(waiting).resolves.toBe("aborted");
    expect(deadline.interruption).toBe("cancelled");
  });

  it("releases the owned timer and abort listener when disposed", async () => {
    vi.useFakeTimers();
    const controller = new AbortController();
    const deadline = new ProviderStartupDeadline(60_000, controller.signal);

    // The deadline owns exactly one live timer while it is undisposed.
    expect(vi.getTimerCount()).toBe(1);
    expect(deadline.interruption).toBeUndefined();

    deadline.dispose();

    // Disposal must release the timer it owns. Leaving it armed would keep a
    // live handle per provider startup for the whole timeout.
    expect(vi.getTimerCount()).toBe(0);

    // Disposal must also detach the abort listener. A listener that survived
    // disposal would still classify this deadline as cancelled, which is the
    // observable difference between a released and a retained deadline.
    controller.abort();
    expect(deadline.interruption).toBeUndefined();

    // No timer may be reintroduced by later ticks either.
    await vi.advanceTimersByTimeAsync(600_000);
    expect(vi.getTimerCount()).toBe(0);
  });

  it("uses one absolute startup deadline across interval waits", async () => {
    vi.useFakeTimers();
    const deadline = new ProviderStartupDeadline(100);
    const waiting = deadline.wait(1_000);

    await vi.advanceTimersByTimeAsync(100);

    await expect(waiting).resolves.toBe("aborted");
    expect(deadline.interruption).toBe("timeout");
    expect(deadline.remainingMs()).toBe(0);
    deadline.dispose();
    // Further ticks must not revive the deadline.
    await vi.advanceTimersByTimeAsync(60_000);
    expect(deadline.remainingMs()).toBe(0);
  });

  it("does not reclassify an elapsed deadline as later cancellation", async () => {
    vi.useFakeTimers();
    const controller = new AbortController();
    const deadline = new ProviderStartupDeadline(50, controller.signal);

    await vi.advanceTimersByTimeAsync(50);
    controller.abort();

    expect(deadline.signal.reason).toMatchObject({ name: "TimeoutError" });
    expect(deadline.interruption).toBe("timeout");
    // After disposal a further tick must not resurrect the abort listener.
    deadline.dispose();
    await vi.advanceTimersByTimeAsync(60_000);
    expect(deadline.interruption).toBe("timeout");
  });

  it("settles pending requests once on cancellation, timeout, and failure", async () => {
    vi.useFakeTimers();
    const operations = new PendingOperations<number, string>();
    const controller = new AbortController();
    const cancelled = operations.wait(1, {
      timeoutMs: 1_000,
      signal: controller.signal,
      timeoutValue: () => "timeout",
      cancelledValue: () => "cancelled",
    });
    controller.abort();
    await expect(cancelled).resolves.toBe("cancelled");
    // Settling by key after the fact proves the pending entry was released.
    expect(operations.settle(1, "late reply")).toBe(false);

    const timedOut = operations.wait(2, {
      timeoutMs: 50,
      timeoutValue: () => "timeout",
      cancelledValue: () => "cancelled",
    });
    await vi.advanceTimersByTimeAsync(50);
    await expect(timedOut).resolves.toBe("timeout");
    expect(operations.settle(2, "late reply")).toBe(false);

    const failed = operations.wait(3, {
      timeoutMs: 1_000,
      timeoutValue: () => "timeout",
      cancelledValue: () => "cancelled",
    });
    operations.failAll((key) => `failed:${String(key)}`);
    await expect(failed).resolves.toBe("failed:3");
    expect(operations.settle(3, "late reply")).toBe(false);
    expect(operations.size).toBe(0);
    // No pending entry and no live timeout: a later tick changes nothing.
    await vi.advanceTimersByTimeAsync(60_000);
    expect(operations.size).toBe(0);
  });
});

describe("provider process output and cleanup primitives", () => {
  it("retains complete stdout and stderr beyond the former fixed ceiling", async () => {
    const outputBytes = 70_000;
    const child = spawnProviderProcessFixture("burst", outputBytes);
    const diagnostics: ProviderProcessDiagnostic[] = [];
    const supervisor = new ProviderProcessSupervisor(
      { process: child, ownsProcessLifetime: true },
      {
        onDiagnostic: (event) => diagnostics.push(event),
      },
    );

    await expect(supervisor.waitForExit(2_000)).resolves.toBe(true);
    const snapshot = supervisor.snapshot();
    expect(snapshot).toMatchObject({
      stdout: {
        bytes: outputBytes,
        observedBytes: outputBytes,
      },
      stderr: {
        bytes: outputBytes,
        observedBytes: outputBytes,
      },
      exitCode: 23,
      signal: null,
    });
    expect(snapshot.stdout.text).toBe("o".repeat(outputBytes));
    expect(snapshot.stderr.text).toBe("e".repeat(outputBytes));
    expect(diagnostics).toContainEqual(
      expect.objectContaining({ type: "exit", code: 23 }),
    );
    await expect(supervisor.stop()).resolves.toEqual({
      status: "already-exited",
    });
  });

  // POSIX-only: libuv maps every signal to TerminateProcess on Windows, so a
  // child that ignores SIGTERM cannot survive the TERM step there and the
  // escalation below is unreachable.
  it.skipIf(process.platform === "win32")(
    "shares double-stop and escalates a stubborn child from TERM to KILL",
    async () => {
      const child = spawnProviderProcessFixture("stubborn");
      const supervisor = new ProviderProcessSupervisor({
        process: child,
        ownsProcessLifetime: true,
      });
      await waitForProviderProcessReady(child);

      const first = supervisor.stop({
        terminationGraceMs: 20,
        killGraceMs: 500,
      });
      const second = supervisor.stop({
        terminationGraceMs: 20,
        killGraceMs: 500,
      });
      // Concurrent callers must share one escalation instead of signalling
      // the same child twice.
      expect(second).toBe(first);
      await expect(first).resolves.toEqual({ status: "killed" });
      expect(child.signalCode).toBe("SIGKILL");
    },
  );

  it("detaches its process listeners when disposed", async () => {
    const child = spawnProviderProcessFixture("stubborn");
    const diagnostics: ProviderProcessDiagnostic[] = [];
    const supervisor = new ProviderProcessSupervisor(
      { process: child, ownsProcessLifetime: true },
      { onDiagnostic: (event) => diagnostics.push(event) },
    );
    await waitForProviderProcessReady(child);
    // The no-op listener only prevents Node's unhandled-'error' throw. The
    // supervisor's own listener is what is under test.
    child.on("error", () => undefined);

    try {
      // Control: while attached, a late event reaches the diagnostic channel.
      child.emit("error", new Error("attached provider failure"));
      expect(diagnostics).toContainEqual(
        expect.objectContaining({
          type: "error",
          message: "attached provider failure",
        }),
      );

      supervisor.dispose();

      // A released supervisor must ignore later events. A retained listener
      // would keep invoking callbacks for an owner that has already let go,
      // which is what accumulates handles across repeated provider lifecycles.
      const observed = diagnostics.length;
      child.emit("error", new Error("late provider failure"));
      child.emit("close", 0, null);
      expect(diagnostics).toHaveLength(observed);
    } finally {
      await stopProviderProcessFixture(child);
    }
  });

  it("honors verified group cleanup instead of direct process signaling", async () => {
    const child = spawnProviderProcessFixture("stubborn");
    await waitForProviderProcessReady(child);
    const cleanup = vi.fn(async () => {
      child.kill("SIGKILL");
      return { cleaned: true, signaled: true } as const;
    });
    const supervisor = new ProviderProcessSupervisor({
      process: child,
      ownsProcessLifetime: true,
      cleanup,
    });

    await expect(supervisor.stop({ killGraceMs: 500 })).resolves.toEqual({
      status: "verified-cleanup",
    });
    // Cleanup must be attempted. The exact attempt count is an internal retry
    // ladder, so it is deliberately not pinned here.
    expect(cleanup).toHaveBeenCalled();
  });

  it("cleans an owned group even after its launcher leader has exited", async () => {
    const child = spawnProviderProcessFixture("exit", 0);
    const cleanup = vi.fn(
      async () => ({ cleaned: true, signaled: false }) as const,
    );
    const supervisor = new ProviderProcessSupervisor({
      process: child,
      ownsProcessLifetime: true,
      cleanup,
    });
    await expect(supervisor.waitForExit(2_000)).resolves.toBe(true);

    await expect(supervisor.stop()).resolves.toEqual({
      status: "verified-cleanup",
    });
    // Cleanup must be attempted. The exact attempt count is an internal retry
    // ladder, so it is deliberately not pinned here.
    expect(cleanup).toHaveBeenCalled();
  });

  it("retains a live child after incomplete cleanup and permits a later stop", async () => {
    const child = spawnProviderProcessFixture("stubborn");
    await waitForProviderProcessReady(child);
    let release = false;
    const cleanup = () => {
      if (release) child.kill("SIGKILL");
      return Promise.resolve({ cleaned: true, signaled: release } as const);
    };
    const supervisor = new ProviderProcessSupervisor({
      process: child,
      ownsProcessLifetime: true,
      cleanup,
    });
    try {
      await expect(supervisor.stop({ killGraceMs: 20 })).resolves.toEqual({
        status: "incomplete",
        reason: "verified process-group cleanup did not stop the launcher",
      });
      expect(child.exitCode).toBeNull();
      expect(child.signalCode).toBeNull();
      release = true;
      await expect(supervisor.stop()).resolves.toEqual({
        status: "verified-cleanup",
      });
      expect(supervisor.snapshot().signal).toBe("SIGKILL");
    } finally {
      await stopProviderProcessFixture(child);
    }
  });
});

describe("provider process host injection", () => {
  it("uses injected host platform and environment when spawning", async () => {
    const spawned = await spawnOwnedProviderProcess({
      command: process.execPath,
      arguments: ["-e", "process.stdout.write(process.env.REA_HOST_TEST)"],
      runId: "injected-host-run",
      platform: "win32",
      hostEnvironment: { PATH: process.env.PATH, REA_HOST_TEST: "host-value" },
      env: { REA_HOST_TEST: "spawn-override" },
    });
    try {
      const supervisor = new ProviderProcessSupervisor({
        process: spawned.process,
        ownsProcessLifetime: true,
      });
      expect(await supervisor.waitForExit(2_000)).toBe(true);
      expect(supervisor.snapshot().stdout.text).toBe("spawn-override");
    } finally {
      if (spawned.process.exitCode === null)
        await new Promise<void>((resolve) =>
          spawned.process.once("close", () => resolve()),
        );
    }
  });
});

describe("provider process memory limits", () => {
  it("refuses a memory limit that the selected launcher cannot enforce", async () => {
    await expect(
      spawnOwnedProviderProcess({
        command: `${tmpdir()}/rea-provider-must-not-spawn-without-limit`,
        arguments: [],
        runId: "provider-process-unenforceable-limit-run",
        platform: "linux",
        memoryLimitBytes: 704 * 1024 * 1024,
      }),
    ).rejects.toThrow("require the Windows Job Object launcher");
  });
});

describe("provider startup cancellation", () => {
  it("rejects an already cancelled launch before preparing or spawning", async () => {
    const controller = new AbortController();
    const cancellation = new Error("provider launch already cancelled");
    controller.abort(cancellation);

    await expect(
      spawnOwnedProviderProcess({
        command: `${tmpdir()}/rea-provider-must-not-spawn-after-cancel`,
        arguments: [],
        runId: "provider-process-pre-aborted-run",
        signal: controller.signal,
      }),
    ).rejects.toBe(cancellation);
  });

  it.skipIf(process.platform !== "darwin")(
    "forwards cancellation through native ownership preparation before spawning",
    async () => {
      const controller = new AbortController();
      const cancellation = new Error("cancelled during native preparation");
      const prepare = vi.spyOn(systemProcessOwnershipHost, "prepare");
      prepare.mockImplementation(async (signal) => {
        expect(signal).toBe(controller.signal);
        await new Promise<void>((resolve) => {
          signal?.addEventListener("abort", () => resolve(), { once: true });
        });
      });

      try {
        const starting = spawnOwnedProviderProcess({
          command: `${tmpdir()}/rea-provider-must-not-spawn-after-cancel`,
          arguments: [],
          runId: "provider-process-cancelled-preparation-run",
          signal: controller.signal,
        });
        controller.abort(cancellation);
        await expect(starting).rejects.toBe(cancellation);
        expect(prepare).toHaveBeenCalledOnce();
      } finally {
        prepare.mockRestore();
      }
    },
  );
});

describe("provider process spawning primitives", () => {
  it("spawns an owned process group with exact identity coordinates", async () => {
    const runId = "provider-process-run";
    const spawned = await spawnOwnedProviderProcess({
      command: process.execPath,
      arguments: [processFixturePath, "graceful"],
      runId,
    });
    try {
      expect(spawned.ownership).toMatchObject({
        runId,
        leaderPid: spawned.process.pid,
        processGroupId: spawned.process.pid,
        expectedCommand: process.execPath,
        expectedParentPid: process.pid,
      });
      expect(spawned.process.stdout).not.toBeNull();
      expect(spawned.process.stderr).not.toBeNull();
    } finally {
      await stopProviderProcessFixture(spawned.process);
    }
  });

  it.skipIf(process.platform === "win32")(
    "cleans a real descendant that starts a distinct process group",
    async () => {
      const spawned = await spawnOwnedProviderProcess({
        command: process.execPath,
        arguments: [processFixturePath, "detached-child"],
        runId: "provider-process-lineage-run",
      });
      let detachedChildPid: number | undefined;
      try {
        detachedChildPid = await waitForDetachedProviderChild(spawned.process);
        const observation = await observeOwnedProcessLineage(spawned.ownership);
        expect(observation).toEqual({
          status: "verified",
          observedAt: expect.stringMatching(
            /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/u,
          ),
          lineage: {
            runId: "provider-process-lineage-run",
            launcherPid: spawned.process.pid,
            launcherParentPid: process.pid,
            processGroupId: spawned.process.pid,
            descendants: [
              {
                pid: detachedChildPid,
                parentPid: spawned.process.pid,
                processGroupId: detachedChildPid,
              },
            ],
          },
        });
        await expect(
          cleanupOwnedProcessGroup(spawned.ownership),
        ).resolves.toEqual({ cleaned: true, signaled: true });
        await Promise.all([
          waitForPidExit(spawned.ownership.leaderPid),
          waitForPidExit(detachedChildPid),
        ]);
      } finally {
        await cleanupOwnedProcessGroup(spawned.ownership);
        if (detachedChildPid !== undefined)
          await cleanupOwnedProcessGroup({
            runId: spawned.ownership.runId,
            leaderPid: detachedChildPid,
            processGroupId: detachedChildPid,
          });
        await waitForPidExit(spawned.ownership.leaderPid);
        if (detachedChildPid !== undefined)
          await waitForPidExit(detachedChildPid);
      }
    },
  );

  it.skipIf(process.platform === "win32")(
    "stops a detached fixture descendant when its launcher receives SIGTERM",
    async () => {
      const launcher = spawnProviderProcessFixture("detached-child");
      let detachedChildPid: number | undefined;
      const cleanup = async (): Promise<void> => {
        await stopProviderProcessFixture(launcher);
        if (detachedChildPid === undefined) return;
        try {
          process.kill(detachedChildPid, "SIGKILL");
        } catch (cause: unknown) {
          if (
            !(
              cause instanceof Error &&
              "code" in cause &&
              cause.code === "ESRCH"
            )
          )
            throw cause;
        }
      };
      try {
        detachedChildPid = await waitForDetachedProviderChild(launcher);
        const launcherPid = launcher.pid;
        if (launcherPid === undefined)
          throw new Error("Detached fixture launcher did not expose a PID");
        launcher.kill("SIGTERM");
        await Promise.all([
          waitForPidExit(launcherPid),
          waitForPidExit(detachedChildPid),
        ]);
      } catch (cause: unknown) {
        await cleanup();
        throw cause;
      }
      await cleanup();
    },
  );

  it("allows interpreter launchers to rely on parent and run-token identity", async () => {
    const spawned = await spawnOwnedProviderProcess({
      command: process.execPath,
      arguments: [processFixturePath, "graceful"],
      runId: "provider-process-interpreter-run",
      expectedCommand: null,
    });
    try {
      expect(spawned.ownership).toMatchObject({
        expectedParentPid: process.pid,
        runId: "provider-process-interpreter-run",
      });
      expect(spawned.ownership).not.toHaveProperty("expectedCommand");
    } finally {
      await stopProviderProcessFixture(spawned.process);
    }
  });

  it("rejects deterministic spawn failures without producing a process", async () => {
    await expect(
      spawnOwnedProviderProcess({
        command: `${tmpdir()}/rea-provider-command-that-does-not-exist`,
        arguments: [],
        runId: "missing-provider",
      }),
    ).rejects.toMatchObject({ code: "ENOENT" });
  });
});

describe("provider process configured-platform spawning", () => {
  it.skipIf(process.platform === "win32")(
    "uses the configured platform when establishing process-group identity",
    async () => {
      const windowsConfigured = await spawnOwnedProviderProcess({
        command: process.execPath,
        arguments: [processFixturePath, "graceful"],
        runId: "provider-process-configured-win32-run",
        platform: "win32",
      });
      const posixConfigured = await spawnOwnedProviderProcess({
        command: process.execPath,
        arguments: [processFixturePath, "graceful"],
        runId: "provider-process-configured-posix-run",
        platform: "linux",
      });
      try {
        await Promise.all([
          waitForProviderProcessReady(windowsConfigured.process),
          waitForProviderProcessReady(posixConfigured.process),
        ]);
        await expect(
          observeOwnedProcessLineage(windowsConfigured.ownership),
        ).resolves.toMatchObject({
          status: "unavailable",
          reason: "owned launcher process-group identity did not match",
          launcherPid: windowsConfigured.process.pid,
        });
        await expect(
          observeOwnedProcessLineage(posixConfigured.ownership),
        ).resolves.toMatchObject({
          status: "verified",
          lineage: {
            runId: "provider-process-configured-posix-run",
            launcherPid: posixConfigured.process.pid,
            launcherParentPid: process.pid,
            processGroupId: posixConfigured.process.pid,
            descendants: [],
          },
        });
      } finally {
        await Promise.all([
          stopProviderProcessFixture(windowsConfigured.process),
          stopProviderProcessFixture(posixConfigured.process),
        ]);
      }
    },
  );
});