feat(cli): accept stdin for structured JSON request documents (#5486) (#7040)

* feat(cli): accept stdin for structured JSON request documents (#5486)

* fix(cli): preserve cancellation and admission for JSON stdin

---------

Co-authored-by: morluto <76467478+morluto@users.noreply.github.com>

src/application/Utf8JsonInput.ts

@@ -81,6 +81,7 @@ export const parseUtf8Json = (
     return {
       ok: false,
       error: "JSON input is not valid UTF-8",
+      userMessage: "JSON input is not valid UTF-8",
       cause,
     };
   }

src/cli/electronCommands.ts

@@ -121,6 +121,7 @@ const registerElectronActiveCommand = (
             const input = await parseCliJsonInput(
               args.inputJson,
               "capture_electron_scenario",
+              signal,
             );
             if (!input.ok) return input.error;
             const parsed = electronActiveObservationInputSchema.safeParse(

src/cli/wasmCommands.ts

@@ -38,6 +38,7 @@ export const registerWasmCommands = (
           const input = await parseCliJsonInput(
             args.input,
             "compare_wasm_interfaces",
+            signal,
           );
           if (!input.ok) return input.error;
           const result = await compareWasmInterfaces(input.value, { signal });

src/cliBrowserScenarioCommands.ts

@@ -41,7 +41,11 @@ export const registerBrowserScenarioCommands = (
     run: ({ args }) =>
       withCommandCancellation((signal) =>
         logCliCommand(logger, CLI_COMMANDS.captureBrowserScenario, async () => {
-          const input = await parseCliJsonInput(args.inputJson, OPERATION);
+          const input = await parseCliJsonInput(
+            args.inputJson,
+            OPERATION,
+            signal,
+          );
           if (!input.ok) return input.error;
           const scenario = browserScenarioSchema.safeParse(
             resolveCliJsonPaths(input.value, [["browser", "executable_path"]]),

src/cliJsonInput.ts

@@ -1,3 +1,7 @@
+import { addAbortListener } from "node:events";
+import type { Readable } from "node:stream";
+import { getHeapStatistics } from "node:v8";
+
 import { absoluteCallerPath } from "./filesystem/callerPath.js";
 
 import { sealOwnedJsonInput } from "./application/OwnedJsonInput.js";
@@ -5,23 +9,34 @@ import {
   JsonInputCancellationFailure,
   JsonInputPhysicalReadFailure,
   readJsonInputFile,
+  nativeJsonParseRequiredBytes,
 } from "./application/JsonInputFile.js";
 import { NonRegularFileReadError } from "./filesystem/RegularFile.js";
 import {
   AnalysisAccessDeniedError,
   AnalysisCancelledError,
   AnalysisInputError,
+  AnalysisResourceConstraintError,
 } from "./domain/analysisErrorCore.js";
 import { projectAnalysisError } from "./domain/analysisErrorProjection.js";
+import { err, ok, type Result } from "./domain/result.js";
 import type { JsonValue } from "./domain/jsonValue.js";
 import { safeParseJson } from "./domain/safeJson.js";
+import {
+  JSON_INPUT_RESOURCE_REMEDIATION,
+  parseUtf8Json,
+} from "./application/Utf8JsonInput.js";
 
-/** Parse inline JSON or one local JSON file, honoring optional caller cancellation. */
+/** Parse inline JSON, standard input ('-'), or one local JSON file, honoring optional caller cancellation. */
 export const parseCliJsonInput = async (
   value: string,
   operation: string,
   signal?: AbortSignal,
-  options?: { readonly snapshot?: boolean },
+  options?: {
+    readonly snapshot?: boolean;
+    readonly stdin?: Readable;
+    readonly maxStdinBytes?: number;
+  },
 ): Promise<
   | { readonly ok: true; readonly value: unknown }
   | { readonly ok: false; readonly error: JsonValue }
@@ -53,13 +68,16 @@ export const parseCliJsonInput = async (
             },
           };
     }
-    const parsed = await readJsonInputFile(
-      value,
-      operation,
-      signal,
-      undefined,
-      options?.snapshot ?? false,
-    );
+    const parsed =
+      value === "-"
+        ? await readCliJsonStdin(operation, signal, options)
+        : await readJsonInputFile(
+            value,
+            operation,
+            signal,
+            undefined,
+            options?.snapshot ?? false,
+          );
     if (parsed.ok && signal?.aborted) return cancelled(signal.reason);
     return parsed.ok
       ? { ok: true, value: parsed.value }
@@ -106,6 +124,126 @@ export const parseCliJsonInput = async (
   }
 };
 
+const DEFAULT_MAX_STDIN_BYTES = 8 * 1024 * 1024;
+
+const readCliJsonStdin = async (
+  operation: string,
+  signal?: AbortSignal,
+  options?: {
+    readonly snapshot?: boolean;
+    readonly stdin?: Readable;
+    readonly maxStdinBytes?: number;
+  },
+): Promise<
+  Result<unknown, AnalysisInputError | AnalysisResourceConstraintError>
+> => {
+  const stream = options?.stdin ?? process.stdin;
+  const maxBytes = options?.maxStdinBytes ?? DEFAULT_MAX_STDIN_BYTES;
+  const chunks: Buffer[] = [];
+  let totalBytes = 0;
+  const requireHeadroom = (): void => {
+    const heap = getHeapStatistics();
+    const requiredBytes = nativeJsonParseRequiredBytes(
+      totalBytes,
+      options?.snapshot,
+    );
+    if (requiredBytes > heap.total_available_size)
+      throw new AnalysisResourceConstraintError(
+        operation,
+        "memory",
+        "Insufficient heap headroom to read and parse the complete JSON stdin document",
+        {
+          input_path: "-",
+          input_bytes: totalBytes,
+          required_heap_bytes: requiredBytes,
+          available_heap_bytes: heap.total_available_size,
+          heap_size_limit_bytes: heap.heap_size_limit,
+        },
+        { remediationAction: JSON_INPUT_RESOURCE_REMEDIATION },
+      );
+  };
+  // Destroying the reader wakes an idle next(); polling inside the loop cannot.
+  const abort =
+    signal === undefined
+      ? undefined
+      : addAbortListener(signal, () => {
+          stream.destroy(new JsonInputCancellationFailure(signal.reason));
+        });
+  try {
+    try {
+      for await (const chunk of stream) {
+        signal?.throwIfAborted();
+        totalBytes +=
+          typeof chunk === "string"
+            ? Buffer.byteLength(chunk)
+            : chunk.byteLength;
+        if (totalBytes > maxBytes)
+          return err(
+            new AnalysisResourceConstraintError(
+              operation,
+              "memory",
+              "JSON input exceeds the maximum allowed stdin size",
+              {
+                input_path: "-",
+                input_bytes: totalBytes,
+                max_stdin_bytes: maxBytes,
+              },
+              { remediationAction: JSON_INPUT_RESOURCE_REMEDIATION },
+            ),
+          );
+        requireHeadroom();
+        chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
+      }
+    } catch (cause: unknown) {
+      if (
+        cause instanceof JsonInputCancellationFailure ||
+        (signal?.aborted && isJsonInputCancellation(cause, signal))
+      )
+        throw cause;
+      if (cause instanceof AnalysisResourceConstraintError) return err(cause);
+      throw new JsonInputPhysicalReadFailure(cause);
+    } finally {
+      abort?.[Symbol.dispose]();
+    }
+    signal?.throwIfAborted();
+    // Recheck live headroom before concatenation, UTF-8 decoding and JSON.parse.
+    requireHeadroom();
+    const parsed = parseUtf8Json(
+      Buffer.concat(chunks, totalBytes),
+      operation,
+      "-",
+    );
+    if (!parsed.ok)
+      return err(
+        new AnalysisInputError(
+          operation,
+          {
+            cause: parsed.cause,
+            ...(parsed.userMessage === undefined
+              ? {}
+              : { userMessage: parsed.userMessage }),
+          },
+          [
+            {
+              path: [],
+              reason: "invalid_format",
+              expected: "JSON",
+              ...(parsed.userMessage === undefined
+                ? {}
+                : { message: parsed.userMessage }),
+            },
+          ],
+        ),
+      );
+    return options?.snapshot
+      ? await sealOwnedJsonInput(parsed.value, operation, signal)
+      : ok(parsed.value);
+  } catch (cause: unknown) {
+    if (cause instanceof AnalysisResourceConstraintError) return err(cause);
+    throw cause;
+  }
+};
+
 const isJsonInputCancellation = (
   cause: unknown,
   signal: AbortSignal,

tests/boundary/cli/cliJsonStdinCli.test.ts

@@ -0,0 +1,168 @@
+import { resolve } from "node:path";
+import { pathToFileURL } from "node:url";
+
+import { expect } from "vitest";
+
+import { nativeJsonParseRequiredBytes } from "../../../src/application/JsonInputFile.js";
+import { JSON_BYTE_ORDER_MARK_MESSAGE } from "../../../src/application/Utf8JsonInput.js";
+import { cliTest } from "../../support/cli/cliFixture.js";
+import { waitForExit } from "../../support/process/processFixture.js";
+
+const probe = pathToFileURL(
+  resolve("tests/fixtures/cliJsonStdinProbe.mjs"),
+).href;
+
+cliTest.for([
+  [Buffer.from("\uFEFF{}"), JSON_BYTE_ORDER_MARK_MESSAGE],
+  [Buffer.from([0xf0, 0x28, 0x8c, 0xbc]), "JSON input is not valid UTF-8"],
+  [Buffer.from('{"secret-value":'), undefined],
+] as const)(
+  "projects piped JSON failures through the public CLI (%j)",
+  async ([input, message], { cli }) => {
+    const result = await cli.run({
+      arguments: ["compare-wasm-interfaces", "-", "--json"],
+      input,
+    });
+    expect(result.exitCode, result.stderr).toBe(1);
+    expect(result.json).toMatchObject({
+      code: "invalid_request",
+      input_path: "-",
+      input_reason: "invalid-json",
+      ...(message === undefined ? {} : { details: { issues: [{ message }] } }),
+    });
+    expect(result.stdout).not.toContain("secret-value");
+  },
+);
+
+cliTest.for([false, true])(
+  "admits piped JSON against real heap headroom (snapshot: %s)",
+  async (snapshot, { processes }) => {
+    const child = processes.spawn(
+      process.execPath,
+      [
+        resolve("scripts/rea.mjs"),
+        ...(snapshot
+          ? ["environment-serve", "-", "--evidence", "{}"]
+          : ["compare-wasm-interfaces", "-"]),
+        "--json",
+      ],
+      {
+        env: {
+          ...process.env,
+          NODE_OPTIONS: "--max-old-space-size=256",
+          REA_LOG_LEVEL: "silent",
+        },
+      },
+    );
+    let stdout = "",
+      stderr = "";
+    let writeFailure: unknown;
+    child.stdout.on("data", (chunk) => {
+      stdout += String(chunk);
+    });
+    child.stderr.on("data", (chunk) => {
+      stderr += String(chunk);
+    });
+    // Admission can close the reader before the producer finishes its write.
+    child.stdin.on("error", (cause) => {
+      writeFailure = cause;
+    });
+    const closed = new Promise<void>((resolve, reject) => {
+      child.once("error", reject);
+      child.once("close", () => resolve());
+    });
+    child.stdin.end(`[${"[],".repeat(1_400_000)}[]]`);
+    expect(await waitForExit(child, 10_000), stderr).toBe(true);
+    await closed;
+    if (writeFailure !== undefined)
+      expect(writeFailure).toMatchObject({ code: "EPIPE" });
+    const result = {
+      exitCode: child.exitCode,
+      stderr,
+      json: JSON.parse(stdout) as unknown,
+    };
+    expect(result.exitCode, result.stderr).toBe(1);
+    expect(result.json).toMatchObject({
+      code: "resource_constraint",
+      input_path: "-",
+      input_reason: "too-large",
+      details: {
+        resource: "memory",
+        reported_limits: {
+          input_path: "-",
+          required_heap_bytes: expect.any(Number),
+          available_heap_bytes: expect.any(Number),
+        },
+      },
+    });
+    const limits = (
+      result.json as {
+        details: {
+          reported_limits: {
+            input_bytes: number;
+            required_heap_bytes: number;
+            available_heap_bytes: number;
+          };
+        };
+      }
+    ).details.reported_limits;
+    expect(limits.input_bytes).toBeLessThanOrEqual(8 * 1024 * 1024);
+    expect(limits.required_heap_bytes).toBe(
+      nativeJsonParseRequiredBytes(limits.input_bytes, snapshot),
+    );
+    expect(limits.required_heap_bytes).toBeGreaterThan(
+      limits.available_heap_bytes,
+    );
+  },
+);
+
+cliTest.skipIf(process.platform === "win32").for(
+  [
+    "capture-browser-scenario",
+    "capture-electron-scenario",
+    "compare-wasm-interfaces",
+  ].flatMap((command) =>
+    (
+      [
+        ["SIGINT", 130],
+        ["SIGTERM", 143],
+      ] as const
+    ).map(([signal, exitCode]) => ({ command, signal, exitCode })),
+  ),
+)(
+  "$command cancels an idle stdin reader on $signal",
+  async ({ command, signal, exitCode }, { processes }) => {
+    const child = processes.spawn(
+      process.execPath,
+      [resolve("scripts/rea.mjs"), command, "-", "--json"],
+      {
+        env: {
+          ...process.env,
+          NODE_OPTIONS: `--import=${probe}`,
+          REA_LOG_LEVEL: "silent",
+        },
+      },
+    );
+    let stdout = "",
+      stderr = "";
+    child.stdout.on("data", (chunk) => {
+      stdout += String(chunk);
+    });
+    child.stderr.on("data", (chunk) => {
+      stderr += String(chunk);
+    });
+    const closed = new Promise<void>((resolve, reject) => {
+      child.once("error", reject);
+      child.once("close", () => resolve());
+    });
+    await expect
+      .poll(() => stderr.includes("REA_TEST_STDIN_READ"), { timeout: 10_000 })
+      .toBe(true);
+    // Keep the pipe open: EOF must not be what releases the pending read.
+    child.kill(signal);
+    expect(await waitForExit(child, 5_000), stderr).toBe(true);
+    await closed;
+    expect(child.exitCode, stderr).toBe(exitCode);
+    expect(JSON.parse(stdout)).toMatchObject({ code: "cancelled" });
+  },
+);

tests/boundary/filesystem/cliJsonInput.test.ts

@@ -1,6 +1,8 @@
+import { getEventListeners } from "node:events";
 import { execFile } from "node:child_process";
 import { chmod, symlink, writeFile } from "node:fs/promises";
 import { join } from "node:path";
+import { Readable } from "node:stream";
 import { promisify } from "node:util";
 
 import { describe, expect, it, onTestFinished } from "vitest";
@@ -346,3 +348,211 @@ describe("CLI JSON file cancellation", () => {
     ).toEqual({ ok: true, value: ["index.js", "main.js", "package.json"] });
   });
 });
+
+describe("CLI JSON standard input", () => {
+  it("reads valid JSON from stdin when input is '-'", async () => {
+    const stream = Readable.from(['{"seed":123,"label":"piped"}']);
+    const result = await parseCliJsonInput("-", "test-input", undefined, {
+      stdin: stream,
+    });
+    expect(result).toEqual({
+      ok: true,
+      value: { seed: 123, label: "piped" },
+    });
+  });
+
+  it("rejects empty stdin with invalid-json", async () => {
+    const stream = Readable.from([]);
+    const result = await parseCliJsonInput("-", "test-input", undefined, {
+      stdin: stream,
+    });
+    expect(result).toMatchObject({
+      ok: false,
+      error: {
+        code: "invalid_request",
+        input_path: "-",
+        input_reason: "invalid-json",
+        details: {
+          issues: [{ path: [], reason: "invalid_format" }],
+        },
+      },
+    });
+  });
+
+  it("rejects malformed JSON from stdin without printing full payload", async () => {
+    const stream = Readable.from(["{ this is not valid json"]);
+    const result = await parseCliJsonInput("-", "test-input", undefined, {
+      stdin: stream,
+    });
+    expect(result).toMatchObject({
+      ok: false,
+      error: {
+        code: "invalid_request",
+        input_path: "-",
+        input_reason: "invalid-json",
+        details: {
+          issues: [{ path: [], reason: "invalid_format" }],
+        },
+      },
+    });
+    expect(JSON.stringify(result)).not.toContain("this is not valid json");
+  });
+
+  it("rejects oversized stdin with too-large", async () => {
+    const stream = Readable.from(['{"a":"' + "x".repeat(100) + '"}']);
+    const result = await parseCliJsonInput("-", "test-input", undefined, {
+      stdin: stream,
+      maxStdinBytes: 50,
+    });
+    expect(result).toMatchObject({
+      ok: false,
+      error: {
+        code: "resource_constraint",
+        input_path: "-",
+        input_reason: "too-large",
+        details: {
+          resource: "memory",
+        },
+      },
+    });
+  });
+
+  it("preserves caller cancellation during stdin read", async () => {
+    const controller = new AbortController();
+    controller.abort("aborted");
+    const stream = Readable.from(['{"seed":1}']);
+    const result = await parseCliJsonInput(
+      "-",
+      "test-input",
+      controller.signal,
+      {
+        stdin: stream,
+      },
+    );
+    expect(result).toMatchObject({
+      ok: false,
+      error: {
+        code: "cancelled",
+      },
+    });
+  });
+});
+
+describe("CLI JSON stdin cancellation and encoding", () => {
+  it.each([false, true])(
+    "interrupts an idle stdin read (partial input: %s)",
+    async (partial) => {
+      const controller = new AbortController();
+      let started!: () => void;
+      const reading = new Promise<void>((resolve) => {
+        started = resolve;
+      });
+      const stream = new Readable({
+        read() {
+          started();
+        },
+      });
+      onTestFinished(() => {
+        stream.destroy();
+      });
+      if (partial) stream.push('{"partial":');
+      const pending = parseCliJsonInput("-", "test-input", controller.signal, {
+        stdin: stream,
+      });
+      await reading;
+      // Let any initial chunk be consumed before aborting the next idle read.
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      controller.abort("stop idle stdin");
+      await expect(pending).resolves.toMatchObject({
+        ok: false,
+        error: { code: "cancelled" },
+      });
+      expect(stream.destroyed).toBe(true);
+      expect(getEventListeners(controller.signal, "abort")).toHaveLength(0);
+    },
+    1_000,
+  );
+
+  it.each([
+    [Buffer.from("\uFEFF{}"), JSON_BYTE_ORDER_MARK_MESSAGE],
+    [Buffer.from([0xf0, 0x28, 0x8c, 0xbc]), "JSON input is not valid UTF-8"],
+    [Buffer.from([0xe2, 0x82]), "JSON input is not valid UTF-8"],
+  ])(
+    "preserves safe stdin encoding diagnostics (%j)",
+    async (bytes, message) => {
+      expect(
+        await parseCliJsonInput("-", "test-input", undefined, {
+          stdin: Readable.from([bytes]),
+        }),
+      ).toMatchObject({
+        ok: false,
+        error: {
+          input_path: "-",
+          input_reason: "invalid-json",
+          details: { issues: [{ message }] },
+        },
+      });
+    },
+  );
+
+  it("reads UTF-8 split between chunks and accepts the exact byte limit", async () => {
+    const bytes = Buffer.from('{"label":"中文🦀"}');
+    expect(
+      await parseCliJsonInput("-", "test-input", undefined, {
+        stdin: Readable.from([...bytes].map((byte) => Buffer.from([byte]))),
+        maxStdinBytes: bytes.length,
+      }),
+    ).toEqual({ ok: true, value: { label: "中文🦀" } });
+  });
+
+  it("projects cancellation during stdin snapshot sealing", async () => {
+    const controller = new AbortController();
+    const stream = Readable.from([
+      JSON.stringify(Array.from({ length: 20_000 }, (_, index) => ({ index }))),
+    ]);
+    stream.once("end", () =>
+      setImmediate(() => controller.abort("stop sealing")),
+    );
+    await expect(
+      parseCliJsonInput("-", "test-input", controller.signal, {
+        stdin: stream,
+        snapshot: true,
+      }),
+    ).resolves.toMatchObject({ ok: false, error: { code: "cancelled" } });
+    expect(getEventListeners(controller.signal, "abort")).toHaveLength(0);
+  });
+
+  it("seals stdin snapshots and releases the abort listener on success", async () => {
+    const controller = new AbortController();
+    const result = await parseCliJsonInput(
+      "-",
+      "test-input",
+      controller.signal,
+      {
+        stdin: Readable.from(['{"nested":{"value":1}}']),
+        snapshot: true,
+      },
+    );
+    expect(result).toEqual({ ok: true, value: { nested: { value: 1 } } });
+    if (result.ok) expect(Object.isFrozen(result.value)).toBe(true);
+    expect(getEventListeners(controller.signal, "abort")).toHaveLength(0);
+  });
+
+  it("preserves stdin physical failure and releases the abort listener", async () => {
+    const controller = new AbortController();
+    const stream = new Readable({
+      read() {
+        this.destroy(Object.assign(new Error("read failed"), { code: "EIO" }));
+      },
+    });
+    expect(
+      await parseCliJsonInput("-", "test-input", controller.signal, {
+        stdin: stream,
+      }),
+    ).toMatchObject({
+      ok: false,
+      error: { input_path: "-", input_reason: "read-failed" },
+    });
+    expect(getEventListeners(controller.signal, "abort")).toHaveLength(0);
+  });
+});

tests/fixtures/cliJsonStdinProbe.mjs

@@ -0,0 +1,6 @@
+// Observe the production stdin reader without supplying data or closing its pipe.
+const iterate = process.stdin[Symbol.asyncIterator];
+process.stdin[Symbol.asyncIterator] = function (...args) {
+  process.stderr.write("REA_TEST_STDIN_READ\n");
+  return iterate.apply(this, args);
+};