rea/tests/boundary/cli/jsonLinesOutput.test.ts

import { Writable } from "node:stream";

import { Cli, z } from "incur";
import { estimateTokenCount } from "tokenx";
import { describe, expect, it } from "vitest";

import {
  createCliResultOutput,
  streamCliCommandResults,
} from "../../../src/cli/cliResultOutput.js";
import { jsonLinesErrorRecord } from "../../../src/cli/jsonLines.js";
import { parseCliOutputArguments } from "../../../src/cliOutput.js";
import { logCliCommand } from "../../../src/cliLogging.js";
import { silentLogger } from "../../../src/logger.js";

const capture = (args: readonly string[]) => {
  const chunks: Buffer[] = [];
  const destination = new Writable({
    write(chunk: Buffer, _encoding, callback) {
      chunks.push(chunk);
      callback();
    },
  });
  const output = createCliResultOutput(
    parseCliOutputArguments(args),
    destination,
  );
  if (output === undefined) throw new Error("Missing output owner");
  return {
    output,
    destination,
    text: () => Buffer.concat(chunks).toString("utf8"),
  };
};
const lines = (text: string): unknown[] =>
  text.trimEnd() === ""
    ? []
    : text
        .trimEnd()
        .split("\n")
        .map((line) => JSON.parse(line));
const metadata = { command: "fixture", duration: "1ms", format: "jsonl" };

describe("row-oriented CLI JSON Lines", () => {
  it.each(["rows", "items", "records", "occurrences"])(
    "preserves all context around %s without a field whitelist",
    async (key) => {
      const records = [
        { id: "one", index: 8, text: "first\n雪𝟠" },
        { id: "two", index: 9 },
      ];
      const context = {
        evidence_id: "ev:parent",
        subject: { path: "/local/subject" },
        provider: "actual-provider",
        authority: "static",
        confidence: "derived",
        locations: [{ path: "/source", line: 4 }],
        parameters: { field: "caller choice" },
        raw_result: null,
        evidence_links: ["ev:source"],
        limitations: ["partial source"],
        normalized_result: {
          kind: "inventory",
          parent_digest: "sha256:parent",
          view_digest: "sha256:view",
          artifact: { id: "art:one" },
          coverage: { complete: false, examined: 2, total: null },
          page: { offset: 8, has_more: true },
          [key]: records,
        },
      };
      const stream = capture(["--format", "jsonl"]);
      await stream.output.write(context, metadata);
      const { [key]: _records, ...viewContext } = context.normalized_result;
      expect(lines(stream.text())).toEqual([
        ...records,
        {
          $rea: "jsonl/v1",
          type: "metadata",
          records: { path: ["normalized_result", key], emitted: 2, source: 2 },
          context: { ...context, normalized_result: viewContext },
        },
      ]);
      expect(context.normalized_result[key]).toBe(records);
    },
  );

  it.each([[], [{ id: "row" }]].map((rows) => ({ rows })))(
    "preserves nested pagination without a coverage object (%j)",
    async ({ rows }) => {
      const stream = capture(["--format", "jsonl", "--full-output"]);
      const value = {
        page: { rows, offset: 12, total: 31, next_offset: 14 },
        limitations: ["more pages"],
      };
      await stream.output.write(value, metadata);
      expect(lines(stream.text())).toEqual([
        ...rows,
        {
          $rea: "jsonl/v1",
          type: "metadata",
          records: {
            path: ["data", "page", "rows"],
            emitted: rows.length,
            source: rows.length,
          },
          context: {
            ok: true,
            data: {
              page: { offset: 12, total: 31, next_offset: 14 },
              limitations: value.limitations,
            },
            meta: { command: "fixture", duration: "1ms" },
          },
        },
      ]);
    },
  );
});

