diff --git a/docs/process-capture.md b/docs/process-capture.md index 3482dc18d8..2bf0d06da7 100644 --- a/docs/process-capture.md +++ b/docs/process-capture.md @@ -383,13 +383,24 @@ exit comparison like `settlement.elapsed_ms`. Every capture requires `truncation_details`, with separate accounting for: - `raw_terminal`: original UTF-8 PTY chunk bytes and observed/retained chunk - counts. A chunk that does not fit `limits.output_bytes` is omitted whole; - a later smaller chunk can still fit. + counts. A chunk that does not fit either its remaining original-text budget + or normalized-text budget is omitted whole; a later smaller chunk can still + fit. A normalized-text omission also omits its original chunk, so raw and + normalized coverage can both be incomplete for that event. +- `normalized_terminal`: normalized comparison-text bytes retained from raw + chunks. This is optional when reading historical captures. A raw-fitting + chunk is omitted if an ordered normalization stage would exceed the remaining + `output_bytes` budget before the final normalized size is known. Its final + size is unknown, not assumed to equal the budget. The capture records the + omitted-frame count and a limitation; later smaller chunks remain eligible. - `rendered_terminal`: cumulative UTF-8 serialized-state and visible-line bytes, plus observed/retained frame counts. This is independent of the raw chunk budget and does not count JSON encoding. Repeated screen snapshots can exhaust it even when all raw output fits. Rendered states use retained PTY - input, so raw omissions also limit rendered coverage. + input, so raw omissions also limit rendered coverage. When bounded + normalization omits a rendered frame before its final size is known, + `unknown_size_frames` reports that count; `observed_bytes` includes only + rendered frame sizes that were determined exactly. - `filesystem_before` and `filesystem_after`: file-count/depth limits, enumeration failures, whole-file hash budget and bytes successfully hashed. Directory names are read incrementally within the remaining entry capacity. diff --git a/src/domain/process/processCapture.validation.test.ts b/src/domain/process/processCapture.validation.test.ts index db36b2c643..4e8ec81a58 100644 --- a/src/domain/process/processCapture.validation.test.ts +++ b/src/domain/process/processCapture.validation.test.ts @@ -62,6 +62,37 @@ it("requires producer coverage accounting in every parsed capture", () => { ); }); +it("does not infer normalized retention budget from missing scenario defaults", () => { + const capture = emptyCapture(); + const scenario = { + ...capture.manifest.scenario, + limits: { + files: 10_000, + file_bytes: 10_000_000, + processes: 1_000, + filesystem_depth: 16, + }, + }; + expect(() => + parseProcessCapture({ + ...capture, + manifest: { + ...capture.manifest, + scenario, + full_scenario_sha256: digestProcessCommitment(scenario), + }, + truncation_details: { + ...capture.truncation_details, + normalized_terminal: { + budget_bytes: 1_000_000, + retained_bytes: 0, + omitted_frames: 0, + }, + }, + }), + ).toThrow("manifest.scenario.limits.output_bytes"); +}); + it("rejects settlement and cleanup combinations that cannot occur", () => { const capture = emptyCapture(); expect(() => diff --git a/src/domain/process/processCaptureCoverage.ts b/src/domain/process/processCaptureCoverage.ts index 1a0e73ee55..5a38e92409 100644 --- a/src/domain/process/processCaptureCoverage.ts +++ b/src/domain/process/processCaptureCoverage.ts @@ -8,6 +8,7 @@ const byteRetentionSchema = z.strictObject({ retained_bytes: z.number().int().nonnegative(), observed_frames: z.number().int().nonnegative(), retained_frames: z.number().int().nonnegative(), + unknown_size_frames: z.number().int().nonnegative().optional(), }); /** Filesystem enumeration and whole-file digest coverage for one checkpoint. */ @@ -54,8 +55,18 @@ export const processCaptureTruncationDetailsSchema = z.strictObject({ "UTF-8 bytes of original PTY chunks; whole chunks are retained or omitted.", ), rendered_terminal: byteRetentionSchema.describe( - "Cumulative UTF-8 bytes of serialized state plus visible lines, evaluated from retained PTY input; whole frames are retained or omitted.", + "Cumulative UTF-8 bytes of serialized state plus visible lines, evaluated from retained PTY input; whole frames are retained or omitted. When normalization exceeds a bound before the frame size is known, observed_bytes contains the exact total for size-known frames and unknown_size_frames counts the other omissions.", ), + normalized_terminal: z + .strictObject({ + budget_bytes: z.number().int().nonnegative(), + retained_bytes: z.number().int().nonnegative(), + omitted_frames: z.number().int().nonnegative(), + }) + .optional() + .describe( + "UTF-8 bytes of normalized raw-chunk text retained under output_bytes; omitted frames can have unknown final size when normalization exceeded an intermediate bound.", + ), filesystem_before: filesystemCoverageSchema, filesystem_after: filesystemCoverageSchema, process: z.strictObject({ @@ -72,6 +83,9 @@ export type ProcessCaptureTruncationDetails = z.infer< typeof processCaptureTruncationDetailsSchema >; export type TerminalRetention = ProcessCaptureTruncationDetails["raw_terminal"]; +export type NormalizedTerminalRetention = NonNullable< + ProcessCaptureTruncationDetails["normalized_terminal"] +>; export type ProcessSamplingCoverage = ProcessCaptureTruncationDetails["process"]; @@ -88,6 +102,8 @@ export const hasCaptureTruncation = ( ): boolean => omitted(details.raw_terminal) || omitted(details.rendered_terminal) || + (details.normalized_terminal?.omitted_frames ?? 0) > 0 || + (details.rendered_terminal.unknown_size_frames ?? 0) > 0 || filesystemTruncated(details.filesystem_before) || filesystemTruncated(details.filesystem_after) || details.process.sampling_partial; diff --git a/src/domain/process/processCaptureValidation.ts b/src/domain/process/processCaptureValidation.ts index 53a7837980..2c76007abe 100644 --- a/src/domain/process/processCaptureValidation.ts +++ b/src/domain/process/processCaptureValidation.ts @@ -1,4 +1,7 @@ -import { digestProcessCommitment } from "./processScenario.js"; +import { + digestProcessCommitment, + processScenarioOutputBudgetProjectionSchema, +} from "./processScenario.js"; import { hasCaptureTruncation } from "./processCaptureCoverage.js"; import type { @@ -312,8 +315,31 @@ const validateCoverage = ( retention.observed_bytes >= retention.retained_bytes && (retention.observed_frames > count || retention.observed_bytes === retention.retained_bytes) && - retention.retained_bytes <= - retention.budget_bytes, `truncation_details.${name}`, "retained observations and budget accounting do not agree"); + retention.retained_bytes <= retention.budget_bytes && + (retention.unknown_size_frames ?? 0) <= + retention.observed_frames - + retention.retained_frames, `truncation_details.${name}`, "retained observations and budget accounting do not agree"); + } + const normalized = details.normalized_terminal; + if (normalized !== undefined) { + const committedOutputBudget = + processScenarioOutputBudgetProjectionSchema.safeParse( + capture.manifest.scenario["limits"], + ); + require(committedOutputBudget.success, "manifest.scenario.limits.output_bytes", "committed process scenario must include an explicit valid output_bytes budget"); + const retainedBytes = capture.frames.reduce( + (total, frame) => total + Buffer.byteLength(frame.data), + 0, + ); + if (committedOutputBudget.success) + require(normalized.budget_bytes === + committedOutputBudget.data.output_bytes && + normalized.retained_bytes === retainedBytes && + normalized.retained_bytes <= normalized.budget_bytes && + normalized.omitted_frames <= + details.raw_terminal.observed_frames - + details.raw_terminal + .retained_frames, "truncation_details.normalized_terminal", "normalized terminal retention must agree with the committed budget and retained frames"); } for (const [name, coverage, files, checkpoint] of [ [ diff --git a/src/domain/process/processScenario.ts b/src/domain/process/processScenario.ts index e98d7a5376..bc91d5b16e 100644 --- a/src/domain/process/processScenario.ts +++ b/src/domain/process/processScenario.ts @@ -3,6 +3,10 @@ import canonicalize from "canonicalize"; import { z } from "zod"; const positiveBudget = z.number().int().safe().positive(); +/** Read the explicitly committed output budget without applying scenario defaults. */ +export const processScenarioOutputBudgetProjectionSchema = z + .object({ output_bytes: positiveBudget }) + .passthrough(); /** * Maximum xterm backing cells for one visible-plus-scrollback buffer. Each * xterm BufferLine stores three Uint32 words per cell (12 MB at this cap), diff --git a/src/process/capture/ProcessCaptureLifecycle.ts b/src/process/capture/ProcessCaptureLifecycle.ts index 0e1ca73ee3..d7ab6afbf7 100644 --- a/src/process/capture/ProcessCaptureLifecycle.ts +++ b/src/process/capture/ProcessCaptureLifecycle.ts @@ -38,11 +38,13 @@ import { assertNotCancelled } from "./ProcessScenarioRuntimeValidation.js"; import { normalizeProcessElapsedTime, normalizeProcessSamples, + type BoundedProcessTextNormalization, } from "./ProcessNormalization.js"; import { PROCESS_PROVIDER } from "../../domain/process/processEvidenceProvider.js"; import { TerminalRenderer } from "./TerminalRenderer.js"; import { hasCaptureTruncation, + type NormalizedTerminalRetention, type TerminalRetention, type ProcessCaptureTruncationDetails, } from "../../domain/process/processCaptureCoverage.js"; @@ -408,7 +410,9 @@ const captureCoverageUnknowns = ( unknowns.push({ scope: "terminal", reason: - "Raw PTY chunks exceeded output_bytes and were omitted; rendered states also lack those inputs. See truncation_details.raw_terminal.", + (details.normalized_terminal?.omitted_frames ?? 0) > 0 + ? "Raw PTY chunks were omitted because original input exceeded output_bytes or an ordered normalization stage exceeded its remaining bound; final normalized sizes for the latter omissions are unknown. Rendered states also lack those inputs. See raw_terminal and normalized_terminal retention details, including the normalized omission count." + : "Raw PTY chunks exceeded output_bytes and were omitted; rendered states also lack those inputs. See truncation_details.raw_terminal.", }); if ( details.rendered_terminal.observed_frames > @@ -417,7 +421,9 @@ const captureCoverageUnknowns = ( unknowns.push({ scope: "terminal", reason: - "Cumulative rendered state and visible-line bytes exceeded output_bytes; whole rendered frames were omitted. Raw PTY coverage is reported separately. See truncation_details.rendered_terminal.", + (details.rendered_terminal.unknown_size_frames ?? 0) > 0 + ? "Rendered frames were omitted because a normalization stage exceeded the remaining output_bytes budget; final rendered byte totals are unknown for those frames. See truncation_details.rendered_terminal." + : "Cumulative rendered state and visible-line bytes exceeded output_bytes; whole rendered frames were omitted. Raw PTY coverage is reported separately. See truncation_details.rendered_terminal.", }); for (const [name, coverage] of [ ["before", details.filesystem_before], @@ -1303,45 +1309,68 @@ export const captureTerminalFrames = (options: { readonly started: number; readonly onOutput: () => void; readonly renderer: TerminalRenderer; - readonly normalizeText: (value: string) => string; + readonly normalizeText: ( + value: string, + maxBytes: number, + ) => BoundedProcessTextNormalization; readonly recordEvent: RecordProcessCaptureEvent; -}): (() => TerminalRetention) => { - let outputBytes = 0; +}): (() => TerminalCaptureRetention) => { + let rawBytes = 0; + let normalizedBytes = 0; let observedBytes = 0; let observedFrames = 0; + let normalizedOmittedFrames = 0; options.terminal.onData((data) => { options.onOutput(); const bytes = Buffer.byteLength(data); observedBytes += bytes; observedFrames += 1; - if (outputBytes + bytes > options.scenario.limits.output_bytes) { + const budget = options.scenario.limits.output_bytes; + if (bytes > budget - rawBytes) { return; } - outputBytes += bytes; + const normalized = options.normalizeText(data, budget - normalizedBytes); + if (normalized.state === "over_budget") { + normalizedOmittedFrames += 1; + return; + } + rawBytes += bytes; + normalizedBytes += normalized.utf8Bytes; const atMs = normalizeProcessElapsedTime( monotonicProcessCaptureClock() - options.started, options.scenario.normalization.time_bucket_ms, ); - const normalized = options.normalizeText(data); const sequence = options.frames.length; options.frames.push({ sequence, at_ms: atMs, - data: normalized, - ...(normalized === data ? {} : { raw_data: data }), + data: normalized.value, + ...(normalized.value === data ? {} : { raw_data: data }), }); options.renderer.write(data, atMs); options.recordEvent("frames", sequence); }); return () => ({ - budget_bytes: options.scenario.limits.output_bytes, - observed_bytes: observedBytes, - retained_bytes: outputBytes, - observed_frames: observedFrames, - retained_frames: options.frames.length, + raw_terminal: { + budget_bytes: options.scenario.limits.output_bytes, + observed_bytes: observedBytes, + retained_bytes: rawBytes, + observed_frames: observedFrames, + retained_frames: options.frames.length, + }, + normalized_terminal: { + budget_bytes: options.scenario.limits.output_bytes, + retained_bytes: normalizedBytes, + omitted_frames: normalizedOmittedFrames, + }, }); }; +export interface TerminalCaptureRetention { + readonly raw_terminal: TerminalRetention; + readonly normalized_terminal: NormalizedTerminalRetention; +} + export const resolveProcessResult = ( capture: PendingProcessCapture | ProcessCapture | undefined, executionFailure: unknown, diff --git a/src/process/capture/ProcessCaptureObservation.test.ts b/src/process/capture/ProcessCaptureObservation.test.ts index ddf0ffd428..51ebe20fa9 100644 --- a/src/process/capture/ProcessCaptureObservation.test.ts +++ b/src/process/capture/ProcessCaptureObservation.test.ts @@ -66,7 +66,12 @@ it("returns detached terminal observations", async () => { rows: 12, scrollback: 100, maxBytes: 100_000, - normalize: (value) => value, + normalize: (value, maxBytes) => { + const utf8Bytes = Buffer.byteLength(value); + return utf8Bytes <= maxBytes + ? { state: "normalized", value, utf8Bytes } + : { state: "over_budget" }; + }, recordEvent: (collection, index) => observed.push(`${collection}:${String(index)}`), }); diff --git a/src/process/capture/ProcessCaptureTerminalRetention.test.ts b/src/process/capture/ProcessCaptureTerminalRetention.test.ts new file mode 100644 index 0000000000..e3bde2264c --- /dev/null +++ b/src/process/capture/ProcessCaptureTerminalRetention.test.ts @@ -0,0 +1,63 @@ +import { expect, it } from "vitest"; +import type { IPty } from "@lydell/node-pty"; + +import { parseProcessScenario } from "../../domain/process/processScenario.js"; +import type { TerminalFrame } from "../../domain/process/processCapture.js"; +import { captureTerminalFrames } from "./ProcessCaptureLifecycle.js"; +import { createBoundedProcessTextNormalizer } from "./ProcessNormalization.js"; + +it("omits an expanding chunk whole and retains a later smaller normalized chunk", () => { + let onData: ((data: string) => void) | undefined; + const terminal = { + onData: (listener: (data: string) => void) => { + onData = listener; + return { dispose: () => undefined }; + }, + } as unknown as IPty; + const scenario = parseProcessScenario({ + executable: "unused", + limits: { output_bytes: 1_000 }, + normalization: { + paths: false, + pids: false, + ports: false, + patterns: [{ pattern: "a", replacement: "x".repeat(500) }], + }, + }); + const frames: TerminalFrame[] = []; + const renderedChunks: string[] = []; + const journal: Array<[string, number]> = []; + const captureRetention = captureTerminalFrames({ + terminal, + scenario, + frames, + started: 0, + onOutput: () => undefined, + renderer: { write: (data: string) => renderedChunks.push(data) } as never, + normalizeText: createBoundedProcessTextNormalizer(scenario, "/tmp", -1), + recordEvent: (collection, index) => journal.push([collection, index]), + }); + + onData?.("aaa"); + onData?.("z"); + + expect(frames).toEqual([ + { sequence: 0, at_ms: expect.any(Number), data: "z" }, + ]); + expect(renderedChunks).toEqual(["z"]); + expect(journal).toEqual([["frames", 0]]); + expect(captureRetention()).toEqual({ + raw_terminal: { + budget_bytes: 1_000, + observed_bytes: 4, + retained_bytes: 1, + observed_frames: 2, + retained_frames: 1, + }, + normalized_terminal: { + budget_bytes: 1_000, + retained_bytes: 1, + omitted_frames: 1, + }, + }); +}); diff --git a/src/process/capture/ProcessHarness.ts b/src/process/capture/ProcessHarness.ts index d12e66837e..948c24a835 100644 --- a/src/process/capture/ProcessHarness.ts +++ b/src/process/capture/ProcessHarness.ts @@ -1,5 +1,4 @@ import type { IPty } from "@lydell/node-pty"; -import type { TerminalRetention } from "../../domain/process/processCaptureCoverage.js"; import type { ProcessCapture } from "../../domain/process/processCaptureParsing.js"; import type { ProcessScenario } from "../../domain/process/processScenario.js"; import { parseProcessCapture } from "../../domain/process/processCaptureParsing.js"; @@ -24,7 +23,10 @@ import { startProcessSampler } from "./ProcessSampling.js"; import { snapshotRoots } from "./FilesystemSnapshot.js"; import { TerminalRenderer } from "./TerminalRenderer.js"; import type { ProcessTimer } from "./ProcessTimer.js"; -import { createProcessTextNormalizer } from "./ProcessNormalization.js"; +import { + createBoundedProcessTextNormalizer, + type BoundedProcessTextNormalization, +} from "./ProcessNormalization.js"; import { selectCapturedProcessGroupIds } from "../ProcessOwnershipProcessTree.js"; import { awaitTerminalExit, @@ -46,6 +48,7 @@ import { describeUnverifiedCause, type ProcessCaptureResourceCleanupOptions, ProcessCaptureResourceScope, + type TerminalCaptureRetention, } from "./ProcessCaptureLifecycle.js"; import { assertNotCancelled, @@ -79,7 +82,7 @@ interface StartedCaptureRuntime { readonly startedAt: Date; readonly executableIdentity: ReturnType; readonly lastOutput: () => number; - readonly rawTerminalRetention: () => TerminalRetention; + readonly terminalRetention: () => TerminalCaptureRetention; readonly stopSampler: ReturnType; /** Signals the captured root only while its launch-time start identity still matches. */ readonly signalRoot: (signal: "SIGTERM" | "SIGKILL") => Promise<{ @@ -222,7 +225,10 @@ const cleanupFailedStartup = async (options: { const createTerminalRenderer = ( scenario: ProcessScenario, - normalizeText: (value: string) => string, + normalizeText: ( + value: string, + maxBytes: number, + ) => BoundedProcessTextNormalization, recordEvent: RecordProcessCaptureEvent, ): TerminalRenderer => new TerminalRenderer({ @@ -269,7 +275,7 @@ const startCaptureRuntime = async ( const startedAt = new Date(); const started = monotonicProcessCaptureClock(); let lastOutput = started; - const normalizeText = createProcessTextNormalizer( + const normalizeText = createBoundedProcessTextNormalizer( scenario, options.temporaryRoot, () => terminal?.pid ?? -1, @@ -295,7 +301,7 @@ const startCaptureRuntime = async ( scenario.executable, selectedExecutable, ); - const rawTerminalRetention = captureTerminalFrames({ + const terminalRetention = captureTerminalFrames({ ...options, terminal, started, @@ -324,7 +330,7 @@ const startCaptureRuntime = async ( startedAt, executableIdentity, lastOutput: () => lastOutput, - rawTerminalRetention, + terminalRetention, stopSampler, signalRoot: retainRootSignaller( terminal.pid, @@ -575,7 +581,7 @@ const completeCapture = async (options: { before: options.before, after, truncationDetails: { - raw_terminal: runtime.rawTerminalRetention(), + ...runtime.terminalRetention(), rendered_terminal: runtime.renderer.retention(), filesystem_before: options.before.coverage, filesystem_after: after.coverage, diff --git a/src/process/capture/ProcessNormalization.test.ts b/src/process/capture/ProcessNormalization.test.ts index 54b042dd40..e23ced1ca9 100644 --- a/src/process/capture/ProcessNormalization.test.ts +++ b/src/process/capture/ProcessNormalization.test.ts @@ -1,6 +1,7 @@ import { expect, it } from "vitest"; import { + createBoundedProcessTextNormalizer, createProcessTextNormalizer, normalizeProcessSamples, normalizeProcessText, @@ -95,6 +96,178 @@ it("replaces caller-declared patterns literally, without expanding replacement p ).toBe("listening on $&$&; $$; $1=alpha; tail $1"); }); +it("bounds ordered normalization before replacement expansion and preserves later smaller chunks", () => { + const normalize = createBoundedProcessTextNormalizer( + scenario({ + paths: false, + ports: false, + pids: false, + patterns: [{ pattern: "a", replacement: "x".repeat(500) }], + }), + "/temporary", + rootPid, + ); + + expect(normalize("a".repeat(100), 1_000)).toEqual({ state: "over_budget" }); + expect(normalize("z", 1_000)).toEqual({ + state: "normalized", + value: "z", + utf8Bytes: 1, + }); +}); + +it("keeps ordered intermediate expansion explicit even when a later rule could shrink it", () => { + const normalize = createBoundedProcessTextNormalizer( + scenario({ + paths: false, + ports: false, + pids: false, + patterns: [ + { pattern: "a", replacement: "x".repeat(20) }, + { pattern: "x", replacement: "y" }, + ], + }), + "/temporary", + rootPid, + ); + + expect(normalize("a", 10)).toEqual({ state: "over_budget" }); +}); + +it("counts UTF-8 bytes across replacement boundaries like the final JavaScript string", () => { + const splitPair = scenario({ + paths: false, + ports: false, + pids: false, + patterns: [{ pattern: "\ude00", replacement: "\ude01" }], + }); + const splitPairResult = createBoundedProcessTextNormalizer( + splitPair, + "/temporary", + rootPid, + )("😀", 4); + expect(splitPairResult).toEqual({ + state: "normalized", + value: normalizeProcessText("😀", splitPair, "/temporary", rootPid), + utf8Bytes: 4, + }); + expect( + splitPairResult.state === "normalized" && splitPairResult.utf8Bytes, + ).toBe( + Buffer.byteLength( + normalizeProcessText("😀", splitPair, "/temporary", rootPid), + ), + ); + + const joinedPair = scenario({ + paths: false, + ports: false, + pids: false, + patterns: [{ pattern: "\ude00\ud83d", replacement: "" }], + }); + const joinedInput = "😀😀"; + const joinedExpected = normalizeProcessText( + joinedInput, + joinedPair, + "/temporary", + rootPid, + ); + expect(joinedExpected).toBe("😀"); + expect( + createBoundedProcessTextNormalizer( + joinedPair, + "/temporary", + rootPid, + )(joinedInput, Buffer.byteLength(joinedExpected)), + ).toEqual({ + state: "normalized", + value: joinedExpected, + utf8Bytes: Buffer.byteLength(joinedExpected), + }); + + const emptyPattern = scenario({ + paths: false, + ports: false, + pids: false, + patterns: [{ pattern: "", replacement: "\ude00" }], + }); + const input = "\ud83d"; + const expected = normalizeProcessText( + input, + emptyPattern, + "/temporary", + rootPid, + ); + expect( + createBoundedProcessTextNormalizer( + emptyPattern, + "/temporary", + rootPid, + )(input, Buffer.byteLength(expected)), + ).toEqual({ + state: "normalized", + value: expected, + utf8Bytes: Buffer.byteLength(expected), + }); + expect( + createBoundedProcessTextNormalizer( + emptyPattern, + "/temporary", + rootPid, + )(input, Buffer.byteLength(expected) - 1), + ).toEqual({ state: "over_budget" }); + + const loneSurrogate = scenario({ + paths: false, + ports: false, + pids: false, + patterns: [{ pattern: "x", replacement: "y" }], + }); + const loneInput = "\ud83dx"; + const loneExpected = normalizeProcessText( + loneInput, + loneSurrogate, + "/temporary", + rootPid, + ); + expect( + createBoundedProcessTextNormalizer( + loneSurrogate, + "/temporary", + rootPid, + )(loneInput, Buffer.byteLength(loneExpected)), + ).toEqual({ + state: "normalized", + value: loneExpected, + utf8Bytes: Buffer.byteLength(loneExpected), + }); +}); + +it("lets a later pattern shrink text when an earlier pattern does not match", () => { + const declared = scenario({ + paths: false, + ports: false, + pids: false, + patterns: [ + { pattern: "z", replacement: "x" }, + { pattern: "aa", replacement: "b" }, + ], + }); + const expected = normalizeProcessText("aa", declared, "/temporary", rootPid); + expect(expected).toBe("b"); + expect( + createBoundedProcessTextNormalizer( + declared, + "/temporary", + rootPid, + )("aa", Buffer.byteLength(expected)), + ).toEqual({ + state: "normalized", + value: expected, + utf8Bytes: Buffer.byteLength(expected), + }); +}); + it("reuses matchers while observing the terminal PID after launch", () => { let pid = -1; const normalize = createProcessTextNormalizer( diff --git a/src/process/capture/ProcessNormalization.ts b/src/process/capture/ProcessNormalization.ts index 85dc409295..1f5402993e 100644 --- a/src/process/capture/ProcessNormalization.ts +++ b/src/process/capture/ProcessNormalization.ts @@ -3,6 +3,165 @@ import { isIP } from "node:net"; import type { ProcessScenario } from "../../domain/process/processScenario.js"; import type { ProcessSample } from "../../domain/process/processCapture.js"; + +export type BoundedProcessTextNormalization = + | { + readonly state: "normalized"; + readonly value: string; + readonly utf8Bytes: number; + } + | { readonly state: "over_budget" }; + +type RegexReplacement = (match: string, ...groups: string[]) => string; + +type BoundedStageResult = + | { + readonly state: "bounded"; + readonly value: string; + readonly utf8Bytes: number; + } + | { readonly state: "unchanged"; readonly value: string } + | { readonly state: "over_budget" }; + +const createUtf8ByteCounter = (maxBytes: number) => { + let bytes = 0; + let pendingHighSurrogate = false; + const addBytes = (added: number): void => { + if (bytes > maxBytes || added > maxBytes - bytes) { + bytes = Number.POSITIVE_INFINITY; + return; + } + bytes += added; + }; + return { + add: (value: string): void => { + if (bytes > maxBytes || value.length === 0) return; + let index = 0; + if (pendingHighSurrogate) { + if (value.length > 0 && isLowSurrogate(value.charCodeAt(0))) { + addBytes(4); + index = 1; + } else { + addBytes(3); + } + pendingHighSurrogate = false; + } + for (; index < value.length; index += 1) { + const code = value.charCodeAt(index); + if (code <= 0x7f) addBytes(1); + else if (code <= 0x7ff) addBytes(2); + else if (isHighSurrogate(code)) { + if (index + 1 === value.length) { + pendingHighSurrogate = true; + } else if (isLowSurrogate(value.charCodeAt(index + 1))) { + addBytes(4); + index += 1; + } else addBytes(3); + } else addBytes(3); + if (bytes > maxBytes) return; + } + }, + finish: (): number => { + if (pendingHighSurrogate) addBytes(3); + return bytes; + }, + }; +}; + +const isHighSurrogate = (code: number): boolean => + code >= 0xd800 && code <= 0xdbff; +const isLowSurrogate = (code: number): boolean => + code >= 0xdc00 && code <= 0xdfff; + +const boundedRegexReplace = ( + value: string, + matcher: RegExp, + replacement: RegexReplacement, + maxBytes: number, +): BoundedStageResult => { + const byteCounter = createUtf8ByteCounter(maxBytes); + let cursor = 0; + let changed = false; + matcher.lastIndex = 0; + let match: RegExpExecArray | null; + while ((match = matcher.exec(value)) !== null) { + const index = match.index; + const matched = match[0] ?? ""; + const nextCursor = index + matched.length; + const replacementValue = replacement(matched, ...match.slice(1)); + changed ||= replacementValue !== matched; + byteCounter.add(value.slice(cursor, index)); + byteCounter.add(replacementValue); + cursor = nextCursor; + } + byteCounter.add(value.slice(cursor)); + const bytes = byteCounter.finish(); + if (bytes > maxBytes) { + return changed ? { state: "over_budget" } : { state: "unchanged", value }; + } + if (!changed) return { state: "unchanged", value }; + matcher.lastIndex = 0; + const normalized = value.replace(matcher, replacement); + return { + state: "bounded", + value: normalized, + utf8Bytes: bytes, + }; +}; + +const boundedLiteralReplace = ( + value: string, + pattern: string, + replacement: string, + maxBytes: number, +): BoundedStageResult => { + const byteCounter = createUtf8ByteCounter(maxBytes); + let cursor = 0; + if (pattern === "") { + if (replacement === "") return { state: "unchanged", value }; + for (let index = 0; index < value.length; index += 1) { + byteCounter.add(replacement); + byteCounter.add(value.slice(index, index + 1)); + } + byteCounter.add(replacement); + const bytes = byteCounter.finish(); + if (bytes > maxBytes) return { state: "over_budget" }; + return { + state: "bounded", + value: value.replaceAll(pattern, () => replacement), + utf8Bytes: bytes, + }; + } + if (pattern === replacement) return { state: "unchanged", value }; + + let index = value.indexOf(pattern, cursor); + if (index === -1) return { state: "unchanged", value }; + while (index !== -1) { + byteCounter.add(value.slice(cursor, index)); + byteCounter.add(replacement); + cursor = index + pattern.length; + index = value.indexOf(pattern, cursor); + } + byteCounter.add(value.slice(cursor)); + const bytes = byteCounter.finish(); + if (bytes > maxBytes) return { state: "over_budget" }; + return { + state: "bounded", + value: value.replaceAll(pattern, () => replacement), + utf8Bytes: bytes, + }; +}; + +const boundedIdentity = ( + value: string, + maxBytes: number, +): BoundedStageResult => { + const utf8Bytes = Buffer.byteLength(value); + return utf8Bytes <= maxBytes + ? { state: "bounded", value, utf8Bytes } + : { state: "unchanged", value }; +}; + /** Bucket one elapsed process-capture timestamp under scenario normalization. */ export const normalizeProcessElapsedTime = ( elapsedMs: number, @@ -41,6 +200,84 @@ export const createProcessTextNormalizer = ( }; }; +/** Normalize terminal text without materializing an intermediate over its byte budget. */ +export const createBoundedProcessTextNormalizer = ( + scenario: ProcessScenario, + temporaryRoot: string, + pid: number | (() => number), +): ((value: string, maxBytes: number) => BoundedProcessTextNormalization) => { + const pathRule = scenario.normalization.paths + ? createPathRule(scenario, temporaryRoot) + : undefined; + let previousPid: number | undefined; + let pidMatcher: RegExp | undefined; + return (value, maxBytes) => { + let normalized = value; + const apply = (result: BoundedStageResult): boolean => { + if (result.state === "over_budget") return false; + if (result.state === "bounded") normalized = result.value; + return true; + }; + if (pathRule !== undefined) { + if ( + !apply( + boundedRegexReplace( + normalized, + pathRule.matcher, + pathRule.replace, + maxBytes, + ), + ) + ) + return { state: "over_budget" }; + } + if (scenario.normalization.ports) { + if (!apply(normalizePortsBounded(normalized, maxBytes))) + return { state: "over_budget" }; + } + if (scenario.normalization.pids) { + const currentPid = typeof pid === "number" ? pid : pid(); + if (currentPid !== previousPid) { + previousPid = currentPid; + pidMatcher = createPidMatcher([currentPid]); + } + if (pidMatcher !== undefined) { + if ( + !apply( + boundedRegexReplace( + normalized, + pidMatcher, + () => "", + maxBytes, + ), + ) + ) + return { state: "over_budget" }; + } + } + for (const pattern of scenario.normalization.patterns) { + if ( + !apply( + boundedLiteralReplace( + normalized, + pattern.pattern, + pattern.replacement, + maxBytes, + ), + ) + ) + return { state: "over_budget" }; + } + return boundedIdentity(normalized, maxBytes).state === "bounded" + ? { + state: "normalized", + value: normalized, + utf8Bytes: Buffer.byteLength(normalized), + } + : { state: "over_budget" }; + }; +}; + /** Normalize one payload when its caller does not retain a capture-local matcher. */ export const normalizeProcessText = ( value: string, @@ -50,10 +287,12 @@ export const normalizeProcessText = ( ): string => createProcessTextNormalizer(scenario, temporaryRoot, pid)(value); /** Match original paths once, preferring the most specific selected identity. */ -const createPathNormalizer = ( +const createPathRule = ( scenario: ProcessScenario, temporaryRoot: string, -): ((value: string) => string) => { +): + | { readonly matcher: RegExp; readonly replace: RegexReplacement } + | undefined => { const aliases = new Map(); const selected: readonly (readonly [string, string])[] = [ [temporaryRoot, ""], @@ -71,12 +310,24 @@ const createPathNormalizer = ( const paths = [...aliases.keys()].sort( (left, right) => right.length - left.length, ); - if (paths.length === 0) return (value) => value; + if (paths.length === 0) return undefined; const pattern = paths .map((path) => path.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&")) .join("|"); const matcher = new RegExp(pattern, "gu"); - return (value) => value.replace(matcher, (path) => aliases.get(path) ?? path); + return { + matcher, + replace: (path) => aliases.get(path) ?? path, + }; +}; + +const createPathNormalizer = ( + scenario: ProcessScenario, + temporaryRoot: string, +): ((value: string) => string) => { + const rule = createPathRule(scenario, temporaryRoot); + if (rule === undefined) return (value) => value; + return (value) => value.replace(rule.matcher, rule.replace); }; // A port may end a sentence, but a period followed by a letter, digit, or @@ -85,21 +336,45 @@ const createPathNormalizer = ( // PTY chunks are normalized independently, so a period that ends the chunk // stays unnormalized: its continuation may arrive in the next chunk. const normalizePortTokens = (value: string): string => { - const endpointNormalized = value.replaceAll( - /(\b[a-z][a-z0-9+.-]*:\/\/[^\s/?#"'<>]+|\blocalhost|\b(?:\d{1,3}\.){3}\d{1,3}|\[[\da-f:.%]+\]):(\d{1,5})(?![\p{L}\p{N}_@:]|\.(?:[\p{L}\p{N}_]|$))/giu, - (match: string, endpoint: string, port: string) => { - if (Number(port) > 65_535) return match; - const isEndpoint = endpoint.includes("://") - ? URL.canParse(`${endpoint}:${port}`) - : endpoint.toLowerCase() === "localhost" || - isIP(endpoint.replace(/^\[|\]$/gu, "")) !== 0; - return isEndpoint ? `${endpoint}:` : match; - }, + return value + .replaceAll(PORT_ENDPOINT_MATCHER, portEndpointReplacement) + .replaceAll(PORT_FIELD_MATCHER, portFieldReplacement); +}; + +const PORT_ENDPOINT_MATCHER = + /(\b[a-z][a-z0-9+.-]*:\/\/[^\s/?#"'<>]+|\blocalhost|\b(?:\d{1,3}\.){3}\d{1,3}|\[[\da-f:.%]+\]):(\d{1,5})(?![\p{L}\p{N}_@:]|\.(?:[\p{L}\p{N}_]|$))/giu; +const PORT_FIELD_MATCHER = + /(\b(?:port|tcp_port|udp_port|listen)\b["']?\s*[:=]\s*)(\d{1,5})(?![\p{L}\p{N}_]|\.(?:[\p{L}\p{N}_]|$))/giu; + +const portEndpointReplacement: RegexReplacement = (match, endpoint, port) => { + if (Number(port) > 65_535) return match; + const isEndpoint = endpoint.includes("://") + ? URL.canParse(`${endpoint}:${port}`) + : endpoint.toLowerCase() === "localhost" || + isIP(endpoint.replace(/^\[|\]$/gu, "")) !== 0; + return isEndpoint ? `${endpoint}:` : match; +}; + +const portFieldReplacement: RegexReplacement = (match, prefix, port) => + Number(port) <= 65_535 ? `${prefix}` : match; + +const normalizePortsBounded = ( + value: string, + maxBytes: number, +): BoundedStageResult => { + const endpoint = boundedRegexReplace( + value, + PORT_ENDPOINT_MATCHER, + portEndpointReplacement, + maxBytes, ); - return endpointNormalized.replaceAll( - /(\b(?:port|tcp_port|udp_port|listen)\b["']?\s*[:=]\s*)(\d{1,5})(?![\p{L}\p{N}_]|\.(?:[\p{L}\p{N}_]|$))/giu, - (match: string, prefix: string, port: string) => - Number(port) <= 65_535 ? `${prefix}` : match, + if (endpoint.state === "over_budget") return endpoint; + const afterEndpoint = endpoint.state === "bounded" ? endpoint.value : value; + return boundedRegexReplace( + afterEndpoint, + PORT_FIELD_MATCHER, + portFieldReplacement, + maxBytes, ); }; @@ -161,11 +436,18 @@ export const normalizeProcessSamples = ( const createPidNormalizer = ( identifiers: readonly number[], ): ((value: string) => string) => { + const matcher = createPidMatcher(identifiers); + if (matcher === undefined) return (value) => value; + return (value) => value.replace(matcher, ""); +}; + +const createPidMatcher = ( + identifiers: readonly number[], +): RegExp | undefined => { const tokens = [...new Set(identifiers)] .filter((identifier) => Number.isSafeInteger(identifier) && identifier > 0) .sort((left, right) => right - left) .map(String); - if (tokens.length === 0) return (value) => value; - const matcher = new RegExp(`(? value.replace(matcher, ""); + if (tokens.length === 0) return undefined; + return new RegExp(`(? string, + ): ((value: string, maxBytes: number) => BoundedProcessTextNormalization) => + (value, maxBytes) => { + const normalized = normalize(value); + const utf8Bytes = Buffer.byteLength(normalized); + return utf8Bytes <= maxBytes + ? { state: "normalized", value: normalized, utf8Bytes } + : { state: "over_budget" }; + }; + const renderFrames = async ( data: string, normalize: (value: string) => string = (value) => value, @@ -22,7 +37,7 @@ const renderFrames = async ( rows: 4, scrollback: 10, maxBytes: 100_000, - normalize, + normalize: boundedNormalize(normalize), }); renderer.write(data, 0); const frames = await renderer.frames(); @@ -56,7 +71,7 @@ it("applies rendered retention cumulatively across sparse frames", async () => { rows: 1, scrollback: 0, maxBytes: 5, - normalize: (value) => value, + normalize: boundedNormalize((value) => value), }); try { renderer.write("a", 0); @@ -66,10 +81,11 @@ it("applies rendered retention cumulatively across sparse frames", async () => { expect(frames.map(({ lines }) => lines)).toEqual([["a"]]); expect(renderer.retention()).toEqual({ budget_bytes: 5, - observed_bytes: 6, + observed_bytes: 2, retained_bytes: 2, observed_frames: 2, retained_frames: 1, + unknown_size_frames: 1, }); expect(renderer.truncated()).toBe(true); } finally { @@ -84,6 +100,44 @@ it("trims only after normalization sees the full-width row", async () => { expect(frames[0]?.lines[0]).toBe("pid "); }); +it("omits rendered frames with unknown expanded size and keeps later bounded states", async () => { + const renderer = new TerminalRenderer({ + columns: 10, + rows: 1, + scrollback: 0, + maxBytes: 1_000, + normalize: createBoundedProcessTextNormalizer( + parseProcessScenario({ + executable: "unused", + normalization: { + paths: false, + pids: false, + ports: false, + patterns: [{ pattern: "a", replacement: "x".repeat(2_000) }], + }, + }), + "/tmp", + -1, + ), + }); + try { + renderer.write("a", 0); + renderer.write("\u001b[2J\rz", 1); + const frames = await renderer.frames(); + + expect(frames).toHaveLength(1); + expect(frames[0]?.lines[0]).toBe("z"); + expect(renderer.retention()).toMatchObject({ + observed_frames: 2, + retained_frames: 1, + unknown_size_frames: 1, + }); + expect(renderer.truncated()).toBe(true); + } finally { + await renderer.dispose(); + } +}); + it.each(["write", "resize"] as const)( "releases terminal resources after a queued %s observation fails", async (operation) => { @@ -95,9 +149,9 @@ it.each(["write", "resize"] as const)( rows: 4, scrollback: 10, maxBytes: 100_000, - normalize: () => { + normalize: boundedNormalize(() => { throw failure; - }, + }), }); const scope = new ProcessCaptureResourceScope(); const temporaryRoot = await mkdtemp( @@ -147,7 +201,7 @@ it("retains disposal failures for a later cleanup retry", async () => { rows: 4, scrollback: 10, maxBytes: 100_000, - normalize: (value) => value, + normalize: boundedNormalize((value) => value), }); const scope = new ProcessCaptureResourceScope(); const temporaryRoot = await mkdtemp(join(tmpdir(), "rea-terminal-retry-")); diff --git a/src/process/capture/TerminalRenderer.ts b/src/process/capture/TerminalRenderer.ts index 7ec2e10e5c..06598bf86c 100644 --- a/src/process/capture/TerminalRenderer.ts +++ b/src/process/capture/TerminalRenderer.ts @@ -1,5 +1,6 @@ import { createRequire } from "node:module"; import type { TerminalRetention } from "../../domain/process/processCaptureCoverage.js"; +import type { BoundedProcessTextNormalization } from "./ProcessNormalization.js"; import type { RecordProcessCaptureEvent, @@ -21,7 +22,10 @@ interface TerminalRendererOptions { readonly rows: number; readonly scrollback: number; readonly maxBytes: number; - readonly normalize: (value: string) => string; + readonly normalize: ( + value: string, + maxBytes: number, + ) => BoundedProcessTextNormalization; readonly recordEvent?: RecordProcessCaptureEvent; } @@ -42,6 +46,7 @@ export class TerminalRenderer { #observedBytes = 0; #observedFrames = 0; #truncated = false; + #unknownSizeFrames = 0; constructor(private readonly options: TerminalRendererOptions) { this.#terminal = new HeadlessPackage.Terminal({ @@ -97,6 +102,9 @@ export class TerminalRenderer { retained_bytes: this.#capturedBytes, observed_frames: this.#observedFrames, retained_frames: this.#frames.length, + ...(this.#unknownSizeFrames === 0 + ? {} + : { unknown_size_frames: this.#unknownSizeFrames }), }; } @@ -118,32 +126,50 @@ export class TerminalRenderer { #capture(atMs: number): void { const buffer = this.#terminal.buffer.active; const lines: string[] = []; + this.#observedFrames += 1; + let remainingBytes = this.options.maxBytes - this.#capturedBytes; + let bytes = 0; for (let row = 0; row < this.#terminal.rows; row += 1) { const line = buffer.getLine(buffer.viewportY + row); + const paddedLine = ( + line?.translateToString(false, 0, this.#terminal.cols) ?? "" + ).padEnd(this.#terminal.cols, " "); + const trailingSpaceBytes = Buffer.byteLength( + paddedLine.match(TRAILING_SPACES)?.[0] ?? "", + ); + const normalized = this.options.normalize( + paddedLine, + Math.min(Number.MAX_SAFE_INTEGER, remainingBytes + trailingSpaceBytes), + ); + if (normalized.state === "over_budget") { + this.#unknownSizeFrames += 1; + this.#truncated = true; + return; + } + const normalizedLine = normalized.value.replace(TRAILING_SPACES, ""); + const lineBytes = Buffer.byteLength(normalizedLine); + if (lineBytes > remainingBytes) { + this.#unknownSizeFrames += 1; + this.#truncated = true; + return; + } + remainingBytes -= lineBytes; + bytes += lineBytes; // Trailing blank cells carry no information beyond `columns`; dropping // them keeps sparse screens from costing a full row per blank line. - lines.push( - this.options - .normalize( - ( - line?.translateToString(false, 0, this.#terminal.cols) ?? "" - ).padEnd(this.#terminal.cols, " "), - ) - .replace(TRAILING_SPACES, ""), - ); + lines.push(normalizedLine); } - const serializedState = this.options.normalize( + const normalizedState = this.options.normalize( this.#serializeAddon.serialize(), + remainingBytes, ); - const bytes = - Buffer.byteLength(serializedState) + - lines.reduce((total, line) => total + Buffer.byteLength(line), 0); - this.#observedBytes += bytes; - this.#observedFrames += 1; - if (this.#capturedBytes + bytes > this.options.maxBytes) { + if (normalizedState.state === "over_budget") { + this.#unknownSizeFrames += 1; this.#truncated = true; return; } + bytes += normalizedState.utf8Bytes; + this.#observedBytes += bytes; this.#capturedBytes += bytes; const sequence = this.#frames.length; this.#frames.push({ @@ -155,7 +181,7 @@ export class TerminalRenderer { cursor_y: buffer.cursorY, active_buffer: buffer.type, lines, - serialized_state: serializedState, + serialized_state: normalizedState.value, }); this.options.recordEvent?.("rendered_frames", sequence); } diff --git a/tests/boundary/process/processSessionMcp.test.ts b/tests/boundary/process/processSessionMcp.test.ts index 7717c7f4e3..41e2c25856 100644 --- a/tests/boundary/process/processSessionMcp.test.ts +++ b/tests/boundary/process/processSessionMcp.test.ts @@ -70,6 +70,98 @@ it("rejects oversized initial and resize terminal buffers through MCP before lau await client.ping(); }); +itWithCaptureCapability( + "bounds normalized PTY text and preserves equal-size output through MCP", + async () => { + const root = await createTestTempDirectory( + "rea-process-normalized-budget-", + ); + const session = createTestBinarySession(createCacheProvider([])); + const server = createServer( + { kind: "session", session }, + { logger: silentLogger }, + ); + const client = new Client({ + name: "process-normalized-budget", + version: "1", + }); + resources.push(client, server, session); + const [clientTransport, serverTransport] = + InMemoryTransport.createLinkedPair(); + await server.connect(serverTransport); + await client.connect(clientTransport); + + const capture = async (replacement: string) => { + const result = await client.callTool({ + name: "capture_process_scenario", + arguments: { + executable: process.execPath, + working_directory: root, + arguments: ["-e", 'process.stdout.write("a".repeat(100) + "\\n")'], + limits: { output_bytes: 1_000 }, + normalization: { + paths: false, + pids: false, + ports: false, + patterns: [{ pattern: "a", replacement }], + }, + settle_ms: 50, + }, + }); + expect(result.isError, JSON.stringify(result.content)).not.toBe(true); + return toolContract("capture_process_scenario").outputSchema.parse( + result.structuredContent, + ).normalized_result; + }; + + const expanded = await capture("x".repeat(500)); + expect(expanded.frames).toHaveLength(0); + expect(expanded.truncated).toBe(true); + expect(expanded.truncation_details.normalized_terminal).toMatchObject({ + budget_bytes: 1_000, + retained_bytes: 0, + omitted_frames: expect.any(Number), + }); + expect( + expanded.truncation_details.normalized_terminal?.omitted_frames, + ).toBeGreaterThan(0); + expect(expanded.truncation_details.raw_terminal).toMatchObject({ + budget_bytes: 1_000, + observed_bytes: expect.any(Number), + retained_bytes: 0, + observed_frames: expect.any(Number), + retained_frames: 0, + }); + expect( + expanded.limitations.some((limitation) => + limitation.includes( + "final normalized sizes for the latter omissions are unknown", + ), + ), + ).toBe(true); + expect(expanded.cleanup).toMatchObject({ + owned_process_group: "verified", + temporary_root: "removed", + }); + + const equalSize = await capture("b"); + expect(equalSize.truncated).toBe(false); + expect(equalSize.frames.map(({ data }) => data).join("")).toBe( + `${"b".repeat(100)}\r\n`, + ); + expect(equalSize.truncation_details.normalized_terminal).toMatchObject({ + budget_bytes: 1_000, + retained_bytes: Buffer.byteLength(`${"b".repeat(100)}\r\n`), + omitted_frames: 0, + }); + expect(equalSize.cleanup).toMatchObject({ + owned_process_group: "verified", + temporary_root: "removed", + }); + }, + 30_000, +); + itWithCaptureCapability( "records process residuals linked to capture Evidence", async () => {