mirror of
https://github.com/morluto/rea
synced 2026-10-25 22:08:11 -04:00
feat(cli)!: export row-oriented results as JSON Lines (#5629)
* feat(cli): add JSON Lines output for row-oriented results (#5487) - Support opt-in JSON Lines format (--format jsonl) for row-oriented results and collection views with clear record boundaries - Preserve stable source identities and stream records deterministically line-by-line - Emit final coverage and limitation metadata as a distinct line with https://rea.dev/schemas/jsonl-coverage-v1.json to prevent confusion with data rows - Distinguish error outputs with https://rea.dev/schemas/jsonl-error-v1.json - Reject non-row-oriented results that lack record boundaries with an unreadable_output error - Document JSON Lines streaming conventions in docs/cli.md - Add boundary tests covering empty results, multiline strings, partial coverage, large fields, and non-row rejection * fix(cli): preserve diagnostics, provenance, and nested collections in JSONL * fix(cli): frame classified failures and reject unrepresented JSONL facets --------- Co-authored-by: gensokyo-dev <342580529+gensokyo-dev@users.noreply.github.com>
This commit is contained in:
+90
@@ -46,6 +46,96 @@ module page. See
|
||||
[JavaScript application workflows](javascript-application-workflows.md) for
|
||||
views, feature traces and comparisons that take the same saved Evidence.
|
||||
|
||||
## JSON Lines record output
|
||||
|
||||
Use `--format jsonl` to consume row-oriented results one record per line:
|
||||
|
||||
```bash
|
||||
cat > observations-view.json <<'JSON'
|
||||
{"source":{"kind":"evidence-file","path":"app-evidence.json"},"view":{"kind":"javascript-facts","property_keys":[],"offset":0,"limit":32}}
|
||||
JSON
|
||||
rea inspect-analysis-view observations-view.json --format jsonl > observations.jsonl
|
||||
```
|
||||
|
||||
This request selects up to 32 observation identities and their provenance,
|
||||
without observation property values.
|
||||
|
||||
This changes JSONL object-collection framing. Scripts that need one complete
|
||||
report or a scalar field should use `--format json`. For a full application
|
||||
graph, first project a supported analysis view rather than selecting an
|
||||
arbitrary nested graph array.
|
||||
|
||||
A top-level array emits its elements in source order. An object must have one
|
||||
clear `rows`, `items`, `records`, or `occurrences` array, directly or under
|
||||
`page`. Evidence uses the same boundaries inside `normalized_result`.
|
||||
An arbitrary report or a scalar selection is refused with an actionable error;
|
||||
use `--format json` for a complete document. If several collections exist,
|
||||
select one with `--filter-output`, for example `normalized_result.rows`.
|
||||
|
||||
Object collections finish with a control record:
|
||||
|
||||
```json
|
||||
{
|
||||
"$rea": "jsonl/v1",
|
||||
"type": "metadata",
|
||||
"records": {
|
||||
"path": ["normalized_result", "rows"],
|
||||
"emitted": 2,
|
||||
"source": 10,
|
||||
"filter": "normalized_result.rows[0,2]"
|
||||
},
|
||||
"context": {
|
||||
"evidence_id": "ev_…",
|
||||
"normalized_result": {
|
||||
"coverage": {
|
||||
"status": "page",
|
||||
"examined": 10,
|
||||
"total": 100,
|
||||
"next_offset": 10,
|
||||
"exhausted": false
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
This abbreviated example illustrates framing. The actual `context` preserves
|
||||
all non-row fields, including Evidence subject, provider, authority, confidence,
|
||||
locations, links, limitations, parent/view digests, pagination and coverage.
|
||||
With `--full-output`, it also retains the command envelope and its metadata.
|
||||
A data filter selects row fields/slices while this context remains available;
|
||||
excluded peer record collections and unselected Evidence `raw_result` are not
|
||||
restored. An explicitly selected `raw_result` remains selected. Retain the
|
||||
original Evidence for complete observations.
|
||||
|
||||
`records.emitted` and `records.source` count the selected output and its source
|
||||
array; they do not assert analysis completeness. Source coverage and unknowns
|
||||
retain their original meaning. Existing row identities and indices remain
|
||||
unchanged unless your field selection excludes them. Empty object collections
|
||||
still emit metadata; an empty top-level array emits no lines. Bare top-level
|
||||
arrays have no synthetic coverage record.
|
||||
|
||||
Command and output-validation failures emit `{"$rea":"jsonl/v1","type":"error","ok":false,"result":…}`
|
||||
and retain the original diagnostic payload. SDK validation/sentinel errors use
|
||||
the SDK's exit status rather than guessing from data fields. `$rea` is reserved
|
||||
at the top level of a data record, so control records are unambiguous; collisions
|
||||
are refused before rows are written. Nested `$rea` values are ordinary data.
|
||||
A missing final metadata record or nonzero exit can indicate an interrupted
|
||||
export; do not treat already emitted rows as a complete result.
|
||||
|
||||
`--token-count` estimates this exact representation, including metadata, one
|
||||
line at a time. Errors remain error records even with token counting enabled.
|
||||
Token windows cannot preserve JSON Lines and are rejected. JSON encoding and
|
||||
stdout writes are incremental, including large escaped strings; this is output
|
||||
streaming over a completed result, not incremental analysis. The analysis still
|
||||
needs its normal resources. A single record must fit the runtime string limit
|
||||
for token estimation; remove `--token-count` to stream a larger record.
|
||||
|
||||
Standalone version and MCP doctor diagnostics remain single JSON documents
|
||||
in JSON Lines mode. Incur's streaming generator events and successful sentinel
|
||||
responses retain their SDK protocol. MCP application results and Evidence
|
||||
contracts are unchanged.
|
||||
|
||||
## Shell completion
|
||||
|
||||
The CLI includes optional shell completion. Run `rea completions --help` to
|
||||
|
||||
+4
-2
@@ -1773,8 +1773,10 @@ the running Node engine's single-string limit. Shared input leaves keep the
|
||||
fixture's graph small; the verifier writes one temporary output file, checks its
|
||||
complete byte count and an independent digest, then removes it. It requires only
|
||||
Node and the built REA runtime, with space for the output plus a 1 GiB reserve.
|
||||
Run `npm run verify:javascript:output -- jsonl` for compact JSONL coverage. This
|
||||
opt-in lane is separate from routine tests and the canonical hash check
|
||||
Run `npm run verify:javascript:output -- jsonl` for compact JSONL coverage. Its
|
||||
row collection must match independent row and final-metadata byte/digest goldens;
|
||||
its token count is checked against the actual file one line at a time, even when
|
||||
the complete export exceeds the engine string limit. This opt-in lane is separate from routine tests and the canonical hash check
|
||||
`npm run verify:javascript:digests`. It verifies serialization rather than an
|
||||
arbitrary third-party application's parsing cost or MCP client capacity.
|
||||
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
# CLI JSON Lines verification — 2026-10-23
|
||||
|
||||
The analyst task was to export retained JavaScript observations as independent
|
||||
records while keeping their Evidence provenance, coverage and unknowns. JSONL
|
||||
framing stays in the CLI adapter; analysis and view projection continue to use
|
||||
the same application workflow as MCP.
|
||||
|
||||
## Real artifact and completed workflow
|
||||
|
||||
The installed, locked `semver@7.8.5` npm package was analyzed through the public
|
||||
`analyze-javascript-application` command, without executing its target code.
|
||||
It parsed 50 selected files and produced a 12,185,829-byte Evidence document.
|
||||
Preparation took 12.295 s elapsed, 12.648 s child CPU and 678.5 MiB sampled peak
|
||||
process-family RSS; preparation is separate from the export measurements.
|
||||
The artifact digest was
|
||||
`65c9cce0fbebc34824c1b9657a514e7c5525ab079b232ce9da095697271b7953`.
|
||||
The saved Evidence file SHA-256 was
|
||||
`b34627bc12894d4c749e5b24c60731dd5a7eaaf90c98954f0e585eb442d484c3`.
|
||||
|
||||
Preparation ran on main `169512a35`; the exact retained parent was reused by
|
||||
both refreshed export variants:
|
||||
|
||||
```json
|
||||
{
|
||||
"source": { "kind": "evidence-file", "path": "/absolute/path/parent.json" },
|
||||
"view": {
|
||||
"kind": "javascript-facts",
|
||||
"property_keys": [],
|
||||
"include_available_property_keys": true,
|
||||
"offset": 0,
|
||||
"limit": 256
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Each completed measurement executed one public CLI call:
|
||||
|
||||
```bash
|
||||
node scripts/rea.mjs inspect-analysis-view request.json --format jsonl
|
||||
```
|
||||
|
||||
The view contained 202 observation rows. Every before/after pair reconstructed
|
||||
exactly equal complete Evidence JSON, including identities, subject/provider,
|
||||
authority/confidence, locations, parent/view digests, links, original coverage,
|
||||
17 view limitations and 20 unknowns. These source limitations remain limitations;
|
||||
exhausting this selected view does not establish complete application analysis.
|
||||
Before emitted one document; after emitted 202 rows and one metadata record.
|
||||
|
||||
## Performance
|
||||
|
||||
Baseline: refreshed main `2443b949a48c3b2eba687e7381682e49e909d5b9`.
|
||||
Both variants used Node 24.22.1 and the same locked dependency installation.
|
||||
Three alternating pairs ran sequentially, with fresh processes and warm
|
||||
filesystem/build artifacts. Node compile caching was not configured. Cold host
|
||||
filesystem caches and repeated analysis were not measured.
|
||||
|
||||
| Completed export, median of three | Before | After |
|
||||
| --------------------------------- | --------: | --------: |
|
||||
| Elapsed time | 4.510 s | 4.576 s |
|
||||
| Wait-accounted child CPU | 5.876 s | 5.987 s |
|
||||
| Sampled peak process-family RSS | 496.3 MiB | 495.1 MiB |
|
||||
| Emitted UTF-8 bytes | 365,494 | 365,608 |
|
||||
| `tokenx` token estimate | 99,004 | 99,041 |
|
||||
| CLI calls | 1 | 1 |
|
||||
|
||||
The small observed overhead does not establish an analysis speedup.
|
||||
The 114-byte / 37-estimated-token increase is framing overhead, with the same
|
||||
facts and no additional tool/schema round trips. The identical request file
|
||||
contributes another 74 estimated tokens to each case. These are heuristic `tokenx` estimates, not model-specific BPE counts;
|
||||
orchestration prompts, agent reasoning and tool-definition tokens were not
|
||||
measured. Actual agent context consumption also depends on which rows the
|
||||
consumer reads. No token reduction is claimed.
|
||||
|
||||
Rows and metadata share the existing bounded JSON encoder and awaited writes.
|
||||
Token counting reuses that exact record plan, including context, and estimates
|
||||
one complete line at a time rather than materializing the complete export.
|
||||
A single line still must fit the estimator's runtime string limit. Token-count
|
||||
performance, much larger applications and time to first consumed row were not
|
||||
benchmarked.
|
||||
|
||||
The harness pinned the process family to two CPUs, used a 2 GiB V8 heap limit,
|
||||
`GOMAXPROCS=2`, `GOMEMLIMIT=2GiB` and two libuv threads. It sampled descendant
|
||||
RSS every 50 ms and would terminate the task at 6 GiB or 30 minutes. No resource
|
||||
stop occurred. RSS sums can double-count shared pages and miss short peaks;
|
||||
the V8 limit is per process and the sampler is not a kernel memory limit.
|
||||
|
||||
## Validation
|
||||
|
||||
`npm run check` passed on the refreshed integration, including type checking,
|
||||
lint, formatting, dependency/metadata checks and architecture guards. The parsed
|
||||
output-control contract from current main is retained; handlers, launcher and
|
||||
reusable test drivers consume those same facts.
|
||||
|
||||
- 241 tests passed across 14 CLI/MCP/acceptance files: actual production MCP
|
||||
catalogs, authenticated Evidence reconstruction, invalid SDK values, ordinary
|
||||
and sentinel failures, generator streaming, formatter recovery, empty/partial
|
||||
views, large output, pipe closure and asynchronous final writes after cleanup
|
||||
success/failure. The run took 171.097 s elapsed, 203.968 s child CPU and
|
||||
1589.3 MiB sampled peak family RSS.
|
||||
- Another 60 tests passed across four consumer files, including ordinary graph
|
||||
reports and scalar selections refusing JSONL, malformed JSON input retaining
|
||||
diagnostics, managed CLI output and the direct SDK update-alias contract.
|
||||
These took 223.152 s elapsed, 262.538 s child CPU and 1079.2 MiB peak family RSS.
|
||||
- A new negative control showed that calling a collection encoder for a single
|
||||
array-valued row undercounted tokens (15 instead of 21, 57 instead of 63).
|
||||
Emission and counting now share `jsonLineParts`; both cases and all 29 JSONL
|
||||
tests pass after the fix. The 241-test run includes these regressions.
|
||||
- A real `rea mcp doctor --format jsonl` run reported healthy in one JSON line.
|
||||
Startup/validation failures, full-output framing and advertised help are
|
||||
exercised separately.
|
||||
|
||||
The opt-in verifier used shared 64 KiB leaves and independently checked complete
|
||||
output bytes and SHA-256, without retaining the complete encoded document:
|
||||
|
||||
| Synthetic serialization fixture | JSON | JSONL |
|
||||
| ----------------------------------------------- | ---------------: | ----------------: |
|
||||
| Output bytes (engine string limit: 536,870,888) | 536,961,655 | 536,961,779 |
|
||||
| Elapsed / child CPU | 9.811 / 10.197 s | 13.940 / 14.451 s |
|
||||
| Sampled peak family RSS | 171.6 MiB | 186.0 MiB |
|
||||
|
||||
The JSONL run also checked its token count against the actual file line by line:
|
||||
89,492,357 heuristic tokens matched. JSONL timing includes this extra counting
|
||||
verification, so the two fixture timings are not equivalent-workflow performance
|
||||
comparisons. Shared leaves keep input memory small; this verifies serialization
|
||||
expansion and counting capacity, not a 512 MiB application's analysis footprint.
|
||||
Both temporary output files were removed by the verifier's cleanup.
|
||||
|
||||
No resource stop occurred. Native provider behavior is outside this CLI change.
|
||||
`npm run docs:check` passed, including nine README consistency tests.
|
||||
|
||||
The refreshed measurements are in the `refreshed/` subdirectory; earlier
|
||||
measurements against `169512a35` remain separate. Raw stdout/stderr, commands,
|
||||
per-run resource samples, equivalence assertions and summaries are retained under
|
||||
`/tmp/rea-maintenance-20261023/5629/`; they are not committed artifacts.
|
||||
+25
-20
@@ -140,6 +140,7 @@ if (!(await compiledRuntimeExists(runtimeFiles))) {
|
||||
return;
|
||||
}
|
||||
if (output.length > 0) wroteOutput = true;
|
||||
if (resultOutput?.captureSdkOutput(output)) return;
|
||||
writeStdout(output);
|
||||
},
|
||||
// A command failure caused by the closed pipe keeps the pipe's status.
|
||||
@@ -147,26 +148,25 @@ if (!(await compiledRuntimeExists(runtimeFiles))) {
|
||||
// shutdown is reported here because this callback is what exits.
|
||||
exit: (code) => {
|
||||
exiting = true;
|
||||
const status = stdoutClosed ? SIGPIPE_EXIT_STATUS : code;
|
||||
void Promise.resolve()
|
||||
.then(() => closeCliJavaScriptResources())
|
||||
.then(
|
||||
() => {
|
||||
process.exit(status);
|
||||
},
|
||||
(cleanup) => {
|
||||
process.stderr.write(
|
||||
`Failed to close the JavaScript analysis worker: ${cleanupFailureText(cleanup)}\n`,
|
||||
);
|
||||
process.exit(
|
||||
stdoutClosed
|
||||
? SIGPIPE_EXIT_STATUS
|
||||
: status === 0
|
||||
? 1
|
||||
: status,
|
||||
);
|
||||
},
|
||||
);
|
||||
void (async () => {
|
||||
let status = code;
|
||||
try {
|
||||
await closeCliJavaScriptResources();
|
||||
} catch (cleanup) {
|
||||
process.stderr.write(
|
||||
`Failed to close the JavaScript analysis worker: ${cleanupFailureText(cleanup)}\n`,
|
||||
);
|
||||
if (status === 0) status = 1;
|
||||
}
|
||||
try {
|
||||
await resultOutput?.finishSdkOutput(code);
|
||||
} catch (cause) {
|
||||
if (!stdoutClosed)
|
||||
process.stderr.write(`${cleanupFailureText(cause)}\n`);
|
||||
if (status === 0) status = 1;
|
||||
}
|
||||
process.exit(stdoutClosed ? SIGPIPE_EXIT_STATUS : status);
|
||||
})();
|
||||
},
|
||||
});
|
||||
} catch (cause) {
|
||||
@@ -184,6 +184,11 @@ if (!(await compiledRuntimeExists(runtimeFiles))) {
|
||||
}
|
||||
if (!wroteOutput && !resultOutput?.handled)
|
||||
writeStdout(renderEmptyFilteredCliOutput(outputOptions) ?? "");
|
||||
// An SDK failure exits through the callback, which owns its pending write.
|
||||
if (!exiting)
|
||||
await resultOutput?.finishSdkOutput(
|
||||
typeof process.exitCode === "number" ? process.exitCode : 0,
|
||||
);
|
||||
}
|
||||
} catch (cause) {
|
||||
dispatchFailure = cause;
|
||||
|
||||
@@ -2,6 +2,8 @@ import assert from "node:assert/strict";
|
||||
import { constants } from "node:buffer";
|
||||
import { createHash } from "node:crypto";
|
||||
import { once } from "node:events";
|
||||
import { createInterface } from "node:readline";
|
||||
import { Writable } from "node:stream";
|
||||
import { createReadStream, createWriteStream } from "node:fs";
|
||||
import { mkdtemp, rm, stat, statfs } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
@@ -9,6 +11,7 @@ import { join } from "node:path";
|
||||
import { finished } from "node:stream/promises";
|
||||
|
||||
import { Cli } from "incur";
|
||||
import { estimateTokenCount } from "tokenx";
|
||||
|
||||
import { parseCliOutputArguments } from "../../../dist/cliOutput.js";
|
||||
import { createCliResultOutput } from "../../../dist/cli/cliResultOutput.js";
|
||||
@@ -44,26 +47,47 @@ try {
|
||||
{
|
||||
operation: "inspect_fixture",
|
||||
parameters: {},
|
||||
result: { payload: Array.from({ length: count }, () => leaf) },
|
||||
result: { rows: Array.from({ length: count }, () => leaf) },
|
||||
},
|
||||
);
|
||||
sampleRss();
|
||||
const metadata = { command: "inspect-output-fixture", duration: "0ms" };
|
||||
const marker = "__REA_OUTPUT_ENCODING_MARKER__";
|
||||
const template = {
|
||||
ok: true,
|
||||
data: { ...evidence, normalized_result: { payload: [marker] } },
|
||||
meta: metadata,
|
||||
};
|
||||
// The destination is a redirected file, so both formats are compact.
|
||||
const skeleton = JSON.stringify(template);
|
||||
const encodedMarker = JSON.stringify(marker);
|
||||
const markerIndex = skeleton.indexOf(encodedMarker);
|
||||
assert.ok(markerIndex >= 0);
|
||||
assert.equal(markerIndex, skeleton.lastIndexOf(encodedMarker));
|
||||
const prefix = skeleton.slice(0, markerIndex);
|
||||
const suffix = `${skeleton.slice(markerIndex + encodedMarker.length)}\n`;
|
||||
const separator = ",";
|
||||
let prefix;
|
||||
let suffix;
|
||||
if (format === "json") {
|
||||
const template = {
|
||||
ok: true,
|
||||
data: { ...evidence, normalized_result: { rows: [marker] } },
|
||||
meta: metadata,
|
||||
};
|
||||
const skeleton = JSON.stringify(template);
|
||||
const encodedMarker = JSON.stringify(marker);
|
||||
const markerIndex = skeleton.indexOf(encodedMarker);
|
||||
assert.ok(markerIndex >= 0);
|
||||
assert.equal(markerIndex, skeleton.lastIndexOf(encodedMarker));
|
||||
prefix = skeleton.slice(0, markerIndex);
|
||||
suffix = `${skeleton.slice(markerIndex + encodedMarker.length)}\n`;
|
||||
} else {
|
||||
// Independent protocol golden: rows are lines; the final record retains
|
||||
// every non-row fact and the caller's full-output envelope.
|
||||
prefix = "";
|
||||
suffix = `\n${JSON.stringify({
|
||||
$rea: "jsonl/v1",
|
||||
type: "metadata",
|
||||
records: {
|
||||
path: ["data", "normalized_result", "rows"],
|
||||
emitted: count,
|
||||
source: count,
|
||||
},
|
||||
context: {
|
||||
ok: true,
|
||||
data: { ...evidence, normalized_result: {} },
|
||||
meta: metadata,
|
||||
},
|
||||
})}\n`;
|
||||
}
|
||||
const separator = format === "jsonl" ? "\n" : ",";
|
||||
const expectedHash = createHash("sha256");
|
||||
expectedHash.update(prefix);
|
||||
for (let index = 0; index < count; index += 1) {
|
||||
@@ -109,6 +133,7 @@ try {
|
||||
assert.equal(exitCode, 0);
|
||||
assert.equal(output.handled, true);
|
||||
assert.equal(output.failed, false);
|
||||
assert.equal(output.error, undefined);
|
||||
assert.equal(unexpectedOutputBytes, 0);
|
||||
|
||||
const actualHash = createHash("sha256");
|
||||
@@ -116,12 +141,50 @@ try {
|
||||
const digest = actualHash.digest("hex");
|
||||
assert.equal(digest, expectedHash.digest("hex"));
|
||||
assert.equal((await stat(path)).size, expectedBytes);
|
||||
let verifiedTokenCount;
|
||||
if (format === "jsonl") {
|
||||
let expectedTokens = 0;
|
||||
for await (const line of createInterface({
|
||||
input: createReadStream(path),
|
||||
crlfDelay: Infinity,
|
||||
}))
|
||||
expectedTokens += estimateTokenCount(line);
|
||||
let tokenText = "";
|
||||
const counterDestination = new Writable({
|
||||
write(chunk, _encoding, callback) {
|
||||
// The numeric count is tiny; never retain a mistakenly emitted report.
|
||||
if (tokenText.length + chunk.length > 64) {
|
||||
callback(
|
||||
new Error("Token counter emitted more than a numeric count"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
tokenText += chunk.toString("utf8");
|
||||
callback();
|
||||
},
|
||||
});
|
||||
const counter = createCliResultOutput(
|
||||
parseCliOutputArguments([...argv, "--token-count"]),
|
||||
counterDestination,
|
||||
);
|
||||
assert.ok(counter);
|
||||
assert.equal(await counter.write(evidence, { ...metadata, format }), true);
|
||||
assert.equal(counter.error, undefined);
|
||||
assert.equal(counter.failed, false);
|
||||
counterDestination.end();
|
||||
await finished(counterDestination);
|
||||
assert.equal(tokenText, `${expectedTokens}\n`);
|
||||
verifiedTokenCount = expectedTokens;
|
||||
}
|
||||
sampleRss();
|
||||
finalReport = {
|
||||
json_output_verified: true,
|
||||
format,
|
||||
output_bytes: expectedBytes,
|
||||
engine_string_limit: constants.MAX_STRING_LENGTH,
|
||||
...(verifiedTokenCount === undefined
|
||||
? {}
|
||||
: { verified_token_count: verifiedTokenCount }),
|
||||
output_sha256: digest,
|
||||
evidence_id: evidence.evidence_id,
|
||||
rss: { method: "node-memoryUsage-rss-sampled", peak_bytes: peakRss },
|
||||
|
||||
+100
-9
@@ -16,6 +16,8 @@ import {
|
||||
type AnalysisErrorProjection,
|
||||
} from "../domain/analysisErrorProjection.js";
|
||||
import { formatCliOutput } from "./cliOutputFormatter.js";
|
||||
import { jsonLinesErrorRecord } from "./jsonLines.js";
|
||||
import { writeCliJsonLines } from "./jsonLinesOutput.js";
|
||||
import {
|
||||
jsonOutputParts,
|
||||
writeJsonOutput,
|
||||
@@ -40,6 +42,13 @@ export interface CliResultOutput {
|
||||
readonly handled: boolean;
|
||||
readonly failed: boolean;
|
||||
readonly error: AnalysisErrorProjection | undefined;
|
||||
/** Start another completed invocation on a reusable command registry. */
|
||||
begin(options: ParsedCliOutputArguments): void;
|
||||
/** SDK generators retain their own streaming event protocol. */
|
||||
delegateStreaming(): void;
|
||||
/** Retain one-shot SDK output until the SDK supplies its authoritative status. */
|
||||
captureSdkOutput(text: string): boolean;
|
||||
finishSdkOutput(exitCode: number): Promise<void>;
|
||||
write(value: unknown, metadata: CliResultOutputMetadata): Promise<boolean>;
|
||||
/** Recover only before any delegated stdout has been written. */
|
||||
recoverFormattingFailure(cause: unknown): Promise<boolean>;
|
||||
@@ -70,8 +79,10 @@ const wrapCommand = (
|
||||
typeof result === "object" &&
|
||||
result !== null &&
|
||||
Symbol.asyncIterator in result
|
||||
)
|
||||
) {
|
||||
output.delegateStreaming();
|
||||
return result;
|
||||
}
|
||||
return Promise.resolve(result).then(async (value: unknown) => {
|
||||
if (
|
||||
typeof value === "object" &&
|
||||
@@ -100,14 +111,22 @@ const wrapCommand = (
|
||||
|
||||
/** Preserve Incur output controls and stream JSON while framing failures. */
|
||||
export const createCliResultOutput = (
|
||||
options: ParsedCliOutputArguments,
|
||||
initialOptions: ParsedCliOutputArguments,
|
||||
destination: Writable,
|
||||
): CliResultOutput | undefined => {
|
||||
if (options.parseError || !options.validation.ok) return undefined;
|
||||
const format = options.format as "json" | "jsonl" | "yaml" | "toon" | "md";
|
||||
let options = initialOptions;
|
||||
if (
|
||||
(options.parseError && options.format !== "jsonl") ||
|
||||
!options.validation.ok
|
||||
)
|
||||
return undefined;
|
||||
let format = options.format as "json" | "jsonl" | "yaml" | "toon" | "md";
|
||||
let handled = false;
|
||||
let failed = false;
|
||||
let error: AnalysisErrorProjection | undefined;
|
||||
let streaming = false;
|
||||
const sdkParts: string[] = [];
|
||||
let sdkCompletion: Promise<void> | undefined;
|
||||
let completed:
|
||||
| {
|
||||
readonly metadata: CliResultOutputMetadata;
|
||||
@@ -160,7 +179,10 @@ export const createCliResultOutput = (
|
||||
: error;
|
||||
// Recovery must remain actionable even when a token window/count was
|
||||
// requested. Do not apply those controls to this small typed failure.
|
||||
const text = formatCliOutput(document, format);
|
||||
const text = formatCliOutput(
|
||||
format === "jsonl" ? jsonLinesErrorRecord(document) : document,
|
||||
format,
|
||||
);
|
||||
handled = true;
|
||||
try {
|
||||
await writeOutputText(destination, `${text}\n`);
|
||||
@@ -193,6 +215,55 @@ export const createCliResultOutput = (
|
||||
get error() {
|
||||
return error;
|
||||
},
|
||||
begin(nextOptions) {
|
||||
options = nextOptions;
|
||||
format = options.format as typeof format;
|
||||
handled = false;
|
||||
failed = false;
|
||||
error = undefined;
|
||||
completed = undefined;
|
||||
streaming = false;
|
||||
sdkParts.length = 0;
|
||||
sdkCompletion = undefined;
|
||||
},
|
||||
delegateStreaming() {
|
||||
streaming = true;
|
||||
},
|
||||
captureSdkOutput(text) {
|
||||
if (format !== "jsonl" || streaming || handled) return false;
|
||||
// Incur already materializes one-shot documents. Retain references only;
|
||||
// generator events bypass this buffer and remain incremental.
|
||||
if (text.length > 0) sdkParts.push(text);
|
||||
return true;
|
||||
},
|
||||
finishSdkOutput(exitCode) {
|
||||
return (sdkCompletion ??= (async () => {
|
||||
const parts = sdkParts.splice(0);
|
||||
if (parts.length === 0) return;
|
||||
try {
|
||||
if (exitCode === 0) {
|
||||
for (const part of parts) await writeOutputText(destination, part);
|
||||
} else {
|
||||
const text = parts.join("");
|
||||
let result: unknown = text;
|
||||
try {
|
||||
result = JSON.parse(text);
|
||||
} catch {
|
||||
/* Preserve non-JSON SDK diagnostics verbatim. */
|
||||
}
|
||||
await writeJsonOutput(
|
||||
jsonLinesErrorRecord(result),
|
||||
destination,
|
||||
"jsonl",
|
||||
);
|
||||
}
|
||||
} catch (cause: unknown) {
|
||||
handled = true;
|
||||
failed = true;
|
||||
throw cause;
|
||||
}
|
||||
})());
|
||||
},
|
||||
write,
|
||||
recoverFormattingFailure,
|
||||
};
|
||||
@@ -209,7 +280,7 @@ export const createCliResultOutput = (
|
||||
? Filter.parse(options.filterOutput)
|
||||
: undefined;
|
||||
const operationFailure =
|
||||
options.fullOutput || filterPaths !== undefined
|
||||
format === "jsonl" || options.fullOutput || filterPaths !== undefined
|
||||
? cliOperationFailurePayload(value)
|
||||
: undefined;
|
||||
const failure = options.fullOutput ? operationFailure : undefined;
|
||||
@@ -222,6 +293,7 @@ export const createCliResultOutput = (
|
||||
if (
|
||||
operationFailure === undefined &&
|
||||
metadata.renderOutput === false &&
|
||||
format !== "jsonl" &&
|
||||
!options.fullOutput &&
|
||||
!options.tokenCount
|
||||
)
|
||||
@@ -254,6 +326,26 @@ export const createCliResultOutput = (
|
||||
!ownsSuccessfulTokenWindow
|
||||
)
|
||||
return false;
|
||||
if (format === "jsonl") {
|
||||
handled = true;
|
||||
try {
|
||||
error = await writeCliJsonLines(
|
||||
{
|
||||
...options,
|
||||
source: value,
|
||||
selected: filtered,
|
||||
operationFailure,
|
||||
command: metadata.command,
|
||||
duration: metadata.duration,
|
||||
},
|
||||
destination,
|
||||
);
|
||||
} catch (cause: unknown) {
|
||||
failed = true;
|
||||
throw cause;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (options.tokenCount) {
|
||||
// Incur/tokenx materialize JSON. Measure its UTF-16 expansion before
|
||||
// allowing that allocation, independently of UTF-8 bytes. Other formats
|
||||
@@ -266,7 +358,7 @@ export const createCliResultOutput = (
|
||||
typeof tokenValue !== "function" &&
|
||||
typeof tokenValue !== "symbol"
|
||||
) {
|
||||
if (format === "json" || format === "jsonl") {
|
||||
if (format === "json") {
|
||||
for (const part of jsonOutputParts(
|
||||
tokenValue,
|
||||
format,
|
||||
@@ -382,8 +474,7 @@ export const createCliResultOutput = (
|
||||
const formattedText =
|
||||
format === "md" ||
|
||||
((operationFailure !== undefined || ownsFilteredOutput) &&
|
||||
format !== "json" &&
|
||||
format !== "jsonl")
|
||||
format !== "json")
|
||||
? formatCliOutput(document, format)
|
||||
: undefined;
|
||||
if (
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
import { err, ok, type Result } from "../domain/result.js";
|
||||
|
||||
import { AnalysisOutputError } from "../domain/analysisErrorCore.js";
|
||||
import { jsonParts } from "../domain/jsonSerialization.js";
|
||||
|
||||
/** Reserved control-record marker, not an advertised external JSON Schema. */
|
||||
export const JSON_LINES_PROTOCOL = "jsonl/v1";
|
||||
|
||||
interface JsonLinesErrorRecord {
|
||||
readonly $rea: typeof JSON_LINES_PROTOCOL;
|
||||
readonly type: "error";
|
||||
readonly ok: false;
|
||||
readonly result: unknown;
|
||||
}
|
||||
|
||||
interface JsonLinesMetadataRecord {
|
||||
readonly $rea: typeof JSON_LINES_PROTOCOL;
|
||||
readonly type: "metadata";
|
||||
readonly records: {
|
||||
readonly path: readonly string[];
|
||||
readonly emitted: number;
|
||||
readonly source: number | null;
|
||||
readonly filter?: string;
|
||||
};
|
||||
readonly context: Readonly<Record<string, unknown>> | null;
|
||||
}
|
||||
|
||||
const COLLECTION_KEYS = ["rows", "items", "records", "occurrences"] as const;
|
||||
|
||||
interface Collection {
|
||||
readonly records: readonly unknown[];
|
||||
readonly path: readonly string[];
|
||||
}
|
||||
|
||||
/** Parsed once, before output or token counting acquires the stream. */
|
||||
export interface JsonLinesPlan extends Collection {
|
||||
readonly context: Readonly<Record<string, unknown>> | undefined;
|
||||
readonly sourceRecords: number | null;
|
||||
readonly filter: string | undefined;
|
||||
}
|
||||
|
||||
const isRecord = (value: unknown): value is Record<string, unknown> =>
|
||||
typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
|
||||
const atPath = (value: unknown, path: readonly string[]): unknown => {
|
||||
for (const key of path) {
|
||||
if (!isRecord(value) || !Object.hasOwn(value, key)) return undefined;
|
||||
value = value[key];
|
||||
}
|
||||
return value;
|
||||
};
|
||||
|
||||
const collections = (
|
||||
value: unknown,
|
||||
prefix: readonly string[],
|
||||
): Collection[] => {
|
||||
if (Array.isArray(value)) return [{ records: value, path: prefix }];
|
||||
if (!isRecord(value)) return [];
|
||||
const found: Collection[] = [];
|
||||
const collect = (
|
||||
object: Record<string, unknown>,
|
||||
path: readonly string[],
|
||||
) => {
|
||||
for (const key of COLLECTION_KEYS) {
|
||||
const records = object[key];
|
||||
if (Object.hasOwn(object, key) && Array.isArray(records))
|
||||
found.push({ records, path: [...path, key] });
|
||||
}
|
||||
};
|
||||
collect(value, prefix);
|
||||
if (Object.hasOwn(value, "page") && isRecord(value.page))
|
||||
collect(value.page, [...prefix, "page"]);
|
||||
return found;
|
||||
};
|
||||
|
||||
const documentCollections = (
|
||||
value: unknown,
|
||||
fullOutput: boolean,
|
||||
): Collection[] => {
|
||||
const prefix = fullOutput ? ["data"] : [];
|
||||
const data = atPath(value, prefix);
|
||||
return [
|
||||
...collections(data, prefix),
|
||||
...collections(atPath(data, ["normalized_result"]), [
|
||||
...prefix,
|
||||
"normalized_result",
|
||||
]),
|
||||
];
|
||||
};
|
||||
|
||||
/** Shallow copies only ancestors of the emitted array; every other fact survives. */
|
||||
const withoutRecords = (
|
||||
source: unknown,
|
||||
selected: unknown,
|
||||
path: readonly string[],
|
||||
filtered: boolean,
|
||||
): Record<string, unknown> | undefined => {
|
||||
if (path.length === 0 || !isRecord(source)) return undefined;
|
||||
const context = { ...source };
|
||||
const selectedObject = isRecord(selected) ? selected : undefined;
|
||||
// A data filter intentionally excludes raw observation payloads. Keep its
|
||||
// selected raw value if requested, while always retaining Evidence provenance.
|
||||
if (
|
||||
filtered &&
|
||||
Object.hasOwn(source, "evidence_id") &&
|
||||
typeof source.evidence_id === "string" &&
|
||||
Object.hasOwn(source, "normalized_result")
|
||||
) {
|
||||
if (selectedObject && Object.hasOwn(selectedObject, "raw_result"))
|
||||
context.raw_result = selectedObject.raw_result;
|
||||
else delete context.raw_result;
|
||||
}
|
||||
const [key, ...rest] = path;
|
||||
if (key === undefined) return context;
|
||||
if (rest.length === 0) delete context[key];
|
||||
else
|
||||
context[key] = withoutRecords(
|
||||
source[key],
|
||||
selectedObject?.[key],
|
||||
rest,
|
||||
filtered,
|
||||
);
|
||||
return context;
|
||||
};
|
||||
|
||||
/** Project a clear collection without interpreting or whitelisting its context. */
|
||||
export const prepareJsonLines = (
|
||||
selected: unknown,
|
||||
options: {
|
||||
readonly command: string;
|
||||
readonly source?: unknown;
|
||||
readonly filter?: string;
|
||||
readonly fullOutput?: boolean;
|
||||
},
|
||||
): Result<JsonLinesPlan, AnalysisOutputError> => {
|
||||
const found = documentCollections(selected, options.fullOutput === true);
|
||||
const reject = (reason: string) =>
|
||||
err(
|
||||
new AnalysisOutputError(options.command, reason, {
|
||||
remediationAction:
|
||||
"Use --format json for the complete document, or select one record collection with --filter-output.",
|
||||
}),
|
||||
);
|
||||
if (found.length !== 1)
|
||||
return reject(
|
||||
found.length === 0
|
||||
? "JSON Lines requires an array or one explicit rows, items, records, or occurrences collection."
|
||||
: "JSON Lines cannot choose between multiple record collections. Select one collection explicitly.",
|
||||
);
|
||||
const collection = found[0];
|
||||
if (collection === undefined)
|
||||
return reject("The record collection is unavailable.");
|
||||
for (const record of collection.records) {
|
||||
if (isRecord(record) && Object.hasOwn(record, "$rea"))
|
||||
return reject(
|
||||
"A data record uses the reserved $rea JSON Lines control marker.",
|
||||
);
|
||||
}
|
||||
const source = Object.hasOwn(options, "source") ? options.source : selected;
|
||||
const originalRecords = atPath(source, collection.path);
|
||||
const sourceMatches = Array.isArray(originalRecords);
|
||||
let context = withoutRecords(
|
||||
sourceMatches ? source : selected,
|
||||
selected,
|
||||
collection.path,
|
||||
options.filter !== undefined,
|
||||
);
|
||||
if (options.filter !== undefined && sourceMatches) {
|
||||
// Excluded peer rows are data, not context. A filter can disambiguate them
|
||||
// without restoring an unselected collection into the final record.
|
||||
for (const peer of documentCollections(
|
||||
source,
|
||||
options.fullOutput === true,
|
||||
)) {
|
||||
if (peer.path.join(".") !== collection.path.join("."))
|
||||
context = withoutRecords(context, selected, peer.path, true);
|
||||
}
|
||||
}
|
||||
return ok({
|
||||
...collection,
|
||||
context,
|
||||
sourceRecords: sourceMatches ? originalRecords.length : null,
|
||||
filter: options.filter,
|
||||
});
|
||||
};
|
||||
|
||||
/** The caller supplies the outcome; error-shaped data is never reclassified here. */
|
||||
export const jsonLinesErrorRecord = (
|
||||
result: unknown,
|
||||
): JsonLinesErrorRecord => ({
|
||||
$rea: JSON_LINES_PROTOCOL,
|
||||
type: "error" as const,
|
||||
ok: false as const,
|
||||
result,
|
||||
});
|
||||
|
||||
export function* jsonLinesParts(plan: JsonLinesPlan): Generator<string> {
|
||||
for (const record of jsonLinesRecords(plan)) {
|
||||
yield* jsonLineParts(record);
|
||||
}
|
||||
}
|
||||
|
||||
/** One record stays one line even when the record itself is an array. */
|
||||
export function* jsonLineParts(record: unknown): Generator<string> {
|
||||
yield* jsonParts(record);
|
||||
yield "\n";
|
||||
}
|
||||
|
||||
/** Newline-separated records also bound the token estimator's working set. */
|
||||
export function* jsonLinesRecords(plan: JsonLinesPlan): Generator<unknown> {
|
||||
yield* plan.records;
|
||||
if (plan.context !== undefined || plan.path.length > 0) {
|
||||
const metadata: JsonLinesMetadataRecord = {
|
||||
$rea: JSON_LINES_PROTOCOL,
|
||||
type: "metadata",
|
||||
records: {
|
||||
path: plan.path,
|
||||
emitted: plan.records.length,
|
||||
source: plan.sourceRecords,
|
||||
...(plan.filter === undefined ? {} : { filter: plan.filter }),
|
||||
},
|
||||
context: plan.context ?? null,
|
||||
};
|
||||
yield metadata;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
import { constants } from "node:buffer";
|
||||
import type { Writable } from "node:stream";
|
||||
import { estimateTokenCount } from "tokenx";
|
||||
|
||||
import { AnalysisResourceConstraintError } from "../domain/analysisErrorCore.js";
|
||||
import {
|
||||
projectAnalysisError,
|
||||
type AnalysisErrorProjection,
|
||||
} from "../domain/analysisErrorProjection.js";
|
||||
import {
|
||||
jsonLineParts,
|
||||
jsonLinesErrorRecord,
|
||||
jsonLinesParts,
|
||||
jsonLinesRecords,
|
||||
prepareJsonLines,
|
||||
} from "./jsonLines.js";
|
||||
import {
|
||||
writeJsonOutput,
|
||||
writeOutputParts,
|
||||
writeOutputText,
|
||||
} from "./jsonOutput.js";
|
||||
|
||||
interface CliJsonLinesInput {
|
||||
readonly source: unknown;
|
||||
readonly selected: unknown;
|
||||
readonly operationFailure: unknown;
|
||||
readonly command: string;
|
||||
readonly duration: string;
|
||||
readonly fullOutput: boolean;
|
||||
readonly tokenCount: boolean;
|
||||
readonly filterOutput: string | undefined;
|
||||
}
|
||||
|
||||
/** Render one already classified result, sharing a plan for output and counting. */
|
||||
export const writeCliJsonLines = async (
|
||||
options: CliJsonLinesInput,
|
||||
destination: Writable,
|
||||
): Promise<AnalysisErrorProjection | undefined> => {
|
||||
const { source, selected, operationFailure } = options;
|
||||
let error: AnalysisErrorProjection | undefined;
|
||||
const meta = { command: options.command, duration: options.duration };
|
||||
const envelope = (data: unknown) =>
|
||||
options.fullOutput ? { ok: true, data, meta } : data;
|
||||
let failureRecord =
|
||||
operationFailure === undefined
|
||||
? undefined
|
||||
: jsonLinesErrorRecord(
|
||||
options.fullOutput
|
||||
? { ok: false, error: operationFailure, meta }
|
||||
: operationFailure,
|
||||
);
|
||||
const plan =
|
||||
failureRecord === undefined
|
||||
? prepareJsonLines(envelope(selected), {
|
||||
command: options.command,
|
||||
source: envelope(source),
|
||||
...(options.filterOutput === undefined
|
||||
? {}
|
||||
: { filter: options.filterOutput }),
|
||||
fullOutput: options.fullOutput,
|
||||
})
|
||||
: undefined;
|
||||
if (plan?.ok === false) {
|
||||
error = projectAnalysisError(plan.error);
|
||||
failureRecord = jsonLinesErrorRecord(
|
||||
options.fullOutput ? { ok: false, error, meta } : error,
|
||||
);
|
||||
}
|
||||
if (failureRecord !== undefined) {
|
||||
await writeJsonOutput(failureRecord, destination, "jsonl");
|
||||
} else if (plan?.ok === true) {
|
||||
if (options.tokenCount) {
|
||||
let count = 0;
|
||||
for (const record of jsonLinesRecords(plan.value)) {
|
||||
let characters = 0;
|
||||
for (const part of jsonLineParts(record)) characters += part.length;
|
||||
if (characters > constants.MAX_STRING_LENGTH) {
|
||||
error = projectAnalysisError(
|
||||
new AnalysisResourceConstraintError(
|
||||
options.command,
|
||||
"memory",
|
||||
"A JSON Lines record exceeds the token estimator's runtime string limit.",
|
||||
{
|
||||
formatted_characters: characters,
|
||||
max_string_characters: constants.MAX_STRING_LENGTH,
|
||||
},
|
||||
{
|
||||
remediationAction:
|
||||
"Remove --token-count to stream the records, or select smaller fields with --filter-output.",
|
||||
},
|
||||
),
|
||||
);
|
||||
break;
|
||||
}
|
||||
// Newlines split tokenx segments, so summing complete lines is
|
||||
// identical to estimating the emitted stream, with one-line memory.
|
||||
count += estimateTokenCount([...jsonLineParts(record)].join(""));
|
||||
}
|
||||
if (error !== undefined)
|
||||
await writeJsonOutput(
|
||||
jsonLinesErrorRecord(
|
||||
options.fullOutput ? { ok: false, error, meta } : error,
|
||||
),
|
||||
destination,
|
||||
"jsonl",
|
||||
);
|
||||
else await writeOutputText(destination, `${count}\n`);
|
||||
} else await writeOutputParts(destination, jsonLinesParts(plan.value));
|
||||
}
|
||||
return error;
|
||||
};
|
||||
+11
-2
@@ -16,9 +16,18 @@ export const writeJsonOutput = async (
|
||||
destination: Writable,
|
||||
format: JsonOutputFormat = "json",
|
||||
): Promise<void> => {
|
||||
for (const part of bufferedJsonParts(
|
||||
await writeOutputParts(
|
||||
destination,
|
||||
jsonOutputParts(value, format, isTerminal(destination)),
|
||||
))
|
||||
);
|
||||
};
|
||||
|
||||
/** Share bounded writes and destination ownership across structured formats. */
|
||||
export const writeOutputParts = async (
|
||||
destination: Writable,
|
||||
parts: Iterable<string>,
|
||||
): Promise<void> => {
|
||||
for (const part of bufferedJsonParts(parts))
|
||||
await writeOutputText(destination, part);
|
||||
};
|
||||
|
||||
|
||||
+2
-1
@@ -1,6 +1,7 @@
|
||||
import { formatCliOutput } from "./cli/cliOutputFormatter.js";
|
||||
|
||||
import { jsonOutputParts } from "./cli/jsonOutput.js";
|
||||
import { jsonLinesErrorRecord } from "./cli/jsonLines.js";
|
||||
import {
|
||||
versionReport,
|
||||
type VersionReportHost,
|
||||
@@ -382,7 +383,7 @@ export const renderCliOutputArgumentError = (
|
||||
};
|
||||
if (error.format === "yaml")
|
||||
return `ok: false\nerror:\n code: ${error.code}\n message: ${JSON.stringify(error.message)}\n`;
|
||||
return `${JSON.stringify(value)}\n`;
|
||||
return `${JSON.stringify(error.format === "jsonl" ? jsonLinesErrorRecord(value) : value)}\n`;
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
+7
-2
@@ -6,6 +6,7 @@ import { resolve } from "node:path";
|
||||
import { StringDecoder } from "node:string_decoder";
|
||||
|
||||
import { createCatalogIdentity } from "./catalogIdentity.js";
|
||||
import { jsonLinesErrorRecord } from "./cli/jsonLines.js";
|
||||
import {
|
||||
CLI_OUTPUT_FORMATS,
|
||||
type CliOutputFormat,
|
||||
@@ -343,7 +344,9 @@ export const runProductionMcpDoctorCli = async (
|
||||
return { output: `${renderMcpDoctorHelp()}\n`, exitCode: 0 };
|
||||
return {
|
||||
output: renderCliDocument(
|
||||
{ code: parsed.code, message: parsed.message },
|
||||
parsed.format === "jsonl"
|
||||
? jsonLinesErrorRecord({ code: parsed.code, message: parsed.message })
|
||||
: { code: parsed.code, message: parsed.message },
|
||||
parsed.format,
|
||||
input.stdoutIsTerminal,
|
||||
),
|
||||
@@ -368,7 +371,9 @@ export const runProductionMcpDoctorCli = async (
|
||||
: result;
|
||||
return {
|
||||
output: renderCliDocument(
|
||||
outputValue,
|
||||
parsed.format === "jsonl" && !result.healthy
|
||||
? jsonLinesErrorRecord(outputValue)
|
||||
: outputValue,
|
||||
parsed.format,
|
||||
input.stdoutIsTerminal,
|
||||
),
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
import { writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { expect } from "vitest";
|
||||
|
||||
import { parseEvidence } from "../../src/domain/evidence.js";
|
||||
import { analysisViewJavaScriptEvidence } from "../fixtures/analysisView.js";
|
||||
import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js";
|
||||
import { cliTest } from "../support/cli/cliFixture.js";
|
||||
|
||||
cliTest(
|
||||
"exports authenticated analysis-view rows, preserving every Evidence fact and empty/partial coverage",
|
||||
async ({ cli }) => {
|
||||
const root = await createTestTempDirectory("rea-json-lines-view-");
|
||||
const parent = analysisViewJavaScriptEvidence();
|
||||
const evidencePath = join(root, "parent.json");
|
||||
const requestPath = join(root, "request.json");
|
||||
await writeFile(evidencePath, JSON.stringify(parent));
|
||||
const request = {
|
||||
source: { kind: "evidence-file", path: evidencePath },
|
||||
view: { kind: "javascript-facts", property_keys: [], limit: 1 },
|
||||
};
|
||||
await writeFile(requestPath, JSON.stringify(request));
|
||||
|
||||
const invalid = await cli.run({
|
||||
arguments: ["inspect-analysis-view", "--format", "jsonl"],
|
||||
});
|
||||
expect(invalid.exitCode).toBe(1);
|
||||
expect(JSON.parse(invalid.stdout)).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "error",
|
||||
result: { code: "VALIDATION_ERROR" },
|
||||
});
|
||||
|
||||
const badOutputValue = await cli.run({
|
||||
arguments: ["--format", "jsonl", "--token-limit", "invalid"],
|
||||
});
|
||||
expect(badOutputValue.exitCode).toBe(1);
|
||||
expect(JSON.parse(badOutputValue.stdout)).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "error",
|
||||
});
|
||||
|
||||
for (const where of [undefined, { paths: [] }]) {
|
||||
await writeFile(
|
||||
requestPath,
|
||||
JSON.stringify({
|
||||
...request,
|
||||
view: { ...request.view, ...(where === undefined ? {} : { where }) },
|
||||
}),
|
||||
);
|
||||
const json = await cli.run({
|
||||
arguments: ["inspect-analysis-view", requestPath, "--json"],
|
||||
});
|
||||
expect(json.exitCode, json.stderr).toBe(0);
|
||||
const expected = parseEvidence(json.json);
|
||||
const result = await cli.run({
|
||||
arguments: ["inspect-analysis-view", requestPath, "--format", "jsonl"],
|
||||
});
|
||||
expect(result.exitCode, result.stderr).toBe(0);
|
||||
const records = result.stdout
|
||||
.trimEnd()
|
||||
.split("\n")
|
||||
.map((line) => JSON.parse(line));
|
||||
const final = records.pop();
|
||||
expect(final).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "metadata",
|
||||
records: {
|
||||
path: ["normalized_result", "rows"],
|
||||
emitted: records.length,
|
||||
source: records.length,
|
||||
},
|
||||
context: {
|
||||
evidence_links: [parent.evidence_id],
|
||||
normalized_result: { parent_evidence_id: parent.evidence_id },
|
||||
},
|
||||
});
|
||||
const reconstructed = {
|
||||
...final.context,
|
||||
normalized_result: {
|
||||
...final.context.normalized_result,
|
||||
rows: records,
|
||||
},
|
||||
};
|
||||
// Authenticate the reconstructed Evidence, rather than checking only a few
|
||||
// convenient fields from a shortened view-shaped object.
|
||||
expect(parseEvidence(reconstructed)).toEqual(expected);
|
||||
expect(records.length).toBe(where === undefined ? 1 : 0);
|
||||
}
|
||||
},
|
||||
);
|
||||
@@ -315,8 +315,15 @@ cliTest(
|
||||
"operation",
|
||||
],
|
||||
});
|
||||
expect(jsonl.exitCode).toBe(0);
|
||||
expect(jsonl.json).toBe("inspect_managed_members");
|
||||
expect(jsonl.exitCode).toBe(1);
|
||||
expect(jsonl.json).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "error",
|
||||
result: {
|
||||
code: "unreadable_output",
|
||||
remediation: { action: expect.stringContaining("--format json") },
|
||||
},
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
@@ -43,7 +43,16 @@ describe("CLI output argument boundary", () => {
|
||||
if (!validation.ok) {
|
||||
const rendered = renderCliOutputArgumentError(validation);
|
||||
expect(rendered).not.toContain("[truncated:");
|
||||
if (format === "json" || format === "jsonl")
|
||||
if (format === "jsonl")
|
||||
expect(JSON.parse(rendered)).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "error",
|
||||
result: {
|
||||
ok: false,
|
||||
error: { code: "UNSUPPORTED_OUTPUT_COMBINATION" },
|
||||
},
|
||||
});
|
||||
else if (format === "json")
|
||||
expect(JSON.parse(rendered)).toMatchObject({
|
||||
ok: false,
|
||||
error: { code: "UNSUPPORTED_OUTPUT_COMBINATION" },
|
||||
|
||||
@@ -97,7 +97,7 @@ describe("executable dispatcher", () => {
|
||||
await installCliRuntime(
|
||||
workspace,
|
||||
'export const closeCliJavaScriptResources = () => {}; export const createCli = () => ({ serve: (_args, options) => { options.stdout("first document\\n"); throw new RangeError("Invalid string length"); } });\n',
|
||||
'export const createCliResultOutput = () => ({ handled: false, failed: false, recoverFormattingFailure: () => { process.stdout.write("unexpected second document"); return true; } });\n',
|
||||
'export const createCliResultOutput = () => ({ handled: false, failed: false, captureSdkOutput: () => false, finishSdkOutput: async () => {}, recoverFormattingFailure: () => { process.stdout.write("unexpected second document"); return true; } });\n',
|
||||
);
|
||||
const result = await processes.run(
|
||||
process.execPath,
|
||||
@@ -225,6 +225,45 @@ describe("executable dispatcher", () => {
|
||||
});
|
||||
|
||||
describe("executable dispatcher JavaScript shutdown", () => {
|
||||
workspaceCliTest.for([
|
||||
"success",
|
||||
"cleanup-failure",
|
||||
"write-failure",
|
||||
"closed-pipe",
|
||||
] as const)(
|
||||
"awaits SDK output after cleanup and retains finalization status: %s",
|
||||
async (scenario, { processes, workspace }) => {
|
||||
const cleanupFails = scenario === "cleanup-failure";
|
||||
const finish =
|
||||
scenario === "closed-pipe"
|
||||
? 'process.stdout.emit("error", Object.assign(new Error("broken pipe"), { code: "EPIPE" })); throw new Error("broken pipe");'
|
||||
: scenario === "write-failure"
|
||||
? 'throw new Error("final write failed");'
|
||||
: 'process.stdout.write("complete document\\n");';
|
||||
await installCliRuntime(
|
||||
workspace,
|
||||
`export const closeCliJavaScriptResources = () => { globalThis.cleaned = true; ${cleanupFails ? 'return Promise.reject(new Error("worker still owned"));' : ""} }; export const createCli = () => ({ serve: (_args, options) => { options.stdout("pending document"); options.exit(0); } });\n`,
|
||||
`export const createCliResultOutput = () => ({ handled: false, failed: false, captureSdkOutput: () => true, finishSdkOutput: async () => { if (!globalThis.cleaned) throw new Error("output preceded cleanup"); await new Promise(resolve => setTimeout(resolve, 10)); ${finish} } });\n`,
|
||||
);
|
||||
const result = await processes.run(
|
||||
process.execPath,
|
||||
[workspace.path("scripts/rea.mjs"), "fixture"],
|
||||
{ cwd: workspace.root },
|
||||
);
|
||||
expect(result.exitCode).toBe(
|
||||
scenario === "closed-pipe" ? 141 : scenario === "success" ? 0 : 1,
|
||||
);
|
||||
expect(result.stdout).toBe(
|
||||
scenario === "success" || cleanupFails ? "complete document\n" : "",
|
||||
);
|
||||
expect(result.stderr).not.toContain("output preceded cleanup");
|
||||
if (cleanupFails) expect(result.stderr).toContain("worker still owned");
|
||||
if (scenario === "write-failure")
|
||||
expect(result.stderr).toContain("final write failed");
|
||||
if (scenario === "closed-pipe") expect(result.stderr).toBe("");
|
||||
},
|
||||
);
|
||||
|
||||
workspaceCliTest(
|
||||
"reports a failed JavaScript worker shutdown instead of exiting successfully",
|
||||
async ({ processes, workspace }) => {
|
||||
|
||||
@@ -97,7 +97,19 @@ it.each(
|
||||
},
|
||||
});
|
||||
const text = Buffer.concat(chunks).toString("utf8");
|
||||
if (full)
|
||||
if (full && format === "jsonl") {
|
||||
const records = text
|
||||
.trimEnd()
|
||||
.split("\n")
|
||||
.map((line) => JSON.parse(line));
|
||||
expect(records.slice(0, -1)).toEqual([1, null, null]);
|
||||
expect(records.at(-1)).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "metadata",
|
||||
records: { path: ["data"], emitted: 3, source: 3, filter: "selected" },
|
||||
context: { ok: true, meta: { command: "rows" } },
|
||||
});
|
||||
} else if (full)
|
||||
expect(JSON.parse(text)).toMatchObject({
|
||||
ok: true,
|
||||
data: [1, null, null],
|
||||
|
||||
@@ -28,7 +28,7 @@ for (const scenario of [
|
||||
{ command: "analyze", format: "jsonl", fullOutput: false, filtered: false },
|
||||
] as const) {
|
||||
cliTest(
|
||||
`${scenario.command} preserves representative ${scenario.format} output`,
|
||||
`${scenario.command} ${scenario.format === "jsonl" ? "refuses graph documents without a supported record boundary" : "preserves representative JSON output"}`,
|
||||
async ({ cli }) => {
|
||||
const root = await createTestTempDirectory("rea-js-json-output-");
|
||||
await writeFile(join(root, "main.js"), "fetch('');\n");
|
||||
@@ -47,6 +47,31 @@ for (const scenario of [
|
||||
...(scenario.filtered ? ["--filter-output", filter] : []),
|
||||
],
|
||||
});
|
||||
if (scenario.format === "jsonl") {
|
||||
expect(output.exitCode).toBe(1);
|
||||
const record = z
|
||||
.object({
|
||||
$rea: z.literal("jsonl/v1"),
|
||||
type: z.literal("error"),
|
||||
result: z.unknown(),
|
||||
})
|
||||
.parse(output.json);
|
||||
const failure = scenario.fullOutput
|
||||
? z
|
||||
.object({
|
||||
ok: z.literal(false),
|
||||
error: z.unknown(),
|
||||
meta: z.object({ command: z.literal(scenario.command) }),
|
||||
})
|
||||
.parse(record.result).error
|
||||
: record.result;
|
||||
expect(failure).toMatchObject({
|
||||
code: "unreadable_output",
|
||||
remediation: { action: expect.stringContaining("--format json") },
|
||||
details: { reason: expect.stringContaining("JSON Lines requires") },
|
||||
});
|
||||
return;
|
||||
}
|
||||
expect(output.exitCode).toBe(0);
|
||||
let data: unknown = output.json;
|
||||
if (scenario.fullOutput) {
|
||||
|
||||
@@ -0,0 +1,451 @@
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -634,7 +634,7 @@ describe("CLI result output surface", () => {
|
||||
describe("streamed CLI filtered selections", () => {
|
||||
it.each([
|
||||
["raw_result", "json", "null\n"],
|
||||
["raw_result", "jsonl", "null\n"],
|
||||
["raw_result", "jsonl", "JSONL_BOUNDARY_ERROR"],
|
||||
["nested.value", "json", '{"nested":{"value":null}}\n'],
|
||||
["missing", "json", "{}\n"],
|
||||
["count", "json", "0\n"],
|
||||
@@ -663,7 +663,14 @@ describe("streamed CLI filtered selections", () => {
|
||||
{ raw_result: null, nested: { value: null }, count: 0, enabled: false },
|
||||
{ command: "fixture", duration: "1ms", format },
|
||||
);
|
||||
expect(Buffer.concat(chunks).toString("utf8")).toBe(expected);
|
||||
const text = Buffer.concat(chunks).toString("utf8");
|
||||
if (expected === "JSONL_BOUNDARY_ERROR")
|
||||
expect(JSON.parse(text)).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "error",
|
||||
result: { code: "unreadable_output" },
|
||||
});
|
||||
else expect(text).toBe(expected);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
@@ -53,6 +53,15 @@ cliTest(
|
||||
});
|
||||
expect(json.stdout).not.toContain(marker);
|
||||
|
||||
const jsonl = await cli.run({
|
||||
arguments: ["--version", "--format", "jsonl"],
|
||||
cwd,
|
||||
environment,
|
||||
});
|
||||
expect(jsonl.exitCode, jsonl.stderr).toBe(0);
|
||||
expect(jsonl.stdout.trimEnd().split("\n")).toHaveLength(1);
|
||||
expect(versionReportSchema.parse(JSON.parse(jsonl.stdout))).toEqual(report);
|
||||
|
||||
const yaml = await cli.run({
|
||||
arguments: ["--version", "--format=yaml"],
|
||||
cwd,
|
||||
|
||||
@@ -209,6 +209,50 @@ describe("production MCP doctor", () => {
|
||||
});
|
||||
|
||||
describe("production MCP doctor output", () => {
|
||||
it.each([false, true])(
|
||||
"frames unhealthy JSONL diagnostics and retains the selected envelope (%s)",
|
||||
async (full) => {
|
||||
const result = await runProductionMcpDoctorCli(
|
||||
["--format", "jsonl", ...(full ? ["--full-output"] : [])],
|
||||
{
|
||||
dispatcherPath: resolve("tests/fixtures/mcpDoctorExit.mjs"),
|
||||
packageRoot: process.cwd(),
|
||||
stdoutIsTerminal: true,
|
||||
},
|
||||
);
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.output.trimEnd().split("\n")).toHaveLength(1);
|
||||
expect(JSON.parse(result.output)).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "error",
|
||||
result: full
|
||||
? {
|
||||
ok: false,
|
||||
error: { healthy: false },
|
||||
meta: { command: "rea mcp doctor" },
|
||||
}
|
||||
: { healthy: false },
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it("frames JSONL validation errors before opening a diagnostic session", async () => {
|
||||
const result = await runProductionMcpDoctorCli(
|
||||
["--format", "jsonl", "--unknown"],
|
||||
{
|
||||
dispatcherPath: "/must-not-start",
|
||||
packageRoot: process.cwd(),
|
||||
stdoutIsTerminal: false,
|
||||
},
|
||||
);
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(JSON.parse(result.output)).toMatchObject({
|
||||
$rea: "jsonl/v1",
|
||||
type: "error",
|
||||
result: { code: "VALIDATION_ERROR" },
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ name: "report", args: ["--json"] },
|
||||
{ name: "full-output envelope", args: ["--json", "--full-output"] },
|
||||
|
||||
@@ -182,6 +182,7 @@ async function runOneShot(args) {
|
||||
return;
|
||||
}
|
||||
if (output.length > 0) wroteOutput = true;
|
||||
if (resultOutput?.captureSdkOutput(output)) return;
|
||||
process.stdout.write(output);
|
||||
},
|
||||
exit: (code) => {
|
||||
@@ -198,6 +199,10 @@ async function runOneShot(args) {
|
||||
const empty = renderEmptyFilteredCliOutput(outputOptions);
|
||||
if (empty) process.stdout.write(empty);
|
||||
}
|
||||
await resultOutput?.finishSdkOutput(
|
||||
explicitExit ??
|
||||
(typeof process.exitCode === "number" ? process.exitCode : 0),
|
||||
);
|
||||
}
|
||||
} finally {
|
||||
process.stdout.write = originalStdoutWrite;
|
||||
|
||||
@@ -68,12 +68,14 @@ const runCommand = async (args) => {
|
||||
|
||||
let wroteOutput = false;
|
||||
let explicitExit;
|
||||
resultOutput?.begin(outputOptions);
|
||||
try {
|
||||
await cli.serve(cliArgs, {
|
||||
stdout: (output) => {
|
||||
if (resultOutput?.handled) return;
|
||||
if (output.length === 0) return;
|
||||
wroteOutput = true;
|
||||
if (resultOutput?.captureSdkOutput(output)) return;
|
||||
destination.write(output);
|
||||
},
|
||||
exit: (code) => {
|
||||
@@ -94,6 +96,10 @@ const runCommand = async (args) => {
|
||||
const empty = renderEmptyFilteredCliOutput(outputOptions);
|
||||
if (empty !== undefined) destination.write(empty);
|
||||
}
|
||||
await resultOutput?.finishSdkOutput(
|
||||
explicitExit ??
|
||||
(typeof process.exitCode === "number" ? process.exitCode : 0),
|
||||
);
|
||||
// logCliCommand reports some failures only through process.exitCode.
|
||||
return {
|
||||
exitCode:
|
||||
|
||||
Reference in New Issue
Block a user