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

import { readFile, rm, writeFile } from "node:fs/promises";
import { dirname, join } from "node:path";
import { expect } from "vitest";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
import { itWithCaptureCapability } from "./processCaptureCapability.js";

import { captureProcessScenario } from "../../../src/application/ProcessHarness.js";
import {
  digestProcessCommitment,
  parseProcessScenario,
  type ProcessExecutionPolicy,
} from "../../../src/domain/processCapture.js";

itWithCaptureCapability(
  "runs a login and reconnect replay machine inside process capture",
  async () => {
    const root = await createTestTempDirectory("rea-replay-machine-test-");
    const script = join(root, "client.mjs");
    await writeFile(
      script,
      [
        'const token = "machine-secret";',
        'await fetch(`${process.env.REA_REPLAY_HTTP_URL}/login`, { method: "POST", body: JSON.stringify({ token }) });',
        "const api = await fetch(`${process.env.REA_REPLAY_HTTP_URL}/api`, { headers: { authorization: token } });",
        'if (await api.text() !== "connect") throw new Error("bad API response");',
        "const connect = (acknowledge) => new Promise((resolve, reject) => {",
        "  const socket = new WebSocket(process.env.REA_REPLAY_WEBSOCKET_URL);",
        "  socket.addEventListener('error', reject);",
        "  socket.addEventListener('message', (event) => {",
        "    if (acknowledge) socket.send('ack');",
        "    else { console.log(`reply:${event.data}`); socket.close(); resolve(); }",
        "  });",
        "  if (acknowledge) socket.addEventListener('close', () => resolve());",
        "});",
        "await connect(true);",
        "await connect(false);",
      ].join("\n"),
    );
    const scenario = parseProcessScenario({
      executable: process.execPath,
      arguments: [script],
      working_directory: root,
      replay: {
        machine: {
          initial_state: "login",
          states: [
            { name: "login" },
            { name: "api" },
            { name: "socket" },
            { name: "ack" },
            { name: "reconnect" },
            { name: "complete", terminal: true },
          ],
          transitions: [
            {
              id: "login",
              from: "login",
              to: "api",
              trigger: { protocol: "http", method: "POST", path: "/login" },
              captures: [
                {
                  variable: "token",
                  value: { source: "request_json", path: ["token"] },
                  sensitive: true,
                },
              ],
              actions: [{ type: "http_response", status: 204, body: "" }],
              max_uses: 1,
            },
            {
              id: "api",
              from: "api",
              to: "socket",
              trigger: { protocol: "http", method: "GET", path: "/api" },
              guards: [
                {
                  variable: "token",
                  value: { source: "request_header", name: "authorization" },
                },
              ],
              actions: [
                { type: "http_response", status: 200, body: "connect" },
              ],
              max_uses: 1,
            },
            {
              id: "initial_socket",
              from: "socket",
              to: "ack",
              trigger: { protocol: "websocket_connect", path: "/ws" },
              actions: [{ type: "websocket_send", data: "welcome" }],
              max_uses: 1,
            },
            {
              id: "ack",
              from: "ack",
              to: "reconnect",
              trigger: {
                protocol: "websocket_message",
                path: "/ws",
                body: "ack",
              },
              actions: [{ type: "disconnect" }],
              max_uses: 1,
            },
            {
              id: "reconnect",
              from: "reconnect",
              to: "complete",
              trigger: { protocol: "websocket_connect", path: "/ws" },
              actions: [{ type: "websocket_send", data: "done" }],
              max_uses: 1,
            },
          ],
          max_transitions: 5,
        },
      },
    });
    try {
      const capture = await captureProcessScenario(scenario, {
        status: "enabled",
        executableRoots: [dirname(process.execPath)],
        workingRoots: [root],
        allowedEnvironment: [],
        networkAccess: "external",
      });
      expect(capture.ok).toBe(true);
      if (!capture.ok) throw capture.error;
      expect(
        capture.value.replay_transitions.map(
          ({ transition_id }) => transition_id,
        ),
      ).toEqual(["login", "api", "initial_socket", "ack", "reconnect"]);
      expect(capture.value.frames.map(({ data }) => data).join("")).toContain(
        "reply:done",
      );
      expect(JSON.stringify(capture.value)).not.toContain("machine-secret");
      expect(capture.value.manifest.replay_plan_sha256).toBe(
        digestProcessCommitment(scenario.replay),
      );
      expect(
        capture.value.event_journal?.filter(
          ({ collection }) => collection === "replay_transitions",
        ),
      ).toHaveLength(capture.value.replay_transitions.length);
      for (const [index, entry] of (
        capture.value.event_journal ?? []
      ).entries())
        if (entry.collection === "replay_transitions")
          expect(capture.value.event_journal?.[index - 1]?.collection).toBe(
            "protocol_events",
          );
    } finally {
      await rm(root, { recursive: true, force: true });
    }
  },
  20_000,
);

itWithCaptureCapability(
  "captures PTY, filesystem, descendants, HTTP replay, and redacts environment",
  async () => {
    const root = await createTestTempDirectory("rea-harness-test-");
    const script = join(root, "fixture.mjs");
    await writeFile(
      script,
      [
        'import { writeFile } from "node:fs/promises";',
        'import { spawn } from "node:child_process";',
        'await writeFile(new URL("result.txt", `file://${process.cwd()}/`), "created");',
        "const response = await fetch(`${process.env.REA_REPLAY_HTTP_URL}/probe`);",
        "console.log(`reply:${await response.text()}`);",
        "console.log(`sensitive:${process.env.SECRET}`);",
        'const child = spawn(process.execPath, ["-e", "setTimeout(()=>{}, 150)"], { stdio: "ignore" });',
        "await new Promise((resolve) => setTimeout(resolve, 80));",
        "child.kill();",
      ].join("\n"),
    );
    const policy: ProcessExecutionPolicy = {
      status: "enabled",
      executableRoots: [dirname(process.execPath)],
      workingRoots: [root],
      allowedEnvironment: ["SECRET"],
      networkAccess: "external",
    };
    const scenario = parseProcessScenario({
      executable: process.execPath,
      arguments: [script],
      working_directory: root,
      filesystem_roots: [root],
      environment: { SECRET: "do-not-record" },
      secret_aliases: ["SECRET"],
      replay: {
        http: [{ method: "GET", path: "/probe", status: 200, body: "ok" }],
      },
    });
    try {
      const capture = await captureProcessScenario(scenario, policy);
      expect(capture.ok).toBe(true);
      if (!capture.ok) throw capture.error;
      expect(
        capture.value.frames.map((frame) => frame.data).join(""),
      ).toContain("reply:ok");
      expect(
        capture.value.frames.map((frame) => frame.data).join(""),
      ).toContain("sensitive:<redacted>");
      expect(
        capture.value.files_after.some((file) =>
          file.path.endsWith("result.txt"),
        ),
      ).toBe(true);
      expect(
        capture.value.filesystem_effects.some(
          (effect) =>
            effect.path.endsWith(":result.txt") && effect.status === "created",
        ),
      ).toBe(true);
      expect(
        capture.value.protocol_events.some(
          (event) => event.protocol === "http" && event.path === "/probe",
        ),
      ).toBe(true);
      expect(JSON.stringify(capture.value)).not.toContain("do-not-record");
      expect(await readFile(join(root, "result.txt"), "utf8")).toBe("created");
      expect(JSON.stringify(capture.value.files_after)).not.toContain(root);
    } finally {
      await rm(root, { recursive: true, force: true });
    }
  },
);