describe("JSON Lines admission and selection", () => {
  it.each([
    null,
    {},
    "scalar",
    { rows: [], items: [] },
    { rows: [], normalized_result: { rows: [] } },
    Object.create({ rows: [{ inherited: true }] }),
    { rows: [{ $rea: "collision" }] },
  ])(
    "refuses an unclear or reserved boundary before writing rows: %j",
    async (value) => {
      const stream = capture(["--format", "jsonl", "--token-count"]);
      await stream.output.write(value, metadata);
      expect(lines(stream.text())).toEqual([
        jsonLinesErrorRecord(stream.output.error),
      ]);
      expect(stream.output.error?.code).toBe("unreadable_output");
    },
  );

  it.each(["missing", "count", "nested.value"])(
    "does not admit a scalar/missing selection merely because a filter exists: %s",
    async (filter) => {
      const stream = capture(["--format", "jsonl", "--filter-output", filter]);
      await stream.output.write(
        { rows: [{ id: 1 }], count: 0, nested: { value: null } },
        metadata,
      );
      expect(lines(stream.text())).toHaveLength(1);
      expect(lines(stream.text())[0]).toMatchObject({
        $rea: "jsonl/v1",
        type: "error",
        result: { code: "unreadable_output" },
      });
    },
  );

  it("records output selection separately from source coverage and keeps provenance", async () => {
    const value = {
      evidence_id: "ev:parent",
      raw_result: { excluded: "large observation" },
      provider: "source-provider",
      normalized_result: {
        rows: [
          { id: "a", text: "first" },
          { id: "b", text: "second" },
          { id: "c", text: "third" },
        ],
        coverage: { complete: true, examined: 3 },
      },
    };
    const filter = "normalized_result.rows[1,3].id";
    const stream = capture(["--format", "jsonl", "--filter-output", filter]);
    await stream.output.write(value, metadata);
    expect(lines(stream.text())).toEqual([
      { id: "b" },
      { id: "c" },
      {
        $rea: "jsonl/v1",
        type: "metadata",
        records: {
          path: ["normalized_result", "rows"],
          emitted: 2,
          source: 3,
          filter,
        },
        context: {
          evidence_id: "ev:parent",
          provider: "source-provider",
          normalized_result: { coverage: { complete: true, examined: 3 } },
        },
      },
    ]);
  });

  it("disambiguates selected collections without restoring excluded peer rows", async () => {
    const stream = capture([
      "--format",
      "jsonl",
      "--filter-output",
      "normalized_result.rows",
    ]);
    await stream.output.write(
      {
        evidence_id: "ev:parent",
        raw_result: null,
        normalized_result: {
          rows: [{ id: 1 }],
          items: [{ excluded: "peer" }],
          coverage: { status: "unknown" },
        },
      },
      metadata,
    );
    expect(lines(stream.text())).toEqual([
      { id: 1 },
      {
        $rea: "jsonl/v1",
        type: "metadata",
        records: {
          path: ["normalized_result", "rows"],
          emitted: 1,
          source: 1,
          filter: "normalized_result.rows",
        },
        context: {
          evidence_id: "ev:parent",
          normalized_result: { coverage: { status: "unknown" } },
        },
      },
    ]);
  });
});

describe("JSON Lines counting and transport", () => {
  it.each([false, true])(
    "frames classified failures without filtering/counting them away (full: %s)",
    async (full) => {
      const stream = capture([
        "--format",
        "jsonl",
        "--token-count",
        "--filter-output",
        "missing",
        ...(full ? ["--full-output"] : []),
      ]);
      const failure = {
        healthy: false,
        checks: [{ ok: false, name: "source" }],
        diagnostics: { path: "/caller/path" },
      };
      await stream.output.write(failure, metadata);
      expect(lines(stream.text())).toEqual([
        jsonLinesErrorRecord(
          full
            ? {
                ok: false,
                error: failure,
                meta: { command: "fixture", duration: "1ms" },
              }
            : failure,
        ),
      ]);
    },
  );

  it.each([false, true])(
    "counts array-valued rows as single records (object collection: %s)",
    async (wrapped) => {
      const rows = [[false, null], [0], [], [{ nested: ["雪", "a\nb"] }]];
      const value = wrapped ? { rows, coverage: { complete: false } } : rows;
      const args = ["--format", "jsonl"];
      const stream = capture(args);
      await stream.output.write(value, metadata);
      const counter = capture([...args, "--token-count"]);
      await counter.output.write(value, metadata);
      expect(counter.text()).toBe(`${estimateTokenCount(stream.text())}\n`);
    },
  );

  it("counts exactly the emitted representation, including escaped multiline text and final context", async () => {
    const value = {
      rows: [{ text: "雪𝟠\nline".repeat(4096) }, { text: 'escaped"\\\t' }],
      coverage: { complete: false, reasons: ["unknown remainder"] },
    };
    const args = ["--format", "jsonl", "--full-output"];
    const stream = capture(args);
    await stream.output.write(value, metadata);
    const counter = capture([...args, "--token-count"]);
    await counter.output.write(value, metadata);
    expect(counter.text()).toBe(`${estimateTokenCount(stream.text())}\n`);
  });

  it("bounds a large escaped field and observes destination backpressure through the final metadata", async () => {
    const chunks: Buffer[] = [];
    let maxBytes = 0;
    const destination = new Writable({
      highWaterMark: 1,
      write(chunk: Buffer, _encoding, callback) {
        maxBytes = Math.max(maxBytes, chunk.length);
        setTimeout(() => {
          chunks.push(chunk);
          callback();
        }, 1);
      },
    });
    const output = createCliResultOutput(
      parseCliOutputArguments(["--format", "jsonl"]),
      destination,
    )!;
    const rows = [{ id: "large", text: "\u0000雪𝟠".repeat(32 * 1024) }];
    await output.write({ rows, coverage: { complete: false } }, metadata);
    expect(maxBytes).toBeLessThan(128 * 1024);
    const records = lines(Buffer.concat(chunks).toString("utf8"));
    expect(records[0]).toEqual(rows[0]);
    expect(records[1]).toMatchObject({
      type: "metadata",
      context: { coverage: { complete: false } },
    });
    expect(destination.writableEnded).toBe(false);
  });

  it("retains an explicitly successful error-shaped row as data", async () => {
    const value = [
      {
        code: "not-an-error",
        message: "ordinary row",
        ok: false,
        healthy: false,
      },
    ];
    const stream = capture(["--format", "jsonl"]);
    const cli = Cli.create("explicit-success", { sync: false }).command(
      "rows",
      {
        run: () =>
          logCliCommand(
            silentLogger,
            "rows",
            async () => value,
            () => false,
          ),
      },
    );
    streamCliCommandResults(cli, stream.output);
    await cli.serve(["rows", "--format", "jsonl"], { stdout: () => undefined });
    expect(lines(stream.text())).toEqual(value);
  });
});

