fix: normalize retained terminal observations with sampled identities

This commit is contained in:
morluto committed 2026-10-24 04:23:56 +08:00
1 parent 9877ab7d1d
commit 20ff68914b
13 files changed
+647 -102

No files matched your search

+17
View File
@@ -283,6 +283,23 @@ while comparisons and trace assertions use the normalized text. Retaining the
original adds at most `limits.output_bytes` of text, because it comes from the
same admitted PTY chunks rather than a second unbounded stream.
When PID normalization is enabled, the root and identifiers in retained process
samples use the same identity set in terminal text and sampled commands. Raw
PTY chunks emitted before a child was sampled are projected again from their
retained source after sampling stops; custom patterns still run after PID
replacement. Earlier declared normalization omissions remain explicit; omitted
chunks are not reconstructed. Numbers not in that observed set remain literal.
Captures commit
`pid_normalization_version: "sampled-identifiers-v2"`, so older root-only terminal
normalization is not silently compared as equivalent.
Headless rendering runs once at terminal finalization from the already retained,
byte-bounded PTY chunks, preserving their elapsed times and resize order. Parsing
latency therefore occurs during finalization; rendered journal entries record
that derived observation. A corrected raw-text projection that exceeds
`limits.output_bytes` returns a typed capacity failure with collected facts in
partial observations. No additional raw-screen copy or replay is retained.
When path normalization is enabled, aliases apply to selected path identities
at clear text delimiters. A selected executable path does not alias a longer
sibling filename such as `tool-helper`; selected parent directories still
+8 -2
View File
@@ -148,17 +148,23 @@ it("keeps scenario identity without finalization while versioning corrected port
).toBe("b50005d26269bd1db4a49c79e05bca67c3a18d837b45cd1c8ff591c01eb7623a");
const comparison = processComparisonContract(scenario);
expect(comparison.port_normalization_version).toBe("contextual-endpoints-v3");
const { pid_normalization_version: pidVersion, ...rootOnlyComparison } =
comparison;
expect(pidVersion).toBe("sampled-identifiers-v2");
expect(
digestProcessCommitment({
...comparison,
...rootOnlyComparison,
port_normalization_version: "contextual-endpoints-v2",
}),
"the earlier comparison contract had no finalization field",
).toBe("675b759347a737a4f1cbfbf2fc759258d9c877216513f2f4dad3ed1e9e6e7631");
expect(
digestProcessCommitment(comparison),
digestProcessCommitment(rootOnlyComparison),
"corrected endpoint normalization has its own comparison commitment",
).toBe("301f70ed68bd81b67d07c88b44c85ed0cf7d583967d2f769e2cb193d4db61c20");
expect(digestProcessCommitment(comparison)).not.toBe(
digestProcessCommitment(rootOnlyComparison),
);
});
it("admits the default and the exact combined terminal-cell budget", () => {
+3
View File
@@ -163,6 +163,9 @@ export const processComparisonContract = (
: {}),
limits: scenario.limits,
normalization: scenario.normalization,
...(scenario.normalization.pids
? { pid_normalization_version: "sampled-identifiers-v2" }
: {}),
...(scenario.normalization.paths
? { path_normalization_version: "path-component-boundaries-v2" }
: {}),
+33 -14
View File
@@ -52,6 +52,7 @@ import { assertNotCancelled } from "./ProcessScenarioRuntimeValidation.js";
import {
normalizeProcessElapsedTime,
normalizeProcessSamples,
normalizeRetainedTerminalFrames,
type BoundedProcessTextNormalization,
} from "./ProcessNormalization.js";
import { PROCESS_PROVIDER } from "../../domain/process/processEvidenceProvider.js";
@@ -1439,12 +1440,14 @@ export const captureTerminalFrames = (options: {
maxBytes: number,
) => BoundedProcessTextNormalization;
readonly recordEvent: RecordProcessCaptureEvent;
readonly identifiers: () => readonly number[];
}): (() => TerminalCaptureRetention) => {
let rawBytes = 0;
let normalizedBytes = 0;
let observedBytes = 0;
let observedFrames = 0;
let normalizedOmittedFrames = 0;
let firstIdentifiers: readonly number[] | undefined;
options.terminal.onData((data) => {
options.onOutput();
const bytes = Buffer.byteLength(data);
@@ -1459,6 +1462,7 @@ export const captureTerminalFrames = (options: {
normalizedOmittedFrames += 1;
return;
}
firstIdentifiers ??= options.identifiers();
rawBytes += bytes;
normalizedBytes += normalized.utf8Bytes;
const atMs = normalizeProcessElapsedTime(
@@ -1475,20 +1479,35 @@ export const captureTerminalFrames = (options: {
options.renderer.write(data, atMs);
options.recordEvent("frames", sequence);
});
return () => ({
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,
},
});
return () => {
const identifiers = options.identifiers();
if (
options.scenario.normalization.pids &&
firstIdentifiers !== undefined &&
identifiers !== firstIdentifiers
) {
normalizedBytes = normalizeRetainedTerminalFrames(
options.frames,
options.normalizeText,
options.scenario.limits.output_bytes,
);
firstIdentifiers = identifiers;
}
return {
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 {
@@ -2,7 +2,12 @@ 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 { AnalysisResourceConstraintError } from "../../domain/analysisErrorCore.js";
import { createProcessIdentifierLookup } from "./ProcessNormalization.js";
import type {
TerminalFrame,
ProcessSample,
} from "../../domain/process/processCapture.js";
import { captureTerminalFrames } from "./ProcessCaptureLifecycle.js";
import { createBoundedProcessTextNormalizer } from "./ProcessNormalization.js";
@@ -36,6 +41,7 @@ it("omits an expanding chunk whole and retains a later smaller normalized chunk"
renderer: { write: (data: string) => renderedChunks.push(data) } as never,
normalizeText: createBoundedProcessTextNormalizer(scenario, "/tmp", -1),
recordEvent: (collection, index) => journal.push([collection, index]),
identifiers: () => [],
});
onData?.("aaa");
@@ -61,3 +67,74 @@ it("omits an expanding chunk whole and retains a later smaller normalized chunk"
},
});
});
it.each([false, true])(
"reprojects bounded retained raw chunks after sampled child discovery, capacity failure=%s",
(overBudget) => {
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: overBudget ? 4 : 1000 },
normalization: {
paths: false,
ports: false,
pids: true,
patterns: overBudget
? []
: [
{ pattern: "<pid>", replacement: "selected" },
{ pattern: "200", replacement: "wrong-order" },
],
},
});
const samples: ProcessSample[] = [];
const frames: TerminalFrame[] = [];
const identifiers = createProcessIdentifierLookup(samples, () => 100);
const readRetention = captureTerminalFrames({
terminal,
scenario,
frames,
started: 0,
onOutput: () => undefined,
renderer: { write: () => undefined } as never,
normalizeText: createBoundedProcessTextNormalizer(
scenario,
"/tmp",
identifiers,
),
identifiers,
recordEvent: () => undefined,
});
const original = overBudget ? "20" : "root=100 child=200 other=201";
onData?.(original);
const first = structuredClone(frames);
samples.push({
at_ms: 1,
pid: overBudget ? 20 : 200,
parent_pid: 100,
process_group_id: 100,
session_id: 100,
command: "child",
});
if (overBudget) {
expect(readRetention).toThrow(AnalysisResourceConstraintError);
expect(frames).toEqual(first);
expect(frames[0]?.data).toBe("20");
} else {
const retention = readRetention();
expect(frames[0]?.data).toBe("root=selected child=selected other=201");
expect(frames[0]?.raw_data).toBe(original);
expect(frames[0]?.at_ms).toBe(first[0]?.at_ms);
expect(retention.normalized_terminal.retained_bytes).toBe(
Buffer.byteLength(frames[0]!.data),
);
expect(readRetention()).toEqual(retention);
}
},
);
+39 -43
View File
@@ -5,6 +5,8 @@ import { parseProcessCapture } from "../../domain/process/processCaptureParsing.
import { err, ok, type Result } from "../../domain/result.js";
import { AnalysisCapabilityUnavailableError } from "../../domain/analysisErrorCore.js";
import { AnalysisError } from "../../domain/analysisErrorBase.js";
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
import { createProcessTerminalFinalizer } from "./ProcessTerminalObservation.js";
import { analysisErrorWithCleanupFailure } from "../../domain/analysisErrorCleanup.js";
import {
describeProcessCaptureExecutionFailure,
@@ -29,6 +31,7 @@ import { TerminalRenderer } from "./TerminalRenderer.js";
import type { ProcessTimer } from "./ProcessTimer.js";
import {
createBoundedProcessTextNormalizer,
createProcessIdentifierLookup,
type BoundedProcessTextNormalization,
} from "./ProcessNormalization.js";
import { selectCapturedProcessGroupIds } from "../ProcessOwnershipProcessTree.js";
@@ -52,7 +55,6 @@ import {
describeUnverifiedCause,
type ProcessCaptureResourceCleanupOptions,
ProcessCaptureResourceScope,
type TerminalCaptureRetention,
captureExitObservation,
} from "./ProcessCaptureLifecycle.js";
import {
@@ -88,7 +90,7 @@ interface StartedCaptureRuntime {
readonly startedAt: Date;
readonly executableIdentity: ReturnType<typeof observeLaunchedExecutable>;
readonly lastOutput: () => number;
readonly terminalRetention: () => TerminalCaptureRetention;
readonly completeTerminal: ReturnType<typeof createProcessTerminalFinalizer>;
readonly stopSampler: ReturnType<typeof startProcessSampler>;
/** One ownership ledger shared by observations and final cleanup for this run. */
readonly ownershipEvidence: ProcessOwnershipEvidenceOwner;
@@ -289,10 +291,14 @@ const startCaptureRuntime = async (
const startedAt = new Date();
const started = monotonicProcessCaptureClock();
let lastOutput = started;
const identifiers = createProcessIdentifierLookup(
options.samples,
() => terminal?.pid ?? -1,
);
const normalizeText = createBoundedProcessTextNormalizer(
scenario,
options.temporaryRoot,
() => terminal?.pid ?? -1,
identifiers,
);
renderer = createTerminalRenderer(
scenario,
@@ -325,6 +331,7 @@ const startCaptureRuntime = async (
onOutput: () => (lastOutput = monotonicProcessCaptureClock()),
renderer,
normalizeText,
identifiers,
});
scheduleScenarioInteractions({
...options,
@@ -348,7 +355,11 @@ const startCaptureRuntime = async (
startedAt,
executableIdentity,
lastOutput: () => lastOutput,
terminalRetention,
completeTerminal: createProcessTerminalFinalizer(
renderer,
terminalRetention,
options.observationBuffer,
),
stopSampler,
signalRoot: retainRootSignaller(
terminal.pid,
@@ -401,6 +412,26 @@ const finishProcessRun = async (options: {
}): Promise<ProcessCapture> => {
options.progress.closeLive();
await options.stopSampler();
let { capture, executionFailure } = options;
if (options.runtime !== undefined && capture === undefined) {
const terminal = await options.runtime.completeTerminal();
if (!terminal.ok) {
if (executionFailure === undefined) executionFailure = terminal.error;
else {
const primary =
executionFailure instanceof AnalysisError
? executionFailure
: new ProcessCaptureError("process capture failed", {
cause: executionFailure,
});
if (primary !== terminal.error)
primary.retainSecondaryFailure(terminal.error, {
terminal_observation_error: projectAnalysisError(terminal.error),
});
executionFailure = primary;
}
}
}
const sampledProcessGroupIds =
options.runtime === undefined
? undefined
@@ -424,7 +455,6 @@ const finishProcessRun = async (options: {
? await releaseProcessResources(cleanupOptions)
: await options.resourceScope.release(cleanupOptions);
const cleanup = cleanupOutcome.report;
let { capture, executionFailure } = options;
if (executionFailure === undefined && options.signal?.aborted)
executionFailure = processCaptureCancelled(options.signal.reason);
const observedExit =
@@ -578,20 +608,9 @@ const completeCapture = async (options: {
},
};
options.recordEvent("filesystem_checkpoints", 1);
let renderedFrames: Awaited<ReturnType<TerminalRenderer["frames"]>>;
try {
renderedFrames = await runtime.renderer.frames();
} catch (cause: unknown) {
options.observationBuffer.rendered_frames = {
state: "unavailable",
reason: unavailableReason("Rendered terminal frame collection", cause),
};
throw cause;
}
options.observationBuffer.rendered_frames = {
state: "available",
value: renderedFrames,
};
const terminal = await runtime.completeTerminal();
if (!terminal.ok) throw terminal.error;
const { renderedFrames, retention } = terminal.value;
const checkpoints: UnverifiedProcessCapture["filesystem_checkpoints"] = [
{
name: "before",
@@ -632,7 +651,7 @@ const completeCapture = async (options: {
before: options.before,
after,
truncationDetails: {
...runtime.terminalRetention(),
...retention,
rendered_terminal: runtime.renderer.retention(),
filesystem_before: options.before.coverage,
filesystem_after: after.coverage,
@@ -832,29 +851,6 @@ const runProcessScenario = async (
});
} catch (cause: unknown) {
executionFailure = normalizeCaptureFailure(cause, signal);
if (
runtime !== undefined &&
observations.rendered_frames.state === "unavailable" &&
observations.rendered_frames.reason.startsWith(
"Rendered terminal frames were not collected",
)
) {
try {
observations.rendered_frames = {
state: "available",
value: await runtime.renderer.frames(),
};
} catch (renderCause: unknown) {
const renderFailure =
renderCause instanceof Error
? renderCause.message
: "unknown failure";
observations.rendered_frames = {
state: "unavailable",
reason: `Rendered terminal frame collection failed: ${renderFailure}`,
};
}
}
}
return finishProcessRun({
runtime,
@@ -3,6 +3,7 @@ import { expect, it } from "vitest";
import {
createBoundedProcessTextNormalizer,
createProcessTextNormalizer,
createProcessIdentifierLookup,
normalizeProcessSamples,
normalizeProcessText,
} from "./ProcessNormalization.js";
@@ -576,3 +577,47 @@ it.each([
});
}
});
it("shares sampled identifiers before custom patterns in terminal and command normalization", () => {
const observed = [
{
at_ms: 0,
pid: 100,
parent_pid: 1,
process_group_id: 100,
session_id: 100,
command: "root=100",
},
];
const lookup = createProcessIdentifierLookup(observed, () => 100);
const before = lookup();
expect(lookup()).toBe(before);
const declared = scenario({
paths: false,
ports: false,
pids: true,
patterns: [
{ pattern: "<pid>", replacement: "selected" },
{ pattern: "200", replacement: "wrong-order" },
],
});
const normalize = createProcessTextNormalizer(declared, "/temporary", lookup);
expect(normalize("root=100 child=200 other=201")).toBe(
"root=selected child=wrong-order other=201",
);
observed.push({
at_ms: 1,
pid: 200,
parent_pid: 100,
process_group_id: 100,
session_id: 100,
command: "root=100 child=200 other=201",
});
expect(lookup()).not.toBe(before);
expect(normalize(observed[1]!.command)).toBe(
"root=selected child=selected other=201",
);
expect(normalizeProcessSamples(observed, declared, 100)[1]?.command).toBe(
"root=selected child=selected other=201",
);
});
+103 -37
View File
@@ -2,7 +2,11 @@ import { isIP } from "node:net";
import type { ProcessScenario } from "../../domain/process/processScenario.js";
import type { ProcessSample } from "../../domain/process/processCapture.js";
import { AnalysisResourceConstraintError } from "../../domain/analysisErrorCore.js";
import type {
TerminalFrame,
ProcessSample,
} from "../../domain/process/processCapture.js";
export type BoundedProcessTextNormalization =
| {
@@ -183,11 +187,52 @@ type ProcessTextRule =
readonly replacement: string;
};
type ProcessIdentifierSource =
| number
| readonly number[]
| (() => number | readonly number[]);
/** Encounter order defines one shared identity set for text and sampled rows. */
const processSampleIdentifiers = (
samples: readonly ProcessSample[],
rootPid: number,
): readonly number[] =>
[
...new Set([
rootPid,
...samples.flatMap((sample) => [
sample.pid,
sample.parent_pid,
sample.process_group_id ?? 0,
sample.session_id ?? 0,
]),
]),
].filter((pid) => pid > 0);
/** Reuse an immutable identifier snapshot until the root or retained samples change. */
export const createProcessIdentifierLookup = (
samples: readonly ProcessSample[],
rootPid: () => number,
): (() => readonly number[]) => {
let count = -1;
let root = -1;
let identifiers: readonly number[] = [];
return () => {
const currentRoot = rootPid();
if (currentRoot !== root || count !== samples.length) {
root = currentRoot;
count = samples.length;
identifiers = processSampleIdentifiers(samples, root);
}
return identifiers;
};
};
/** Compile one ordered policy for both bounded retention and unrestricted text. */
const createProcessTextRules = (
scenario: ProcessScenario,
temporaryRoot: string,
pid: number | (() => number),
pid: ProcessIdentifierSource,
): (() => readonly (ProcessTextRule | undefined)[]) => {
const rules: (ProcessTextRule | undefined)[] = [];
const pathRule = scenario.normalization.paths
@@ -211,13 +256,15 @@ const createProcessTextRules = (
if (scenario.normalization.pids) rules.push(undefined);
for (const pattern of scenario.normalization.patterns)
rules.push({ kind: "literal", ...pattern });
let previousPid: number | undefined;
let previousPid: number | readonly number[] | undefined;
return () => {
if (scenario.normalization.pids) {
const currentPid = typeof pid === "number" ? pid : pid();
const currentPid = typeof pid === "function" ? pid() : pid;
if (currentPid !== previousPid) {
previousPid = currentPid;
const matcher = createPidMatcher([currentPid]);
const matcher = createPidMatcher(
typeof currentPid === "number" ? [currentPid] : currentPid,
);
rules[pidIndex] =
matcher === undefined
? undefined
@@ -232,7 +279,7 @@ const createProcessTextRules = (
export const createProcessTextNormalizer = (
scenario: ProcessScenario,
temporaryRoot: string,
pid: number | (() => number),
pid: ProcessIdentifierSource,
): ((value: string) => string) => {
const rules = createProcessTextRules(scenario, temporaryRoot, pid);
return (value) => {
@@ -252,7 +299,7 @@ export const createProcessTextNormalizer = (
export const createBoundedProcessTextNormalizer = (
scenario: ProcessScenario,
temporaryRoot: string,
pid: number | (() => number),
pid: ProcessIdentifierSource,
): ((value: string, maxBytes: number) => BoundedProcessTextNormalization) => {
const rules = createProcessTextRules(scenario, temporaryRoot, pid);
return (value, maxBytes) => {
@@ -363,32 +410,17 @@ export const normalizeProcessSamples = (
scenario: ProcessScenario,
rootPid: number,
): readonly ProcessSample[] => {
const identifiers = [
rootPid,
...samples.flatMap((sample) => [
sample.pid,
sample.parent_pid,
sample.process_group_id ?? 0,
sample.session_id ?? 0,
]),
];
const mapping = new Map<number, number>();
for (const identifier of identifiers)
if (identifier > 0 && !mapping.has(identifier))
mapping.set(identifier, mapping.size + 1);
const identifiers = processSampleIdentifiers(samples, rootPid);
const mapping = new Map(
identifiers.map((identifier, index) => [identifier, index + 1]),
);
const normalizeText = createProcessTextNormalizer(
scenario,
"<no-temporary-root>",
rootPid,
identifiers,
);
const normalizeSamplePids = scenario.normalization.pids
? createPidNormalizer(identifiers)
: undefined;
const normalizeCommand = (command: string): string => {
const normalized = normalizeText(command);
const withNormalizedPids = normalizeSamplePids?.(normalized) ?? normalized;
return withNormalizedPids.trim();
};
const normalizeCommand = (command: string): string =>
normalizeText(command).trim();
return samples.map((sample) => ({
at_ms: normalizeProcessElapsedTime(
sample.at_ms,
@@ -412,14 +444,6 @@ 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, "<pid>");
};
const createPidMatcher = (
identifiers: readonly number[],
): RegExp | undefined => {
@@ -430,3 +454,45 @@ const createPidMatcher = (
if (tokens.length === 0) return undefined;
return new RegExp(`(?<!\\d)(?:${tokens.join("|")})(?!\\d)`, "gu");
};
/** Reproject retained raw chunks atomically when later samples reveal more identifiers. */
export const normalizeRetainedTerminalFrames = (
frames: TerminalFrame[],
normalize: (
value: string,
maxBytes: number,
) => BoundedProcessTextNormalization,
maximumBytes: number,
): number => {
let bytes = 0;
const projected: TerminalFrame[] = [];
for (const frame of frames) {
const raw = frame.raw_data ?? frame.data;
const value = normalize(raw, maximumBytes - bytes);
if (value.state === "over_budget")
throw new AnalysisResourceConstraintError(
"capture_process",
"capacity",
"Retained terminal text exceeded the selected output budget after sampled PID normalization.",
{
maximum_output_bytes: maximumBytes,
frame_index: frame.sequence,
boundary: "terminal-pid-normalization",
},
{
retryable: false,
remediationAction:
"Increase limits.output_bytes within its advertised range, or select a narrower scenario. Completed raw PTY chunks remain in partial observations; the corrected normalized representation did not fit.",
},
);
bytes += value.utf8Bytes;
projected.push({
sequence: frame.sequence,
at_ms: frame.at_ms,
data: value.value,
...(value.value === raw ? {} : { raw_data: raw }),
});
}
for (const [index, frame] of projected.entries()) frames[index] = frame;
return bytes;
};
@@ -0,0 +1,90 @@
import { expect, it, vi } from "vitest";
import { AnalysisResourceConstraintError } from "../../domain/analysisErrorCore.js";
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
import { parseProcessScenario } from "../../domain/process/processScenario.js";
import {
createProcessCaptureObservationBuffer,
resolveProcessResult,
} from "./ProcessCaptureLifecycle.js";
import { createProcessTerminalFinalizer } from "./ProcessTerminalObservation.js";
it("retains completed terminal facts and the typed capacity failure through finalization", async () => {
const failure = new AnalysisResourceConstraintError(
"capture_process",
"capacity",
"Sampled PID projection exceeded output_bytes",
{ maximum_output_bytes: 4 },
);
const readRetention = vi.fn(() => {
throw failure;
});
const frames = [{ sequence: 0, at_ms: 17, data: "20" }];
const observations = createProcessCaptureObservationBuffer({
frames,
samples: [],
interactions: [],
eventJournal: [],
before: undefined,
});
const rendered = [
{
sequence: 0,
at_ms: 17,
columns: 10,
rows: 1,
cursor_x: 0,
cursor_y: 0,
active_buffer: "normal" as const,
lines: [""],
serialized_state: "",
},
];
const renderer = { frames: vi.fn(async () => rendered) };
const finalize = createProcessTerminalFinalizer(
renderer,
readRetention,
observations,
);
const result = await finalize();
expect(await finalize()).toBe(result);
expect(readRetention).toHaveBeenCalledOnce();
expect(renderer.frames).toHaveBeenCalledOnce();
if (result.ok) throw new Error("Expected terminal capacity refusal");
expect(result.error).toBe(failure);
expect(observations.rendered_frames).toEqual({
state: "available",
value: rendered,
});
const scenario = parseProcessScenario({
executable: "unused",
limits: { output_bytes: 4 },
});
const cleanup = { state: "not_required" as const, reason: null };
expect(() =>
resolveProcessResult(
undefined,
result.error,
{
report: {
owned_process_group: cleanup,
terminal_renderer: cleanup,
temporary_root: cleanup,
},
cleanupLease: undefined,
},
observations,
{ scenario, rootPid: 100 },
),
).toThrow(failure);
expect(projectAnalysisError(failure)).toMatchObject({
code: "resource_constraint",
details: {
partial_observation: {
observations: {
frames: { state: "available", value: frames },
rendered_frames: { state: "available", value: rendered },
},
},
},
});
});
@@ -0,0 +1,72 @@
import { err, ok, type Result } from "../../domain/result.js";
import { AnalysisError } from "../../domain/analysisErrorBase.js";
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
import { ProcessCaptureError } from "./ProcessCaptureError.js";
import type { TerminalRenderer } from "./TerminalRenderer.js";
import type {
ProcessCaptureObservationBuffer,
TerminalCaptureRetention,
} from "./ProcessCaptureLifecycle.js";
import type { RenderedTerminalFrame } from "../../domain/process/processCapture.js";
interface ProcessTerminalObservation {
readonly retention: TerminalCaptureRetention;
readonly renderedFrames: readonly RenderedTerminalFrame[];
}
const terminalFailure = (cause: unknown): AnalysisError =>
cause instanceof AnalysisError
? cause
: new ProcessCaptureError(
`Terminal observation failed: ${cause instanceof Error ? cause.message : String(cause)}`,
{ cause },
);
/** Finalize retained chunks once after sampling, for both completion and error cleanup. */
export const createProcessTerminalFinalizer = (
renderer: Pick<TerminalRenderer, "frames">,
retention: () => TerminalCaptureRetention,
observations: ProcessCaptureObservationBuffer,
): (() => Promise<Result<ProcessTerminalObservation, AnalysisError>>) => {
let completed:
| Promise<Result<ProcessTerminalObservation, AnalysisError>>
| undefined;
return () =>
(completed ??= observeTerminal(renderer, retention, observations));
};
const observeTerminal = async (
renderer: Pick<TerminalRenderer, "frames">,
readRetention: () => TerminalCaptureRetention,
observations: ProcessCaptureObservationBuffer,
): Promise<Result<ProcessTerminalObservation, AnalysisError>> => {
let retention: Result<TerminalCaptureRetention, AnalysisError>;
try {
retention = ok(readRetention());
} catch (cause) {
retention = err(terminalFailure(cause));
}
let rendered: Result<readonly RenderedTerminalFrame[], AnalysisError>;
try {
rendered = ok(await renderer.frames());
observations.rendered_frames = {
state: "available",
value: rendered.value,
};
} catch (cause) {
rendered = err(terminalFailure(cause));
observations.rendered_frames = {
state: "unavailable",
reason: `Rendered terminal frame collection failed: ${rendered.error.message}`,
};
}
if (!retention.ok) {
if (!rendered.ok)
retention.error.retainSecondaryFailure(rendered.error, {
rendered_terminal_failure: projectAnalysisError(rendered.error),
});
return retention;
}
if (!rendered.ok) return rendered;
return ok({ retention: retention.value, renderedFrames: rendered.value });
};
@@ -282,3 +282,88 @@ it.each([1024, 2048])(
}
},
);
it("renders retained chunks once after late PID knowledge, preserving event times and resize order", async () => {
let identifiers: readonly number[] = [100];
const scenario = parseProcessScenario({
executable: "unused",
normalization: {
paths: false,
ports: false,
pids: true,
patterns: [
{ pattern: "<pid>", replacement: "selected" },
{ pattern: "200", replacement: "wrong-order" },
],
},
});
const write = vi.spyOn(Terminal.prototype, "write");
const renderer = new TerminalRenderer({
columns: 30,
rows: 2,
scrollback: 0,
maxBytes: 1000,
normalize: createBoundedProcessTextNormalizer(
scenario,
"/tmp",
() => identifiers,
),
});
try {
renderer.write("child=200", 17);
renderer.resize(40, 3, 18);
expect(write).not.toHaveBeenCalled();
identifiers = [100, 200];
const frames = await renderer.frames();
expect(write).toHaveBeenCalledTimes(1);
expect(
frames.map(({ at_ms, columns, rows }) => ({ at_ms, columns, rows })),
).toEqual([
{ at_ms: 17, columns: 30, rows: 2 },
{ at_ms: 18, columns: 40, rows: 3 },
]);
for (const frame of frames) {
expect(frame.lines[0]).toBe("child=selected");
expect(frame.serialized_state).toContain("child=selected");
}
} finally {
await renderer.dispose();
write.mockRestore();
}
});
it("renders a bounded sequence of retained chunks and resizes", async () => {
const scenario = parseProcessScenario({
executable: "unused",
normalization: { paths: false, ports: false, pids: true, patterns: [] },
});
const renderer = new TerminalRenderer({
columns: 20,
rows: 4,
scrollback: 10,
maxBytes: 1_048_576,
normalize: createBoundedProcessTextNormalizer(scenario, "/tmp", 100),
});
try {
for (let index = 0; index < 100; index += 1) {
renderer.write(`root=100 row=${index}\r\n`, index * 10);
if (index === 24 || index === 74)
renderer.resize(index === 24 ? 30 : 20, 4, index * 10 + 1);
}
const frames = await renderer.frames();
expect(frames).toHaveLength(102);
expect(
frames
.filter(({ at_ms }) => at_ms % 10 === 1)
.map(({ at_ms, columns }) => ({ at_ms, columns })),
).toEqual([
{ at_ms: 241, columns: 30 },
{ at_ms: 741, columns: 20 },
]);
expect(frames.at(-1)?.serialized_state).toContain("root=<pid> row=99");
expect(renderer.retention().retained_frames).toBe(102);
expect(renderer.truncated()).toBe(false);
} finally {
await renderer.dispose();
}
});
+12 -5
View File
@@ -42,6 +42,7 @@ export class TerminalRenderer {
readonly #serializeAddon = new SerializePackage.SerializeAddon();
readonly #frames: RenderedTerminalFrame[] = [];
#pending: Promise<void> = Promise.resolve();
#operations: Array<() => void | Promise<void>> = [];
#capturedBytes = 0;
#observedBytes = 0;
#observedFrames = 0;
@@ -86,7 +87,13 @@ export class TerminalRenderer {
/** Await all queued parsing and return immutable rendered observations. */
async frames(): Promise<readonly RenderedTerminalFrame[]> {
await this.#pending;
do {
const operations = this.#operations;
this.#operations = [];
for (const operation of operations)
this.#pending = this.#pending.then(operation);
await this.#pending;
} while (this.#operations.length > 0);
return structuredClone(this.#frames);
}
@@ -119,15 +126,15 @@ export class TerminalRenderer {
async dispose(): Promise<void> {
// frames() owns observation failures; cleanup reports only disposal failures.
await this.#pending.catch(() => undefined);
this.#operations = [];
// xterm owns and disposes every addon loaded through loadAddon().
this.#terminal.dispose();
}
#queue(operation: () => void | Promise<void>): void {
this.#pending = this.#pending.then(operation);
// PTY callbacks can fail before capture completion awaits frames(). Keep
// the original rejection observable there without an unhandled rejection.
void this.#pending.catch(() => undefined);
// Raw chunks are already retained by the capture byte owner. Keep their
// references and resize order; render once after sampled identities settle.
this.#operations.push(operation);
}
#capture(atMs: number): void {
@@ -334,3 +334,65 @@ itWithCaptureCapability(
);
},
);
itWithCaptureCapability.each([false, true])(
"normalizes a sampled child in raw and rendered terminal projections, pids=%s",
async (pids) => {
const root = await createTestTempDirectory("rea-sampled-terminal-pids-");
const scenario = parseProcessScenario({
executable: process.execPath,
working_directory: root,
arguments: [
"-e",
`const {spawn}=require('node:child_process');const child=spawn(process.execPath,['-e','setTimeout(()=>{},600)'],{stdio:'inherit'});console.log('rootpid='+process.pid+' childpid='+child.pid);setTimeout(()=>{},700);`,
],
timeout_ms: 3000,
settle_ms: 100,
normalization: {
paths: false,
ports: false,
pids,
time_bucket_ms: 10000,
patterns: [{ pattern: "<pid>", replacement: "<selected-pid>" }],
},
});
const result = await captureProcessScenario(scenario);
if (!result.ok) throw result.error;
const capture = result.value;
const raw = capture.frames
.map(({ data, raw_data }) => raw_data ?? data)
.join("");
const match = /rootpid=(\d+) childpid=(\d+)/u.exec(raw);
if (match === null)
throw new Error("Expected exact root and child PID observations");
expect(
capture.process_samples.some((row) =>
pids
? row.parent_pid === 1 && row.pid !== 1
: row.pid === Number(match[2]),
),
).toBe(true);
const expected = pids
? "rootpid=<selected-pid> childpid=<selected-pid>"
: match[0];
expect(capture.frames.map(({ data }) => data).join("")).toContain(expected);
expect(
capture.rendered_frames.some(
({ lines, serialized_state }) =>
lines.includes(expected) && serialized_state.includes(expected),
),
).toBe(true);
if (pids)
expect(
capture.manifest.comparison_contract.pid_normalization_version,
).toBe("sampled-identifiers-v2");
else
expect(capture.manifest.comparison_contract).not.toHaveProperty(
"pid_normalization_version",
);
expect(capture.cleanup).toMatchObject({
owned_process_group: "verified",
temporary_root: "removed",
});
},
);