shifathosn
/
rea
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(process): record global capture event order

This commit is contained in:
morluto 2026-07-22 20:07:49 +00:00
parent 3df9a07413
commit 8bcdb98689
11 changed files with 349 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/application/CommandShimReplay.ts
View File

@ -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<CommandShimReplay> => {
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;

10
src/application/LoopbackReplay.ts
View File

@ -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<void> => {
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<LoopbackReplay> => {
const recorder = new ReplayRecorder(scenario);
const recorder = new ReplayRecorder(scenario, recordEvent);
const server = createServer(createHttpHandler(recorder));
const websocket = createWebSocketReplay(server, recorder);
let closePromise: Promise<void> | undefined;

27
src/application/LoopbackReplayRecorder.ts
View File

@ -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<ReplayMachineRuntime["dispatch"]>[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<ProtocolEvent, "sequence">): 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);
}
}

37
src/application/ProcessCaptureLifecycle.ts
View File

@ -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<NodeJS.Timeout>;
readonly interactions: InteractionEvent[];
readonly dispatchedEventIndexes: ReadonlySet<number>;
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<Omit<ProcessCapture["settlement"], "cleanup_outcome">> => {
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<void>((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);
});

17
src/application/ProcessCheckpoints.ts
View File

@ -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<string, number>();
readonly #terminalTails = new Map<string, string>();
readonly #timers = new Set<NodeJS.Timeout>();
readonly #signal: AbortSignal | undefined;
readonly #recordEvent: RecordProcessCaptureEvent;
#pending: Promise<void> = 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);
}

41
src/application/ProcessHarness.ts
View File

@ -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<number>;
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<NodeJS.Timeout>;
readonly dispatchedEventIndexes: Set<number>;
readonly recordEvent: RecordProcessCaptureEvent;
readonly signal?: AbortSignal;
}): Promise<StartedCaptureRuntime> => {
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<ReturnType<typeof awaitTerminalExit>>;
readonly signal?: AbortSignal;
readonly initiallyTruncated: boolean;
readonly eventJournal: readonly ProcessCaptureEventJournalEntry[];
readonly recordEvent: RecordProcessCaptureEvent;
}): Promise<ProcessCapture> => {
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<NodeJS.Timeout>();
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) {

8
src/application/ProcessSampling.ts
View File

@ -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<number, string>();
@ -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) => {

10
src/application/TerminalRenderer.ts
View File

@ -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);
}
}

50
src/domain/processCapture.ts
View File

@ -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<ProcessCapture> = 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";

49
src/domain/processCaptureValidation.ts
View File

@ -72,6 +72,54 @@ const validateOrdering = (
require(orderedTimestamps(values), name, "timestamps must be ordered");
};
const journalCollectionSizes = (
capture: ProcessCapture,
): Readonly<
Record<
NonNullable<ProcessCapture["event_journal"]>[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<string>();
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);

123
tests/processCapture.test.ts
View File

@ -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 });
}