diff --git a/docs/process-capture.md b/docs/process-capture.md index 7277dbb700..3f6353af2b 100644 --- a/docs/process-capture.md +++ b/docs/process-capture.md @@ -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 diff --git a/src/domain/process/processScenario.test.ts b/src/domain/process/processScenario.test.ts index 5d45f61a74..734c7d5522 100644 --- a/src/domain/process/processScenario.test.ts +++ b/src/domain/process/processScenario.test.ts @@ -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", () => { diff --git a/src/domain/process/processScenario.ts b/src/domain/process/processScenario.ts index 6378fa715f..bc372a583e 100644 --- a/src/domain/process/processScenario.ts +++ b/src/domain/process/processScenario.ts @@ -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" } : {}), diff --git a/src/process/capture/ProcessCaptureLifecycle.ts b/src/process/capture/ProcessCaptureLifecycle.ts index 64dffd18a5..0e96e0af22 100644 --- a/src/process/capture/ProcessCaptureLifecycle.ts +++ b/src/process/capture/ProcessCaptureLifecycle.ts @@ -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 { diff --git a/src/process/capture/ProcessCaptureTerminalRetention.test.ts b/src/process/capture/ProcessCaptureTerminalRetention.test.ts index e3bde2264c..0c1f288570 100644 --- a/src/process/capture/ProcessCaptureTerminalRetention.test.ts +++ b/src/process/capture/ProcessCaptureTerminalRetention.test.ts @@ -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: "", 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); + } + }, +); diff --git a/src/process/capture/ProcessHarness.ts b/src/process/capture/ProcessHarness.ts index e26db834be..e287b3d0c3 100644 --- a/src/process/capture/ProcessHarness.ts +++ b/src/process/capture/ProcessHarness.ts @@ -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; readonly lastOutput: () => number; - readonly terminalRetention: () => TerminalCaptureRetention; + readonly completeTerminal: ReturnType; readonly stopSampler: ReturnType; /** 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 => { 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>; - 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, diff --git a/src/process/capture/ProcessNormalization.test.ts b/src/process/capture/ProcessNormalization.test.ts index b851b962c4..a8bc0daed5 100644 --- a/src/process/capture/ProcessNormalization.test.ts +++ b/src/process/capture/ProcessNormalization.test.ts @@ -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: "", 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", + ); +}); diff --git a/src/process/capture/ProcessNormalization.ts b/src/process/capture/ProcessNormalization.ts index 329d366f58..730f65435c 100644 --- a/src/process/capture/ProcessNormalization.ts +++ b/src/process/capture/ProcessNormalization.ts @@ -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(); - 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, "", - 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, ""); -}; - const createPidMatcher = ( identifiers: readonly number[], ): RegExp | undefined => { @@ -430,3 +454,45 @@ const createPidMatcher = ( if (tokens.length === 0) return undefined; return new RegExp(`(? 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; +}; diff --git a/src/process/capture/ProcessTerminalObservation.test.ts b/src/process/capture/ProcessTerminalObservation.test.ts new file mode 100644 index 0000000000..5e3652f7dd --- /dev/null +++ b/src/process/capture/ProcessTerminalObservation.test.ts @@ -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 }, + }, + }, + }, + }); +}); diff --git a/src/process/capture/ProcessTerminalObservation.ts b/src/process/capture/ProcessTerminalObservation.ts new file mode 100644 index 0000000000..e218b8af65 --- /dev/null +++ b/src/process/capture/ProcessTerminalObservation.ts @@ -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, + retention: () => TerminalCaptureRetention, + observations: ProcessCaptureObservationBuffer, +): (() => Promise>) => { + let completed: + | Promise> + | undefined; + return () => + (completed ??= observeTerminal(renderer, retention, observations)); +}; + +const observeTerminal = async ( + renderer: Pick, + readRetention: () => TerminalCaptureRetention, + observations: ProcessCaptureObservationBuffer, +): Promise> => { + let retention: Result; + try { + retention = ok(readRetention()); + } catch (cause) { + retention = err(terminalFailure(cause)); + } + let rendered: Result; + 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 }); +}; diff --git a/src/process/capture/TerminalRenderer.test.ts b/src/process/capture/TerminalRenderer.test.ts index 9381979e97..32e37b4818 100644 --- a/src/process/capture/TerminalRenderer.test.ts +++ b/src/process/capture/TerminalRenderer.test.ts @@ -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: "", 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= row=99"); + expect(renderer.retention().retained_frames).toBe(102); + expect(renderer.truncated()).toBe(false); + } finally { + await renderer.dispose(); + } +}); diff --git a/src/process/capture/TerminalRenderer.ts b/src/process/capture/TerminalRenderer.ts index 749e22aca7..b2061b2ad0 100644 --- a/src/process/capture/TerminalRenderer.ts +++ b/src/process/capture/TerminalRenderer.ts @@ -42,6 +42,7 @@ export class TerminalRenderer { readonly #serializeAddon = new SerializePackage.SerializeAddon(); readonly #frames: RenderedTerminalFrame[] = []; #pending: Promise = Promise.resolve(); + #operations: Array<() => void | Promise> = []; #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 { - 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 { // 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 { - 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 { diff --git a/tests/boundary/process/processCaptureNormalization.test.ts b/tests/boundary/process/processCaptureNormalization.test.ts index 93185f5d07..ade22f8816 100644 --- a/tests/boundary/process/processCaptureNormalization.test.ts +++ b/tests/boundary/process/processCaptureNormalization.test.ts @@ -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: "", replacement: "" }], + }, + }); + 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= childpid=" + : 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", + }); + }, +);