From 8bcdb986893becd151c4dbb825f69bb6ced8b52e Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:07:49 +0000 Subject: [PATCH] feat(process): record global capture event order --- src/application/CommandShimReplay.ts | 14 ++- src/application/LoopbackReplay.ts | 10 +- src/application/LoopbackReplayRecorder.ts | 27 ++++- src/application/ProcessCaptureLifecycle.ts | 37 +++++-- src/application/ProcessCheckpoints.ts | 17 ++- src/application/ProcessHarness.ts | 41 +++++-- src/application/ProcessSampling.ts | 8 +- src/application/TerminalRenderer.ts | 10 +- src/domain/processCapture.ts | 50 +++++++++ src/domain/processCaptureValidation.ts | 49 ++++++++ tests/processCapture.test.ts | 123 ++++++++++++++++++++- 11 files changed, 349 insertions(+), 37 deletions(-) diff --git a/src/application/CommandShimReplay.ts b/src/application/CommandShimReplay.ts index ecd5868ec..860e41caf 100644 --- a/src/application/CommandShimReplay.ts +++ b/src/application/CommandShimReplay.ts @@ -2,7 +2,11 @@ import { createServer, type Server } from "node:http"; import { chmod, mkdir, writeFile } from "node:fs/promises"; import { join } from "node:path"; -import type { ProcessScenario, ShimEvent } from "../domain/processCapture.js"; +import type { + ProcessScenario, + RecordProcessCaptureEvent, + ShimEvent, +} from "../domain/processCapture.js"; interface Invocation { readonly command: string; @@ -52,6 +56,7 @@ export const startCommandShimReplay = async ( scenario: ProcessScenario, temporaryRoot: string, started: number, + recordEvent: RecordProcessCaptureEvent = () => undefined, ): Promise => { const binPath = join(temporaryRoot, "shims"); await mkdir(binPath); @@ -94,9 +99,10 @@ export const startCommandShimReplay = async ( ? "exhausted" : "matched"; if (events.length >= scenario.limits.protocol_events) truncated = true; - else + else { + const index = events.length; events.push({ - sequence: events.length, + sequence: index, at_ms: Math.max(0, Date.now() - started), command: invocation.command, route_index: @@ -105,6 +111,8 @@ export const startCommandShimReplay = async ( working_directory: invocation.working_directory, outcome, }); + recordEvent("shim_events", index); + } if (outcome !== "matched" || route === undefined) { response.writeHead(409).end(); return; diff --git a/src/application/LoopbackReplay.ts b/src/application/LoopbackReplay.ts index 92efc96df..8450c0853 100644 --- a/src/application/LoopbackReplay.ts +++ b/src/application/LoopbackReplay.ts @@ -8,6 +8,7 @@ import { WebSocketServer, type WebSocket } from "ws"; import type { ProcessScenario, ProtocolEvent, + RecordProcessCaptureEvent, } from "../domain/processCapture.js"; import { type ReplayMachineDecision, @@ -65,7 +66,7 @@ const handleMachineHttp = async ( ): Promise => { const machine = recorder.machine; if (machine === undefined) return; - const decision = machine.dispatch({ + const decision = recorder.dispatchMachine({ protocol: "http", connection: "not_applicable", at_ms: request.rawAtMs, @@ -352,7 +353,7 @@ const handleMachineWebSocket = (options: { void recorder .enqueueMachine(async () => { const connectedAt = recorder.atMs(); - const connectionDecision = machine.dispatch({ + const connectionDecision = recorder.dispatchMachine({ protocol: "websocket_connect", connection: connection === 1 ? "initial" : "reconnect", at_ms: recorder.rawAtMs(), @@ -388,7 +389,7 @@ const handleMachineWebSocket = (options: { const recordedAtMs = recorder.atMs(); void recorder .enqueueMachine(async () => { - const decision = machine.dispatch({ + const decision = recorder.dispatchMachine({ protocol: "websocket_message", connection: connection === 1 ? "initial" : "reconnect", at_ms: rawAtMs, @@ -468,8 +469,9 @@ const createWebSocketReplay = ( /** Start bounded HTTP and WebSocket replay endpoints on IPv4 loopback only. */ export const startLoopbackReplay = async ( scenario: ProcessScenario, + recordEvent: RecordProcessCaptureEvent = () => undefined, ): Promise => { - const recorder = new ReplayRecorder(scenario); + const recorder = new ReplayRecorder(scenario, recordEvent); const server = createServer(createHttpHandler(recorder)); const websocket = createWebSocketReplay(server, recorder); let closePromise: Promise | undefined; diff --git a/src/application/LoopbackReplayRecorder.ts b/src/application/LoopbackReplayRecorder.ts index 827ed6282..d676ab434 100644 --- a/src/application/LoopbackReplayRecorder.ts +++ b/src/application/LoopbackReplayRecorder.ts @@ -2,6 +2,7 @@ import type { IncomingMessage, Server } from "node:http"; import type { ProcessScenario, ProtocolEvent, + RecordProcessCaptureEvent, } from "../domain/processCapture.js"; import { ReplayMachineRuntime, @@ -18,7 +19,10 @@ export class ReplayRecorder { #machineAdmissionOpen = true; truncated = false; - constructor(readonly scenario: ProcessScenario) { + constructor( + readonly scenario: ProcessScenario, + private readonly recordEvent: RecordProcessCaptureEvent = () => undefined, + ) { this.machine = scenario.replay.machine === null ? undefined @@ -61,10 +65,25 @@ export class ReplayRecorder { await this.#machineQueue; } + dispatchMachine( + event: Parameters[0], + ): ReplayMachineDecision { + if (this.machine === undefined) + throw new Error("replay machine is not configured"); + const decision = this.machine.dispatch(event); + if (decision.transition !== null) + this.recordEvent("replay_transitions", decision.transition.sequence); + return decision; + } + record(event: Omit): void { - if (this.events.length < this.scenario.limits.protocol_events) - this.events.push({ sequence: this.events.length, ...event }); - else this.truncated = true; + if (this.events.length >= this.scenario.limits.protocol_events) { + this.truncated = true; + return; + } + const sequence = this.events.length; + this.events.push({ sequence, ...event }); + this.recordEvent("protocol_events", sequence); } } diff --git a/src/application/ProcessCaptureLifecycle.ts b/src/application/ProcessCaptureLifecycle.ts index b42adf7cf..61e4908d4 100644 --- a/src/application/ProcessCaptureLifecycle.ts +++ b/src/application/ProcessCaptureLifecycle.ts @@ -13,6 +13,8 @@ import type { ProcessPolicyDecision, ProcessSample, ProcessScenario, + ProcessCaptureEventJournalEntry, + RecordProcessCaptureEvent, TerminalFrame, } from "../domain/processCapture.js"; import { @@ -56,6 +58,7 @@ interface TerminalExitOptions { readonly timers: Set; readonly interactions: InteractionEvent[]; readonly dispatchedEventIndexes: ReadonlySet; + readonly recordEvent: RecordProcessCaptureEvent; } interface CaptureResultOptions { @@ -79,6 +82,7 @@ interface CaptureResultOptions { readonly shimEvents: ProcessCapture["shim_events"]; readonly settlement: ProcessCapture["settlement"]; readonly manifest: ProcessCapture["manifest"]; + readonly eventJournal: readonly ProcessCaptureEventJournalEntry[]; } export const buildCaptureResult = ( @@ -88,12 +92,7 @@ export const buildCaptureResult = ( manifest: options.manifest, normalization: options.scenario.normalization, frames: options.frames, - rendered_frames: [...options.renderedFrames] - .sort( - (left, right) => - left.at_ms - right.at_ms || left.sequence - right.sequence, - ) - .map((frame, sequence) => ({ ...frame, sequence })), + rendered_frames: options.renderedFrames, interaction_events: options.interactions, exit: { code: @@ -116,6 +115,7 @@ export const buildCaptureResult = ( options.scenario, ), replay_transitions: options.replay.transitions, + event_journal: options.eventJournal, files_before: options.before.files, files_after: options.after.files, filesystem_effects: classifyFilesystemEffects( @@ -189,9 +189,12 @@ export const observeSettlement = async ( runId: string, processGroupIds: readonly number[], settleMs: number, + recordEvent: RecordProcessCaptureEvent = () => undefined, ): Promise> => { - if (process.platform === "win32") + if (process.platform === "win32") { + recordEvent("lifecycle", 1); return { state: "unverifiable", elapsed_ms: 0 }; + } const started = Date.now(); let consecutiveEmpty = 0; let deadlineReached = false; @@ -205,17 +208,22 @@ export const observeSettlement = async ( }), ), ); - if (observations.some(({ state }) => state === "unverifiable")) + if (observations.some(({ state }) => state === "unverifiable")) { + recordEvent("lifecycle", 1); return { state: "unverifiable", elapsed_ms: Date.now() - started }; + } if (observations.every(({ state }) => state === "empty")) { consecutiveEmpty += 1; - if (consecutiveEmpty >= 2) + if (consecutiveEmpty >= 2) { + recordEvent("lifecycle", 1); return { state: "quiesced", elapsed_ms: Date.now() - started }; + } } else consecutiveEmpty = 0; deadlineReached = Date.now() - started >= settleMs; if (deadlineReached) break; await new Promise((resolve) => setTimeout(resolve, 50)); } + recordEvent("lifecycle", 1); return { state: "alive_at_deadline", elapsed_ms: Date.now() - started }; }; @@ -228,6 +236,7 @@ export const awaitTerminalExit = async ({ timers, interactions, dispatchedEventIndexes, + recordEvent, }: TerminalExitOptions): Promise<{ exitCode: number; signal?: number; @@ -239,10 +248,12 @@ export const awaitTerminalExit = async ({ // exit from harness-owned timeout, idle-timeout, and cancellation cleanup. let reason: "exited" | "timeout" | "idle_timeout" | "cancelled" = "exited"; terminal.onExit((exit) => { + recordEvent("lifecycle", 0); for (const [eventIndex, event] of scenario.events.entries()) { if (dispatchedEventIndexes.has(eventIndex)) continue; + const sequence = interactions.length; interactions.push({ - sequence: interactions.length, + sequence, scheduled_at_ms: event.at_ms, dispatched_at_ms: Math.max(0, Date.now() - started), type: event.type, @@ -254,6 +265,7 @@ export const awaitTerminalExit = async ({ : event.signal, outcome: "target_exited", }); + recordEvent("interaction_events", sequence); } for (const timer of timers) { clearTimeout(timer); @@ -337,6 +349,7 @@ export const captureTerminalFrames = (options: { readonly onOutput: () => void; readonly renderer: TerminalRenderer; readonly checkpoints: ProcessCheckpoints; + readonly recordEvent: RecordProcessCaptureEvent; }): (() => boolean) => { let outputBytes = 0; let truncated = false; @@ -362,11 +375,13 @@ export const captureTerminalFrames = (options: { options.temporaryRoot, options.terminal.pid, ); + const sequence = options.frames.length; options.frames.push({ - sequence: options.frames.length, + sequence, at_ms: atMs, data: normalized, }); + options.recordEvent("frames", sequence); options.renderer.write(data, atMs); options.checkpoints.observeTerminal(normalized); }); diff --git a/src/application/ProcessCheckpoints.ts b/src/application/ProcessCheckpoints.ts index 37075cd4b..c693cc2d0 100644 --- a/src/application/ProcessCheckpoints.ts +++ b/src/application/ProcessCheckpoints.ts @@ -2,6 +2,7 @@ import type { FileState, FilesystemCheckpoint, ProcessScenario, + RecordProcessCaptureEvent, } from "../domain/processCapture.js"; import { snapshotRoots, type SnapshotResult } from "./FilesystemSnapshot.js"; @@ -36,14 +37,21 @@ export class ProcessCheckpoints { readonly #terminalCounts = new Map(); readonly #terminalTails = new Map(); readonly #timers = new Set(); + readonly #signal: AbortSignal | undefined; + readonly #recordEvent: RecordProcessCaptureEvent; #pending: Promise = Promise.resolve(); constructor( private readonly scenario: ProcessScenario, private readonly started: number, before: SnapshotResult, - private readonly signal: AbortSignal | undefined, + options: { + readonly signal: AbortSignal | undefined; + readonly recordEvent?: RecordProcessCaptureEvent; + } = { signal: undefined }, ) { + this.#signal = options.signal; + this.#recordEvent = options.recordEvent ?? (() => undefined); this.#captures.push({ name: "before", at_ms: 0, @@ -51,6 +59,7 @@ export class ProcessCheckpoints { effects: [], truncated: before.truncated, }); + this.#recordEvent("filesystem_checkpoints", 0); this.#captured.add("before"); for (const checkpoint of scenario.checkpoints) { if (checkpoint.trigger.type !== "time") continue; @@ -100,8 +109,9 @@ export class ProcessCheckpoints { // together. Each effect set is intentionally relative to the immediately // preceding checkpoint, which preserves transient create/delete behavior. this.#pending = this.#pending.then(async () => { - const snapshot = await snapshotRoots(this.scenario, this.signal); + const snapshot = await snapshotRoots(this.scenario, this.#signal); const previous = this.#captures.at(-1)?.files ?? []; + const index = this.#captures.length; this.#captures.push({ name, at_ms: Math.max(0, Date.now() - this.started), @@ -109,6 +119,7 @@ export class ProcessCheckpoints { effects: classifyFilesystemEffects(previous, snapshot.files), truncated: snapshot.truncated, }); + this.#recordEvent("filesystem_checkpoints", index); }); } @@ -119,6 +130,7 @@ export class ProcessCheckpoints { for (const timer of this.#timers) clearTimeout(timer); await this.#pending; const previous = this.#captures.at(-1)?.files ?? []; + const index = this.#captures.length; this.#captures.push({ name: "after_settlement", at_ms: Math.max(0, Date.now() - this.started), @@ -126,6 +138,7 @@ export class ProcessCheckpoints { effects: classifyFilesystemEffects(previous, after.files), truncated: after.truncated, }); + this.#recordEvent("filesystem_checkpoints", index); return structuredClone(this.#captures); } diff --git a/src/application/ProcessHarness.ts b/src/application/ProcessHarness.ts index a88a6a8a0..098adcaa8 100644 --- a/src/application/ProcessHarness.ts +++ b/src/application/ProcessHarness.ts @@ -2,9 +2,11 @@ import type { IPty } from "@lydell/node-pty"; import type { InteractionEvent, ProcessCapture, + ProcessCaptureEventJournalEntry, ProcessExecutionPolicy, ProcessSample, ProcessScenario, + RecordProcessCaptureEvent, TerminalFrame, } from "../domain/processCapture.js"; import { validateProcessCapture } from "../domain/processCapture.js"; @@ -83,6 +85,7 @@ interface ScenarioEventOptions { readonly renderer: TerminalRenderer; readonly started: number; readonly dispatchedEventIndexes: Set; + readonly recordEvent: RecordProcessCaptureEvent; } const scheduleScenarioEvents = (options: ScenarioEventOptions): void => { @@ -106,8 +109,9 @@ const scheduleScenarioEvents = (options: ScenarioEventOptions): void => { outcome = "failed"; } } + const sequence = interactions.length; interactions.push({ - sequence: interactions.length, + sequence, scheduled_at_ms: event.at_ms, dispatched_at_ms: dispatchedAt, type: event.type, @@ -126,6 +130,7 @@ const scheduleScenarioEvents = (options: ScenarioEventOptions): void => { : event.signal, outcome, }); + options.recordEvent("interaction_events", sequence); }, event.at_ms); timers.add(timer); } @@ -173,6 +178,7 @@ const createTerminalRenderer = ( scenario: ProcessScenario, temporaryRoot: string, terminalPid: () => number, + recordEvent: RecordProcessCaptureEvent, ): TerminalRenderer => new TerminalRenderer({ columns: scenario.terminal.columns, @@ -182,6 +188,7 @@ const createTerminalRenderer = ( maxBytes: scenario.limits.output_bytes, normalize: (value) => normalizeProcessText(value, scenario, temporaryRoot, terminalPid()), + recordEvent, }); const startCaptureRuntime = async (options: { @@ -195,6 +202,7 @@ const startCaptureRuntime = async (options: { readonly interactions: InteractionEvent[]; readonly timers: Set; readonly dispatchedEventIndexes: Set; + readonly recordEvent: RecordProcessCaptureEvent; readonly signal?: AbortSignal; }): Promise => { const { scenario } = options; @@ -205,7 +213,7 @@ const startCaptureRuntime = async (options: { let terminal: IPty | undefined; try { const { spawn } = await import("@lydell/node-pty"); - replay = await startLoopbackReplay(scenario); + replay = await startLoopbackReplay(scenario, options.recordEvent); const startedAt = new Date(); const started = Date.now(); let lastOutput = started; @@ -213,17 +221,17 @@ const startCaptureRuntime = async (options: { scenario, options.temporaryRoot, () => terminal?.pid ?? -1, + options.recordEvent, ); - checkpoints = new ProcessCheckpoints( - scenario, - started, - options.before, - options.signal, - ); + checkpoints = new ProcessCheckpoints(scenario, started, options.before, { + signal: options.signal, + recordEvent: options.recordEvent, + }); shimReplay = await startCommandShimReplay( scenario, options.temporaryRoot, started, + options.recordEvent, ); terminal = spawn(scenario.executable, [...scenario.arguments], { cwd: scenario.working_directory, @@ -252,6 +260,7 @@ const startCaptureRuntime = async (options: { started, limit: scenario.limits.processes, samples: options.samples, + recordEvent: options.recordEvent, }); return { replay, @@ -366,6 +375,8 @@ const completeCapture = async (options: { readonly exit: Awaited>; readonly signal?: AbortSignal; readonly initiallyTruncated: boolean; + readonly eventJournal: readonly ProcessCaptureEventJournalEntry[]; + readonly recordEvent: RecordProcessCaptureEvent; }): Promise => { const { runtime, scenario } = options; runtime.checkpoints.trigger("root_exit"); @@ -375,6 +386,7 @@ const completeCapture = async (options: { options.runId, selectCapturedProcessGroupIds(runtime.terminal.pid, options.samples), scenario.settle_ms, + options.recordEvent, ); runtime.checkpoints.trigger("settled"); const samplingPartial = (await runtime.stopSampler()).partial; @@ -414,6 +426,7 @@ const completeCapture = async (options: { shimEvents: normalizedShimEvents(runtime, scenario, options.temporaryRoot), settlement: { ...settlement, cleanup_outcome: "not_required" }, manifest, + eventJournal: options.eventJournal, }); }; @@ -430,6 +443,14 @@ const runProcessScenario = async ( ); const frames: TerminalFrame[] = []; const samples: ProcessSample[] = []; + const eventJournal: ProcessCaptureEventJournalEntry[] = []; + const recordEvent: RecordProcessCaptureEvent = (collection, index) => { + eventJournal.push({ + capture_order: eventJournal.length, + collection, + index, + }); + }; let runtime: StartedCaptureRuntime | undefined; const timers = new Set(); let capture: ProcessCapture | undefined; @@ -450,6 +471,7 @@ const runProcessScenario = async ( interactions, timers, dispatchedEventIndexes, + recordEvent, ...(signal === undefined ? {} : { signal }), }); stopSampler = runtime.stopSampler; @@ -462,6 +484,7 @@ const runProcessScenario = async ( timers, interactions, dispatchedEventIndexes, + recordEvent, }); capture = await completeCapture({ scenario, @@ -474,6 +497,8 @@ const runProcessScenario = async ( interactions, exit, initiallyTruncated: before.truncated, + eventJournal, + recordEvent, ...(signal === undefined ? {} : { signal }), }); } catch (cause: unknown) { diff --git a/src/application/ProcessSampling.ts b/src/application/ProcessSampling.ts index fe5f51a21..933ba60f5 100644 --- a/src/application/ProcessSampling.ts +++ b/src/application/ProcessSampling.ts @@ -1,7 +1,10 @@ import { execFile } from "node:child_process"; import { readFile, readdir } from "node:fs/promises"; import { promisify } from "node:util"; -import type { ProcessSample } from "../domain/processCapture.js"; +import type { + ProcessSample, + RecordProcessCaptureEvent, +} from "../domain/processCapture.js"; import { readProcessRunId } from "../process/ProcessOwnership.js"; const execFileAsync = promisify(execFile); @@ -452,6 +455,7 @@ export const startProcessSampler = (options: { readonly started: number; readonly limit: number; readonly samples: ProcessSample[]; + readonly recordEvent?: RecordProcessCaptureEvent; }): (() => Promise<{ readonly partial: boolean }>) => { const { rootPid, runId, started, limit, samples } = options; const identities = new Map(); @@ -501,7 +505,9 @@ export const startProcessSampler = (options: { continue; } lastObservations.set(value.pid, observation); + const index = samples.length; samples.push(value); + options.recordEvent?.("process_samples", index); } }) .catch((cause: unknown) => { diff --git a/src/application/TerminalRenderer.ts b/src/application/TerminalRenderer.ts index 562dae3af..314f7532c 100644 --- a/src/application/TerminalRenderer.ts +++ b/src/application/TerminalRenderer.ts @@ -1,6 +1,9 @@ import { createRequire } from "node:module"; -import type { RenderedTerminalFrame } from "../domain/processCapture.js"; +import type { + RecordProcessCaptureEvent, + RenderedTerminalFrame, +} from "../domain/processCapture.js"; const require = createRequire(import.meta.url); // SAFETY: both pinned xterm packages publish CommonJS at runtime and matching declarations. @@ -17,6 +20,7 @@ interface TerminalRendererOptions { readonly maxFrames: number; readonly maxBytes: number; readonly normalize: (value: string) => string; + readonly recordEvent?: RecordProcessCaptureEvent; } /** Owns one headless terminal and serializes writes into deterministic frames. */ @@ -112,8 +116,9 @@ export class TerminalRenderer { return; } this.#capturedBytes += bytes; + const sequence = this.#frames.length; this.#frames.push({ - sequence: this.#frames.length, + sequence, at_ms: atMs, columns: this.#terminal.cols, rows: this.#terminal.rows, @@ -123,5 +128,6 @@ export class TerminalRenderer { lines, serialized_state: serializedState, }); + this.options.recordEvent?.("rendered_frames", sequence); } } diff --git a/src/domain/processCapture.ts b/src/domain/processCapture.ts index 636ade6e4..f32d5567a 100644 --- a/src/domain/processCapture.ts +++ b/src/domain/processCapture.ts @@ -110,6 +110,35 @@ export interface ProtocolEvent { | "limit_exhausted"; } +/** Capture collection whose members participate in global observation order. */ +export const PROCESS_CAPTURE_EVENT_COLLECTIONS = [ + "frames", + "rendered_frames", + "interaction_events", + "lifecycle", + "process_samples", + "filesystem_checkpoints", + "shim_events", + "protocol_events", + "replay_transitions", +] as const; + +export type ProcessCaptureEventCollection = + (typeof PROCESS_CAPTURE_EVENT_COLLECTIONS)[number]; + +/** One reference from global observation order into a capture collection. */ +export interface ProcessCaptureEventJournalEntry { + readonly capture_order: number; + readonly collection: ProcessCaptureEventCollection; + readonly index: number; +} + +/** Shared observation callback used by every process-capture producer. */ +export type RecordProcessCaptureEvent = ( + collection: ProcessCaptureEventCollection, + index: number, +) => void; + /** * Process Capture v4 observation set. * @@ -156,6 +185,12 @@ export interface ProcessCapture { readonly shim_events: readonly ShimEvent[]; readonly protocol_events: readonly ProtocolEvent[]; readonly replay_transitions: readonly ReplayTransitionRecord[]; + /** + * Global observation order across independently recorded collections. + * + * Absence is accepted for captures written before this journal existed. + */ + readonly event_journal?: readonly ProcessCaptureEventJournalEntry[]; readonly files_before: readonly FileState[]; readonly files_after: readonly FileState[]; readonly filesystem_effects: readonly FileEffect[]; @@ -328,6 +363,15 @@ export const processCaptureSchema: z.ZodType = z.object({ }), ) .default([]), + event_journal: z + .array( + z.object({ + capture_order: z.number().int().nonnegative(), + collection: z.enum(PROCESS_CAPTURE_EVENT_COLLECTIONS), + index: z.number().int().nonnegative(), + }), + ) + .default([]), files_before: z.array(fileStateSchema), files_after: z.array(fileStateSchema), filesystem_effects: z.array(fileEffectSchema), @@ -383,3 +427,9 @@ export { PROCESS_COMPARISON_DIMENSIONS, processCaptureComparisonSchema, } from "./processComparison.js"; +export { + compareProcessTraces, + processTraceComparisonResultSchema, + processTraceSourceSchema, + processTraceSpecificationSchema, +} from "./processTraceComparison.js"; diff --git a/src/domain/processCaptureValidation.ts b/src/domain/processCaptureValidation.ts index 1fd4b99a5..32bc73193 100644 --- a/src/domain/processCaptureValidation.ts +++ b/src/domain/processCaptureValidation.ts @@ -72,6 +72,54 @@ const validateOrdering = ( require(orderedTimestamps(values), name, "timestamps must be ordered"); }; +const journalCollectionSizes = ( + capture: ProcessCapture, +): Readonly< + Record< + NonNullable[number]["collection"], + number + > +> => ({ + frames: capture.frames.length, + rendered_frames: capture.rendered_frames.length, + interaction_events: capture.interaction_events.length, + lifecycle: 2, + process_samples: capture.process_samples.length, + filesystem_checkpoints: capture.filesystem_checkpoints.length, + shim_events: capture.shim_events.length, + protocol_events: capture.protocol_events.length, + replay_transitions: capture.replay_transitions.length, +}); + +const validateEventJournal = ( + capture: ProcessCapture, + require: RequireInvariant, +): void => { + const journal = capture.event_journal ?? []; + if (journal.length === 0) return; + const sizes = journalCollectionSizes(capture); + const references = new Set(); + for (const [position, entry] of journal.entries()) { + require(entry.capture_order === + position, `event_journal.${String(position)}.capture_order`, "capture order must be contiguous from zero"); + require(entry.index < + sizes[ + entry.collection + ], `event_journal.${String(position)}.index`, "journal reference is outside its capture collection"); + const reference = `${entry.collection}:${String(entry.index)}`; + require(!references.has( + reference, + ), `event_journal.${String(position)}`, "journal references must be unique"); + references.add(reference); + } + const expectedSize = Object.values(sizes).reduce( + (total, size) => total + size, + 0, + ); + require(journal.length === + expectedSize, "event_journal", "nonempty journal must reference every captured observation"); +}; + const validateLifecycle = ( capture: ProcessCapture, require: RequireInvariant, @@ -262,6 +310,7 @@ export const validateProcessCapture = ( }; validateCommitments(capture, require); validateOrdering(capture, require); + validateEventJournal(capture, require); validateLifecycle(capture, require); validateShimEvents(capture, require); validateReplayEvents(capture, require); diff --git a/tests/processCapture.test.ts b/tests/processCapture.test.ts index c7298e0cd..ad25d6176 100644 --- a/tests/processCapture.test.ts +++ b/tests/processCapture.test.ts @@ -15,6 +15,7 @@ import { } from "../src/application/ProcessHarness.js"; import { snapshotRoots } from "../src/application/FilesystemSnapshot.js"; import { + buildCaptureResult, prepareProcessCapture, type ProcessPreparationHost, } from "../src/application/ProcessCaptureLifecycle.js"; @@ -100,6 +101,7 @@ const emptyCapture = (): ProcessCapture => { shim_events: [], protocol_events: [], replay_transitions: [], + event_journal: [], files_before: [], files_after: [], filesystem_effects: [], @@ -121,6 +123,7 @@ describe("process capture domain", () => { }; it("returns detached terminal and filesystem checkpoint observations", async () => { + const observed: string[] = []; const renderer = new TerminalRenderer({ columns: 40, rows: 12, @@ -128,9 +131,14 @@ describe("process capture domain", () => { maxFrames: 10, maxBytes: 10_000, normalize: (value) => value, + recordEvent: (collection, index) => + observed.push(`${collection}:${String(index)}`), }); - renderer.write("A", 0); + renderer.write("A", 20); + renderer.resize(40, 12, 10); const frames = await renderer.frames(); + expect(frames.map(({ at_ms }) => at_ms)).toEqual([20, 10]); + expect(observed).toEqual(["rendered_frames:0", "rendered_frames:1"]); if (frames[0] !== undefined) Reflect.set(frames[0], "cursor_x", 999); expect((await renderer.frames())[0]?.cursor_x).not.toBe(999); await renderer.dispose(); @@ -140,7 +148,7 @@ describe("process capture domain", () => { scenario, Date.now(), { files: [], truncated: false }, - undefined, + { signal: undefined }, ); const first = await checkpoints.finish({ files: [], truncated: false }); if (first[0] !== undefined) Reflect.set(first[0], "name", "tampered"); @@ -149,6 +157,62 @@ describe("process capture domain", () => { await checkpoints.dispose(); }); + it("preserves rendered observation order instead of timestamp sorting", () => { + const capture = emptyCapture(); + const renderedFrames: ProcessCapture["rendered_frames"] = [ + { + sequence: 0, + at_ms: 20, + columns: 1, + rows: 1, + cursor_x: 0, + cursor_y: 0, + active_buffer: "normal", + lines: ["first"], + serialized_state: "first", + }, + { + sequence: 1, + at_ms: 10, + columns: 1, + rows: 1, + cursor_x: 0, + cursor_y: 0, + active_buffer: "normal", + lines: ["second"], + serialized_state: "second", + }, + ]; + const result = buildCaptureResult({ + frames: [], + exit: { exitCode: 0, reason: "exited" }, + samples: [], + replay: { + httpUrl: "http://127.0.0.1", + websocketUrl: "ws://127.0.0.1/ws", + events: [], + transitions: [], + truncated: false, + close: () => Promise.resolve(), + }, + before: { files: [], truncated: false }, + after: { files: [], truncated: false }, + truncated: false, + scenario: parseProcessScenario(base), + rootPid: 1, + samplingPartial: false, + renderedFrames, + interactions: [], + checkpoints: capture.filesystem_checkpoints, + shimEvents: [], + settlement: capture.settlement, + manifest: capture.manifest, + eventJournal: [], + }); + + expect(result.rendered_frames).toEqual(renderedFrames); + }); + it("cleans the temporary root when capture home creation fails", async () => { const cleaned: string[] = []; const host: ProcessPreparationHost = { @@ -426,6 +490,56 @@ describe("process capture domain", () => { ).toThrow("executable_sha256"); }); + it("accepts old captures without a journal and validates complete journals", () => { + const capture = emptyCapture(); + const { event_journal: _eventJournal, ...oldCapture } = capture; + expect(parseProcessCapture(oldCapture).event_journal).toEqual([]); + + const eventJournal = [ + { capture_order: 0, collection: "filesystem_checkpoints", index: 0 }, + { capture_order: 1, collection: "lifecycle", index: 0 }, + { capture_order: 2, collection: "lifecycle", index: 1 }, + { capture_order: 3, collection: "filesystem_checkpoints", index: 1 }, + ] as const; + expect( + validateProcessCapture({ ...capture, event_journal: eventJournal }), + ).toEqual([]); + + for (const [candidate, message] of [ + [ + eventJournal.map((entry, index) => + index === 1 ? { ...entry, capture_order: 2 } : entry, + ), + "contiguous", + ], + [ + eventJournal.map((entry, index) => + index === 3 + ? { ...entry, collection: "lifecycle" as const, index: 1 } + : entry, + ), + "unique", + ], + [eventJournal.slice(0, 3), "every captured observation"], + [ + eventJournal.map((entry, index) => + index === 3 ? { ...entry, index: 2 } : entry, + ), + "outside", + ], + ] as const) { + expect( + validateProcessCapture({ ...capture, event_journal: candidate }), + ).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + message: expect.stringContaining(message), + }), + ]), + ); + } + }); + it.each([ ["machine limit", 1, 5, 10], ["transition use limit", 5, 1, 10], @@ -1272,6 +1386,11 @@ describe("process capture adapter", () => { expect(capture.value.manifest.replay_plan_sha256).toBe( digestProcessCommitment(scenario.replay), ); + expect( + capture.value.event_journal?.filter( + ({ collection }) => collection === "replay_transitions", + ), + ).toHaveLength(capture.value.replay_transitions.length); } finally { await rm(root, { recursive: true, force: true }); }