describe("SDK-owned JSON Lines outcomes", () => {
  it("uses SDK status to frame validation errors and resets ownership between invocations", async () => {
    const stream = capture(["--format", "jsonl"]);
    const cli = Cli.create("sdk-status", { sync: false }).command("rows", {
      args: z.object({ required: z.string() }),
      run: () => [{ id: "row" }],
    });
    streamCliCommandResults(cli, stream.output);
    const serve = async (args: string[]) => {
      let status = 0;
      stream.output.begin(parseCliOutputArguments(args));
      await cli.serve(args, {
        stdout: (text) => {
          if (!stream.output.handled && !stream.output.captureSdkOutput(text))
            throw new Error("Unexpected delegation");
        },
        exit: (code) => {
          status = code;
        },
      });
      await stream.output.finishSdkOutput(status);
      return status;
    };
    expect(await serve(["rows", "--format", "jsonl"])).toBe(1);
    expect(lines(stream.text())[0]).toMatchObject({
      $rea: "jsonl/v1",
      type: "error",
      result: { code: "VALIDATION_ERROR" },
    });
    expect(await serve(["rows", "present", "--format", "jsonl"])).toBe(0);
    expect(lines(stream.text())[1]).toEqual({ id: "row" });
    expect(stream.output.error).toBeUndefined();
  });

  it.each([false, true])(
    "preserves sentinel data and frames sentinel failure using SDK outcome (%s)",
    async (failure) => {
      const stream = capture(["--format", "jsonl"]);
      const cli = Cli.create("sdk-sentinel", { sync: false }).command(
        "result",
        {
          run: (context) =>
            failure
              ? context.error({
                  code: "EXPLICIT_FAILURE",
                  message: "Caller-owned error",
                })
              : context.ok({
                  code: "EXPLICIT_FAILURE",
                  message: "Successful data",
                }),
        },
      );
      streamCliCommandResults(cli, stream.output);
      let status = 0;
      await cli.serve(["result", "--format", "jsonl"], {
        stdout: (text) => {
          stream.output.captureSdkOutput(text);
        },
        exit: (code) => {
          status = code;
        },
      });
      expect(stream.text()).toBe("");
      await stream.output.finishSdkOutput(status);
      expect(lines(stream.text())[0]).toEqual(
        failure
          ? jsonLinesErrorRecord({
              code: "EXPLICIT_FAILURE",
              message: "Caller-owned error",
            })
          : { code: "EXPLICIT_FAILURE", message: "Successful data" },
      );
    },
  );

  it("delegates generator events immediately without retaining the stream", async () => {
    const stream = capture(["--format", "jsonl"]);
    const delegated: string[] = [];
    const cli = Cli.create("sdk-stream", { sync: false }).command("events", {
      run: async function* () {
        yield { text: "first" };
        expect(delegated.length).toBeGreaterThan(0);
        yield { text: "second" };
      },
    });
    streamCliCommandResults(cli, stream.output);
    await cli.serve(["events", "--format", "jsonl"], {
      stdout: (text) => {
        expect(stream.output.captureSdkOutput(text)).toBe(false);
        delegated.push(text);
      },
    });
    await stream.output.finishSdkOutput(0);
    expect(stream.text()).toBe("");
    expect(delegated.join("")).toContain("first");
    expect(delegated.join("")).toContain("second");
  });
});