mirror/rea
1
0
Fork 0
mirror of https://github.com/morluto/rea synced 2026-10-25 22:08:11 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix(process): attribute finite trace omissions and cleanup states

Browse Source
This commit is contained in:
morluto
2026-10-15 03:06:19 +08:00
parent cb4ff50202
commit fc7e41ff3b
7 changed files with 496 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/domain/process/processTraceComparison.test.ts
+320 -31
View File
@@ -1,10 +1,12 @@
import { describe, expect, it } from "vitest";
import { jsonValueSchema } from "../jsonValue.js";
import { parseProcessCapture } from "./processCaptureParsing.js";
import type { ProcessTraceSpecification } from "./processTraceSpecification.js";
import { processTraceSpecificationSchema as specificationSchema } from "./processTraceSpecification.js";
import { processTraceComparisonResultSchema } from "./processTraceEvaluation.js";
import {
capture,
emptyCapture,
partialSpecification,
processStarted,
terminal,
@@ -13,6 +15,35 @@ import {
import { compareProcessCaptures } from "./processComparison.js";
import { compareProcessTraces } from "./processTraceComparison.js";
const traced = (
left: ReturnType<typeof capture>,
right: ReturnType<typeof capture>,
) =>
compareProcessCaptures(left, right, {
traceSpecification: partialSpecification(),
});
const withPayloads = (data: string, command: string) =>
capture({
...values(["terminal", "process"]),
frames: [{ ...terminal, data }],
process_samples: [{ ...processStarted, command }],
});
type FiniteTraceLanguage = Extract<
ProcessTraceSpecification["language"],
{ kind: "finite_traces" }
>;
const finiteTraceSpecification = (
events: ProcessTraceSpecification["events"],
variants: FiniteTraceLanguage["variants"],
): ProcessTraceSpecification => ({
events,
language: { kind: "finite_traces", variants },
});
describe("generic capture trace comparison", () => {
it("requires explicit partial ordering and never infers causality from timestamps", () => {
const specification = partialSpecification();
@@ -73,8 +104,8 @@ describe("generic capture trace comparison", () => {
{ collection: "frames", index: 1 },
],
});
const specification: ProcessTraceSpecification = {
events: [
const specification = finiteTraceSpecification(
[
{
id: "tick1",
source: "terminal_raw",
@@ -88,14 +119,11 @@ describe("generic capture trace comparison", () => {
cardinality: { kind: "optional" },
},
],
language: {
kind: "finite_traces",
variants: [
{ id: "one", trace: ["tick1"] },
{ id: "two", trace: ["tick1", "tick2"] },
],
},
};
[
{ id: "one", trace: ["tick1"] },
{ id: "two", trace: ["tick1", "tick2"] },
],
);
expect(compareProcessTraces(one, two, specification)).toMatchObject({
verdict: "equivalent",
left: { matched_variant: "one" },
@@ -108,8 +136,8 @@ describe("generic capture trace comparison", () => {
// echoes those identifiers verbatim, so both must accept the same values
// rather than handing the caller a raw schema error for its own input.
const eventId = `e${"0".repeat(64)}`;
const specification: ProcessTraceSpecification = {
events: [
const specification = finiteTraceSpecification(
[
{
id: eventId,
source: "terminal_raw",
@@ -117,11 +145,8 @@ describe("generic capture trace comparison", () => {
cardinality: { kind: "required" },
},
],
language: {
kind: "finite_traces",
variants: [{ id: "only", trace: [eventId] }],
},
};
[{ id: "only", trace: [eventId] }],
);
expect(specificationSchema.safeParse(specification).success).toBe(true);
const left = capture(values(["terminal", "process"]));
expect(compareProcessTraces(left, left, specification)).toMatchObject({
@@ -131,6 +156,53 @@ describe("generic capture trace comparison", () => {
});
});
describe("finite trace variant mismatch attribution", () => {
it("reports tied best-prefix omissions across variants", () => {
const observed = capture({
frames: [terminal],
process_samples: [],
event_journal: [{ collection: "frames", index: 0 }],
});
const specification = finiteTraceSpecification(
[
{
id: "ready",
source: "terminal_raw",
exact: terminal,
cardinality: { kind: "required" },
},
{
id: "worker",
source: "process",
exact: processStarted,
cardinality: { kind: "optional" },
},
{
id: "worker_alt",
source: "process",
exact: { ...processStarted, command: "alternate-worker" },
cardinality: { kind: "optional" },
},
],
[
{ id: "ready_worker", trace: ["ready", "worker"] },
{ id: "ready_worker_alt", trace: ["ready", "worker_alt"] },
],
);
expect(
compareProcessTraces(observed, observed, specification),
).toMatchObject({
verdict: "nonconforming",
diagnostic: {
kind: "trace",
event_ids: ["worker", "worker_alt"],
locations: [],
},
});
});
});
describe("process trace constraints and evidence sufficiency", () => {
it("enforces exact and range cardinality for a repeated terminal observation", () => {
const repeated = (count: number) =>
@@ -288,38 +360,255 @@ describe("unknown process trace evidence", () => {
});
describe("process trace dimension attribution", () => {
const traced = (
left: ReturnType<typeof capture>,
right: ReturnType<typeof capture>,
) =>
compareProcessCaptures(left, right, {
traceSpecification: partialSpecification(),
it("applies passing filesystem trace timestamp omissions to exact comparison", () => {
const base = capture(values(["terminal", "process"]));
const left = base;
const right = parseProcessCapture({
...base,
filesystem_checkpoints: base.filesystem_checkpoints.map((checkpoint) =>
checkpoint.name === "after_settlement"
? { ...checkpoint, at_ms: checkpoint.at_ms + 10 }
: checkpoint,
),
});
const make = (data: string, command: string) =>
capture({
...values(["terminal", "process"]),
frames: [{ ...terminal, data }],
process_samples: [{ ...processStarted, command }],
const specification = finiteTraceSpecification(
emptyCapture.filesystem_checkpoints.map(
({ at_ms: _atMs, ...checkpoint }) => ({
id: checkpoint.name,
source: "filesystem" as const,
exact: jsonValueSchema.parse(checkpoint),
ignore_fields: ["at_ms"],
cardinality: { kind: "required" as const },
}),
),
[{ id: "checkpoints", trace: ["before", "after_settlement"] }],
);
expect(compareProcessCaptures(left, right)).toMatchObject({
filesystem: "changed",
first_divergence: { status: "found", dimension: "filesystem" },
});
expect(
compareProcessCaptures(left, right, {
traceSpecification: specification,
}),
).toMatchObject({
trace: { verdict: "equivalent" },
filesystem: "unchanged",
first_divergence: { status: "none" },
});
const finalFile = {
path: "root_0:created.txt",
type: "file" as const,
mode: 0o644,
size: 4,
sha256: "a".repeat(64),
symlink_target: null,
};
const created = {
path: finalFile.path,
status: "created" as const,
before: null,
after: finalFile,
};
const changedRight = parseProcessCapture({
...right,
filesystem_checkpoints: right.filesystem_checkpoints.map((checkpoint) =>
checkpoint.name === "after_settlement"
? { ...checkpoint, files: [finalFile], effects: [created] }
: checkpoint,
),
files_after: [finalFile],
filesystem_effects: [created],
truncation_details: {
...right.truncation_details,
filesystem_after: {
...right.truncation_details.filesystem_after,
hashed_bytes: finalFile.size,
},
},
});
expect(
compareProcessCaptures(left, changedRight, {
traceSpecification: specification,
}),
).toMatchObject({
trace: { verdict: "different" },
filesystem: "changed",
first_divergence: { status: "found", dimension: "filesystem" },
});
});
it("keeps equal dimensions unchanged when one side fails a predicate", () => {
expect(
traced(make("Ready", "worker"), make("Different", "worker")),
traced(
withPayloads("Ready", "worker"),
withPayloads("Different", "worker"),
),
).toMatchObject({
trace: { verdict: "different" },
terminal: "changed",
process: "unchanged",
});
expect(
traced(make("Ready", "worker"), make("Ready", "other-worker")),
traced(
withPayloads("Ready", "worker"),
withPayloads("Ready", "other-worker"),
),
).toMatchObject({
trace: { verdict: "different" },
terminal: "unchanged",
process: "changed",
});
});
});
it("still attributes an ordering difference to both ordered dimensions", () => {
describe("finite trace dimension attribution", () => {
it("attributes a finite-trace omission to the omitted event dimension", () => {
const left = capture({
frames: [terminal],
process_samples: [],
event_journal: [{ collection: "frames", index: 0 }],
});
const right = capture(values(["terminal", "process"]));
const specification = finiteTraceSpecification(
[
{
id: "ready",
source: "terminal_raw",
exact: terminal,
cardinality: { kind: "required" },
},
{
id: "worker",
source: "process",
exact: processStarted,
cardinality: { kind: "optional" },
},
],
[{ id: "ready_worker", trace: ["ready", "worker"] }],
);
expect(
compareProcessCaptures(left, right, {
traceSpecification: specification,
}),
).toMatchObject({
status: "changed",
process: "changed",
terminal: "unchanged",
trace: {
verdict: "different",
diagnostic: { kind: "trace", event_ids: ["worker"], locations: [] },
},
first_divergence: {
status: "found",
dimension: "process",
index: 0,
},
});
});
});
describe("unordered finite trace mismatch attribution", () => {
it("attributes only the missing member of an unordered group", () => {
const db = { sequence: 1, at_ms: 901, data: "DB" };
const alert = { ...processStarted, pid: 2, command: "other-worker" };
const left = capture({
frames: [terminal],
process_samples: [processStarted],
event_journal: [
{ collection: "frames", index: 0 },
{ collection: "process_samples", index: 0 },
],
});
const right = capture({
frames: [terminal, db],
process_samples: [processStarted],
event_journal: [
{ collection: "frames", index: 0 },
{ collection: "process_samples", index: 0 },
{ collection: "frames", index: 1 },
],
});
const specification = finiteTraceSpecification(
[
{
id: "ready",
source: "terminal_raw",
exact: terminal,
cardinality: { kind: "required" },
},
{
id: "worker",
source: "process",
exact: processStarted,
cardinality: { kind: "optional" },
},
{
id: "db",
source: "terminal_raw",
exact: db,
cardinality: { kind: "optional" },
},
{
id: "alert",
source: "process",
exact: alert,
cardinality: { kind: "optional" },
},
],
[
{
id: "ready_then_work",
trace: ["ready", { unordered: ["worker", "db"] }],
},
],
);
expect(
compareProcessCaptures(left, right, {
traceSpecification: specification,
}),
).toMatchObject({
trace: {
verdict: "different",
diagnostic: {
side: "left",
event_ids: ["db"],
locations: [],
},
},
terminal: "changed",
process: "unchanged",
first_divergence: { status: "found", dimension: "terminal", index: 1 },
});
const unexpected = capture({
frames: [terminal],
process_samples: [processStarted, alert],
event_journal: [
{ collection: "frames", index: 0 },
{ collection: "process_samples", index: 0 },
{ collection: "process_samples", index: 1 },
],
});
expect(
compareProcessTraces(unexpected, unexpected, specification),
).toMatchObject({
verdict: "nonconforming",
diagnostic: {
kind: "trace",
event_ids: ["db", "alert"],
locations: [{ collection: "process_samples", index: 1 }],
},
});
});
});
describe("partial-order dimension attribution", () => {
it("attributes an ordering difference to both ordered dimensions", () => {
expect(
traced(
capture(values(["terminal", "process"])),
src/domain/process/processTraceDimensionProjection.ts
+1 -1
View File
@@ -46,7 +46,7 @@ export const traceCoversObservedDimension = (
case "exit":
return sources.has("lifecycle");
case "filesystem":
return false;
return sources.has("filesystem");
case "process":
return sources.has("process");
}
src/domain/process/processTraceEvaluation.ts
+91 -25
View File
@@ -123,29 +123,92 @@ const failure = (input: FailureInput): EvaluatedSide => ({
},
});
type FiniteTraceEvaluation =
| { readonly status: "pass"; readonly variant: string }
| {
readonly status: "fail";
readonly eventIds: readonly string[];
readonly locationIndexes: readonly number[];
};
/** Match variants once, retaining the best-prefix mismatch for attribution. */
const evaluateFiniteTrace = (
ids: readonly string[],
specification: ProcessTraceSpecification,
): string | null => {
if (specification.language.kind !== "finite_traces") return null;
for (const variant of specification.language.variants) {
let offset = 0;
let accepted = true;
for (const token of variant.trace) {
if (typeof token === "string") {
if (ids[offset] !== token) accepted = false;
offset += 1;
continue;
}
const actual = ids.slice(offset, offset + token.unordered.length).sort();
const expected = [...token.unordered].sort();
if (canonicalTraceJson(actual) !== canonicalTraceJson(expected))
accepted = false;
offset += token.unordered.length;
language: Extract<
ProcessTraceSpecification["language"],
{ kind: "finite_traces" }
>,
): FiniteTraceEvaluation => {
let bestPrefix = -1;
const bestEventIds = new Set<string>();
const bestLocationIndexes = new Set<number>();
const recordMismatch = (
offset: number,
missing: readonly string[],
unexpected: readonly { readonly id: string; readonly index: number }[],
): void => {
if (offset > bestPrefix) {
bestPrefix = offset;
bestEventIds.clear();
bestLocationIndexes.clear();
}
if (accepted && offset === ids.length) return variant.id;
if (offset !== bestPrefix) return;
for (const eventId of missing) bestEventIds.add(eventId);
for (const { id, index } of unexpected) {
bestEventIds.add(id);
bestLocationIndexes.add(index);
}
};
for (const variant of language.variants) {
let offset = 0;
let mismatched = false;
for (const token of variant.trace) {
const expected = typeof token === "string" ? [token] : token.unordered;
const actual = ids.slice(offset, offset + expected.length);
const remaining = new Map<string, number>();
for (const eventId of expected)
remaining.set(eventId, (remaining.get(eventId) ?? 0) + 1);
const unexpected: Array<{ readonly id: string; readonly index: number }> =
[];
for (const [index, eventId] of actual.entries()) {
const count = remaining.get(eventId) ?? 0;
if (count === 0)
unexpected.push({ id: eventId, index: offset + index });
else if (count === 1) remaining.delete(eventId);
else remaining.set(eventId, count - 1);
}
const missing = [...remaining].flatMap(([eventId, count]) =>
Array.from({ length: count }, () => eventId),
);
if (
missing.length > 0 ||
unexpected.length > 0 ||
actual.length < expected.length
) {
recordMismatch(offset, missing, unexpected);
mismatched = true;
break;
}
offset += expected.length;
}
if (mismatched) continue;
if (offset === ids.length) return { status: "pass", variant: variant.id };
const unexpected = ids[offset];
recordMismatch(
offset,
[],
unexpected === undefined ? [] : [{ id: unexpected, index: offset }],
);
}
return null;
return {
status: "fail",
eventIds: [...bestEventIds],
locationIndexes: [...bestLocationIndexes],
};
};
type MatchedTrace = {
@@ -391,20 +454,23 @@ const evaluateMatchedTrace = (
if (specification.language.kind === "partial_order")
return evaluatePartialOrder(trace, specification.language);
const ids = trace.rawTrace.map(({ event_id }) => event_id);
const variant = evaluateFiniteTrace(ids, specification);
return variant === null
const evaluation = evaluateFiniteTrace(ids, specification.language);
return evaluation.status === "fail"
? failure({
kind: "trace",
message: "Observed events do not match any finite trace variant.",
eventIds: ids,
locations: trace.rawTrace.map(({ location }) => location),
eventIds: evaluation.eventIds,
locations: evaluation.locationIndexes.flatMap((index) => {
const location = trace.rawTrace[index]?.location;
return location === undefined ? [] : [location];
}),
rawTrace: trace.rawTrace,
})
: {
result: {
status: "pass",
matched_variant: variant,
satisfied_constraints: [`variant:${variant}`],
matched_variant: evaluation.variant,
satisfied_constraints: [`variant:${evaluation.variant}`],
raw_trace: trace.rawTrace,
},
diagnostic: null,
src/process/capture/ProcessCaptureProgress.test.ts
+27
View File
@@ -110,4 +110,31 @@ it("still delivers terminal status after an observer rejects an update", async (
completed: 2,
terminal: true,
});
expect(delivered.at(-1)?.message).toContain(
"owned_process_group=unverified terminal_renderer=unverified temporary_root=unverified",
);
});
it("preserves known cleanup states in terminal progress", async () => {
const delivered: Parameters<ProcessCaptureProgress["report"]>[0][] = [];
const tracker = createProcessCaptureProgressTracker({
async report(update) {
delivered.push(update);
},
});
await tracker.report({
phase: "cleanup",
counts: { frames: 0, samples: 0, interactions: 0 },
disposition: "failed",
terminal: true,
cleanup: {
owned_process_group: { state: "not_required", reason: null },
terminal_renderer: { state: "not_required", reason: null },
temporary_root: { state: "cleaned", reason: null },
},
});
expect(delivered[0]?.message).toContain(
"owned_process_group=not_required terminal_renderer=not_required temporary_root=cleaned",
);
});
src/process/capture/ProcessCaptureProgress.ts
+5 -3
View File
@@ -53,9 +53,11 @@ const formatProgressMessage = (
const dispositionText =
disposition === undefined ? "" : ` disposition=${disposition}`;
const cleanupText =
cleanup === undefined
? ""
: ` owned_process_group=${cleanup.owned_process_group.state} terminal_renderer=${cleanup.terminal_renderer.state} temporary_root=${cleanup.temporary_root.state}`;
cleanup !== undefined
? ` owned_process_group=${cleanup.owned_process_group.state} terminal_renderer=${cleanup.terminal_renderer.state} temporary_root=${cleanup.temporary_root.state}`
: disposition === undefined
? ""
: " owned_process_group=unverified terminal_renderer=unverified temporary_root=unverified";
return `${status}${dispositionText}${cleanupText}`;
};
src/process/capture/ProcessHarness.ts
+4
View File
@@ -672,6 +672,10 @@ const runProcessScenario = async (
counts: progressCounts(),
terminal: true,
disposition: signal?.aborted === true ? "cancelled" : "failed",
...(cause instanceof ProcessCaptureError &&
cause.cleanupReport !== undefined
? { cleanup: cause.cleanupReport }
: {}),
});
throw cause;
}
tests/boundary/process/processCaptureProgress.test.ts
+48 -2
View File
@@ -99,12 +99,20 @@ const writeScenario = async (
const runCaptureCli = async (
scenarioPath: string,
options: { readonly stopOnRunning?: boolean } = {},
options: {
readonly stopOnRunning?: boolean;
readonly env?: NodeJS.ProcessEnv;
} = {},
): Promise<CliCapture> => {
const child = spawn(
process.execPath,
["scripts/rea.mjs", "capture-process", scenarioPath, "--json"],
{ stdio: ["ignore", "pipe", "pipe"] },
{
stdio: ["ignore", "pipe", "pipe"],
...(options.env === undefined
? {}
: { env: { ...process.env, ...options.env } }),
},
);
let stdout = "";
let stderr = "";
@@ -274,6 +282,44 @@ itWithCaptureCapability(
30_000,
);
itWithCaptureCapability(
"reports unknown resource cleanup when temporary-root preparation fails",
async () => {
const scenario = await writeScenario(
"rea-process-progress-prepare-failed-",
{
executable: process.execPath,
arguments: childScript("process.stdout.write('ready\\n')"),
timeout_ms: 5_000,
},
);
const missingTempRoot = join(
await createTestTempDirectory("rea-process-progress-missing-tmpdir-"),
"missing",
);
const capture = await runCaptureCli(scenario, {
env: {
TMPDIR: missingTempRoot,
TMP: missingTempRoot,
TEMP: missingTempRoot,
},
});
expect(capture.exitCode).toBe(1);
expect(capture.lines.at(-1)?.update).toMatchObject({
phase: "cleanup",
terminal: true,
details: {
disposition: "failed",
owned_process_group: "unverified",
terminal_renderer: "unverified",
temporary_root: "unverified",
},
});
},
30_000,
);
itWithCaptureCapability(
"bounds stderr progress while a busy child emits many output chunks",
async () => {