From 9ab7aa320c2f234e86d16ff8535b11768a0e080e Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:45:27 +0000 Subject: [PATCH] feat(process): compare repeatable paired experiments --- src/application/ProcessPairedExperiment.ts | 140 ++++++++++++ src/domain/processComparison.ts | 27 ++- src/domain/processPairedExperiment.ts | 206 +++++++++++++++++ tests/processPairedExperiment.test.ts | 247 +++++++++++++++++++++ 4 files changed, 619 insertions(+), 1 deletion(-) create mode 100644 src/application/ProcessPairedExperiment.ts create mode 100644 src/domain/processPairedExperiment.ts create mode 100644 tests/processPairedExperiment.test.ts diff --git a/src/application/ProcessPairedExperiment.ts b/src/application/ProcessPairedExperiment.ts new file mode 100644 index 0000000000..4e4cdb1aba --- /dev/null +++ b/src/application/ProcessPairedExperiment.ts @@ -0,0 +1,140 @@ +import { + processCaptureCancelled, + type ProcessCaptureError, +} from "./ProcessCaptureError.js"; +import { captureProcessScenario } from "./ProcessHarness.js"; +import type { + ProcessCapture, + ProcessExecutionPolicy, + ProcessScenario, +} from "../domain/processCapture.js"; +import { compareProcessCaptures } from "../domain/processComparison.js"; +import { PROCESS_COMPARISON_DIMENSIONS } from "../domain/processComparison.js"; +import { + analyzeProcessRepeatability, + bindProcessScenario, + type ProcessPairedExperiment, +} from "../domain/processPairedExperiment.js"; +import { err, ok, type Result } from "../domain/result.js"; + +/** Capture seam used by paired process experiments. */ +export interface ProcessCapturePort { + capture( + scenario: ProcessScenario, + policy: ProcessExecutionPolicy, + signal?: AbortSignal, + ): Promise>; +} + +/** Repeatability evidence and an optional stable cross-side comparison. */ +export interface PairedProcessExperimentResult { + readonly authority_runs: readonly ProcessCapture[]; + readonly candidate_runs: readonly ProcessCapture[]; + readonly authority_repeatability: ReturnType< + typeof analyzeProcessRepeatability + >; + readonly candidate_repeatability: ReturnType< + typeof analyzeProcessRepeatability + >; + readonly cross_side: ReturnType | null; + readonly cross_side_blocked_reason: string | null; +} + +const productionCapturePort: ProcessCapturePort = { + capture: captureProcessScenario, +}; + +const captureRepeats = async (input: { + readonly scenario: ProcessScenario; + readonly repeatCount: number; + readonly policy: ProcessExecutionPolicy; + readonly port: ProcessCapturePort; + readonly signal?: AbortSignal; +}): Promise> => { + const captures: ProcessCapture[] = []; + for (let index = 0; index < input.repeatCount; index += 1) { + if (input.signal?.aborted === true) return err(processCaptureCancelled()); + const capture = await input.port.capture( + input.scenario, + input.policy, + input.signal, + ); + if (!capture.ok) return err(capture.error); + captures.push(capture.value); + } + return ok(captures); +}; + +/** Execute repeatability-first authority/candidate captures under one contract. */ +export const runPairedProcessExperiment = async ( + experiment: ProcessPairedExperiment, + policy: ProcessExecutionPolicy, + options: { + readonly signal?: AbortSignal; + readonly capturePort?: ProcessCapturePort; + readonly now?: () => number; + } = {}, +): Promise> => { + const port = options.capturePort ?? productionCapturePort; + const authorityScenario = bindProcessScenario( + experiment.shared_scenario, + experiment.authority, + ); + const candidateScenario = bindProcessScenario( + experiment.shared_scenario, + experiment.candidate, + ); + const authorityRuns = await captureRepeats({ + scenario: authorityScenario, + repeatCount: experiment.repeat_count, + policy, + port, + ...(options.signal === undefined ? {} : { signal: options.signal }), + }); + if (!authorityRuns.ok) return authorityRuns; + const candidateRuns = await captureRepeats({ + scenario: candidateScenario, + repeatCount: experiment.repeat_count, + policy, + port, + ...(options.signal === undefined ? {} : { signal: options.signal }), + }); + if (!candidateRuns.ok) return candidateRuns; + const authorityRepeatability = analyzeProcessRepeatability( + authorityRuns.value, + experiment.required_dimensions, + ); + const candidateRepeatability = analyzeProcessRepeatability( + candidateRuns.value, + experiment.required_dimensions, + ); + const stable = authorityRepeatability.stable && candidateRepeatability.stable; + const allDimensionsRequired = PROCESS_COMPARISON_DIMENSIONS.every( + (dimension) => experiment.required_dimensions.includes(dimension), + ); + const authorityLatest = authorityRuns.value.at(-1); + const candidateLatest = candidateRuns.value.at(-1); + const crossSide = + stable && + allDimensionsRequired && + authorityLatest !== undefined && + candidateLatest !== undefined + ? compareProcessCaptures(authorityLatest, candidateLatest, { + maxCaptureAgeMs: experiment.freshness_policy.max_capture_age_ms, + ...(options.now === undefined ? {} : { now: options.now }), + }) + : null; + return ok({ + authority_runs: authorityRuns.value, + candidate_runs: candidateRuns.value, + authority_repeatability: authorityRepeatability, + candidate_repeatability: candidateRepeatability, + cross_side: crossSide, + cross_side_blocked_reason: + crossSide !== null + ? null + : !allDimensionsRequired + ? "Cross-side comparison is blocked until every returned dimension is required and repeatable." + : "Cross-side comparison is blocked because a required same-side dimension is unstable.", + }); +}; diff --git a/src/domain/processComparison.ts b/src/domain/processComparison.ts index 19e371b4ad..6b9eb011b9 100644 --- a/src/domain/processComparison.ts +++ b/src/domain/processComparison.ts @@ -86,7 +86,9 @@ export const processCaptureComparisonSchema = z path: ["status"], }); }); -type ProcessCaptureComparison = z.infer; +export type ProcessCaptureComparison = z.infer< + typeof processCaptureComparisonSchema +>; const terminalObservations = (capture: ProcessCapture): readonly unknown[] => [ @@ -113,6 +115,29 @@ const filesystemObservations = ( }, ]; +/** Canonical observations for one process-comparison dimension. */ +export const processDimensionObservations = ( + capture: ProcessCapture, + dimension: (typeof PROCESS_COMPARISON_DIMENSIONS)[number], +): readonly unknown[] => { + switch (dimension) { + case "terminal": + return terminalObservations(capture); + case "interaction": + return capture.interaction_events; + case "exit": + return [{ ...capture.exit, settlement: capture.settlement }]; + case "filesystem": + return filesystemObservations(capture); + case "protocol": + return capture.protocol_events; + case "process": + return capture.process_samples; + case "shim": + return capture.shim_events; + } +}; + const classifyCollection = ( left: readonly unknown[], right: readonly unknown[], diff --git a/src/domain/processPairedExperiment.ts b/src/domain/processPairedExperiment.ts new file mode 100644 index 0000000000..a28d236b4d --- /dev/null +++ b/src/domain/processPairedExperiment.ts @@ -0,0 +1,206 @@ +import { z } from "zod"; + +import type { ProcessCapture } from "./processCapture.js"; +import { + PROCESS_COMPARISON_DIMENSIONS, + compareProcessCaptures, + processDimensionObservations, + type ProcessCaptureComparison, +} from "./processComparison.js"; +import { + digestProcessCommitment, + processScenarioSchema, + type ProcessScenario, +} from "./processScenario.js"; + +const sideIdentifierSchema = z + .string() + .regex(/^[A-Za-z][A-Za-z0-9._-]{0,127}$/u); +const environmentNameSchema = z.string().regex(/^[A-Za-z_][A-Za-z0-9_]*$/u); +const processDimensionSchema = z.enum(PROCESS_COMPARISON_DIMENSIONS); + +const sideBindingSchema = z + .object({ + id: sideIdentifierSchema, + executable: z.string().startsWith("/"), + arguments: z.array(z.string()).max(256).optional(), + environment: z.record(environmentNameSchema, z.string()).default({}), + }) + .strict(); + +/** Boundary contract for one bounded authority/candidate process experiment. */ +export const processPairedExperimentSchema = z + .object({ + approved: z.literal(true), + shared_scenario: processScenarioSchema, + authority: sideBindingSchema, + candidate: sideBindingSchema, + repeat_count: z.number().int().min(2).max(10).default(3), + freshness_policy: z + .object({ + max_capture_age_ms: z.number().int().positive().max(86_400_000), + }) + .strict(), + required_dimensions: z + .array(processDimensionSchema) + .min(1) + .max(PROCESS_COMPARISON_DIMENSIONS.length) + .default([...PROCESS_COMPARISON_DIMENSIONS]), + }) + .strict() + .superRefine((experiment, context) => { + if (experiment.authority.id === experiment.candidate.id) + context.addIssue({ + code: "custom", + message: "authority and candidate identities must differ", + path: ["candidate", "id"], + }); + if ( + new Set(experiment.required_dimensions).size !== + experiment.required_dimensions.length + ) + context.addIssue({ + code: "custom", + message: "required process dimensions must be unique", + path: ["required_dimensions"], + }); + }); + +export type ProcessPairedExperiment = z.infer< + typeof processPairedExperimentSchema +>; +/** Executable and environment values applied to one experiment side. */ +export type ProcessSideBinding = ProcessPairedExperiment[ + | "authority" + | "candidate"]; + +type ProcessRepeatabilityClassification = + | "stable" + | "timing_variance" + | "nondeterministic_output" + | "flaky_exit" + | "environmental_drift" + | "insufficient_samples"; + +/** Stability classification for repeated captures of one experiment side. */ +export interface ProcessSideRepeatability { + readonly classification: ProcessRepeatabilityClassification; + readonly stable: boolean; + readonly digest: string; + readonly comparisons: readonly ProcessCaptureComparison[]; + readonly unstable_dimensions: readonly (typeof PROCESS_COMPARISON_DIMENSIONS)[number][]; +} + +/** Apply a side binding without changing the shared observation contract. */ +export const bindProcessScenario = ( + shared: ProcessScenario, + side: ProcessSideBinding, +): ProcessScenario => + processScenarioSchema.parse({ + ...shared, + executable: side.executable, + arguments: side.arguments ?? shared.arguments, + environment: { ...shared.environment, ...side.environment }, + }); + +const classifyInstability = ( + dimensions: ReadonlySet<(typeof PROCESS_COMPARISON_DIMENSIONS)[number]>, +): ProcessRepeatabilityClassification => { + if (dimensions.has("exit")) return "flaky_exit"; + if (dimensions.has("filesystem") || dimensions.has("process")) + return "environmental_drift"; + return "nondeterministic_output"; +}; + +const timingKeys = new Set([ + "at_ms", + "scheduled_at_ms", + "dispatched_at_ms", + "elapsed_ms", +]); + +const withoutTiming = (value: unknown): unknown => + typeof value !== "object" || value === null || Array.isArray(value) + ? value + : Object.fromEntries( + Object.entries(value).filter(([key]) => !timingKeys.has(key)), + ); + +const observationsWithoutTiming = ( + capture: ProcessCapture, + dimension: (typeof PROCESS_COMPARISON_DIMENSIONS)[number], +): readonly unknown[] => + dimension === "exit" + ? [ + { + ...capture.exit, + settlement: { + state: capture.settlement.state, + cleanup_outcome: capture.settlement.cleanup_outcome, + }, + }, + ] + : processDimensionObservations(capture, dimension).map(withoutTiming); + +const hasOnlyTimingVariance = ( + captures: readonly ProcessCapture[], + dimensions: readonly (typeof PROCESS_COMPARISON_DIMENSIONS)[number][], +): boolean => + dimensions.every((dimension) => { + const first = captures[0]; + if (first === undefined) return false; + const baseline = JSON.stringify( + observationsWithoutTiming(first, dimension), + ); + return captures + .slice(1) + .every( + (capture) => + JSON.stringify(observationsWithoutTiming(capture, dimension)) === + baseline, + ); + }); + +/** Compare same-side repeats before permitting a cross-side claim. */ +export const analyzeProcessRepeatability = ( + captures: readonly ProcessCapture[], + requiredDimensions: readonly (typeof PROCESS_COMPARISON_DIMENSIONS)[number][], +): ProcessSideRepeatability => { + if (captures.length < 2) + return { + classification: "insufficient_samples", + stable: false, + digest: digestProcessCommitment({ captures: [], requiredDimensions }), + comparisons: [], + unstable_dimensions: [], + }; + const baseline = captures[0]; + if (baseline === undefined) + throw new TypeError("Repeatability analysis lost its baseline capture"); + const comparisons = captures + .slice(1) + .map((capture) => compareProcessCaptures(baseline, capture)); + const unstable = new Set( + requiredDimensions.filter((dimension) => + comparisons.some((comparison) => comparison[dimension] !== "unchanged"), + ), + ); + const unstableDimensions = PROCESS_COMPARISON_DIMENSIONS.filter((dimension) => + unstable.has(dimension), + ); + return { + classification: + unstable.size === 0 + ? "stable" + : hasOnlyTimingVariance(captures, unstableDimensions) + ? "timing_variance" + : classifyInstability(unstable), + stable: unstable.size === 0, + digest: digestProcessCommitment({ + requiredDimensions, + comparisons, + }), + comparisons, + unstable_dimensions: unstableDimensions, + }; +}; diff --git a/tests/processPairedExperiment.test.ts b/tests/processPairedExperiment.test.ts new file mode 100644 index 0000000000..a36553dbcb --- /dev/null +++ b/tests/processPairedExperiment.test.ts @@ -0,0 +1,247 @@ +import { describe, expect, it } from "vitest"; + +import { + runPairedProcessExperiment, + type ProcessCapturePort, +} from "../src/application/ProcessPairedExperiment.js"; +import { EMPTY_PROCESS_CAPTURE_EXAMPLE } from "../src/contracts/processCaptureExample.js"; +import { parseProcessCapture } from "../src/domain/processCapture.js"; +import { + analyzeProcessRepeatability, + bindProcessScenario, + processPairedExperimentSchema, +} from "../src/domain/processPairedExperiment.js"; +import { ok } from "../src/domain/result.js"; + +const experiment = () => + processPairedExperimentSchema.parse({ + approved: true, + shared_scenario: { + approved: true, + executable: "/bin/template", + arguments: ["shared"], + working_directory: "/tmp", + environment: { SHARED: "yes" }, + }, + authority: { + id: "authority", + executable: "/bin/authority", + environment: { SIDE: "authority" }, + }, + candidate: { + id: "candidate", + executable: "/bin/candidate", + arguments: ["candidate"], + environment: { SIDE: "candidate" }, + }, + repeat_count: 3, + freshness_policy: { max_capture_age_ms: 60_000 }, + required_dimensions: [ + "terminal", + "interaction", + "exit", + "filesystem", + "protocol", + "process", + "shim", + ], + }); + +const policy = { + enabled: true, + executableRoots: ["/bin"], + workingRoots: ["/tmp"], + allowedEnvironment: ["SHARED", "SIDE"], + allowExternalNetwork: false, +}; + +describe("paired process experiments", () => { + it("binds only declared side inputs onto the shared observation contract", () => { + const parsed = experiment(); + const authority = bindProcessScenario( + parsed.shared_scenario, + parsed.authority, + ); + const candidate = bindProcessScenario( + parsed.shared_scenario, + parsed.candidate, + ); + expect(authority).toMatchObject({ + executable: "/bin/authority", + arguments: ["shared"], + environment: { SHARED: "yes", SIDE: "authority" }, + }); + expect(candidate).toMatchObject({ + executable: "/bin/candidate", + arguments: ["candidate"], + environment: { SHARED: "yes", SIDE: "candidate" }, + }); + }); + + it("runs same-side comparisons first and compares stable sides", async () => { + const observedExecutables: string[] = []; + const capture = parseProcessCapture(EMPTY_PROCESS_CAPTURE_EXAMPLE); + const capturePort: ProcessCapturePort = { + capture: (scenario) => { + observedExecutables.push(scenario.executable); + return Promise.resolve(ok(structuredClone(capture))); + }, + }; + const result = await runPairedProcessExperiment(experiment(), policy, { + capturePort, + now: () => Date.parse("2026-01-01T00:00:01.000Z"), + }); + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(observedExecutables).toEqual([ + "/bin/authority", + "/bin/authority", + "/bin/authority", + "/bin/candidate", + "/bin/candidate", + "/bin/candidate", + ]); + expect(result.value.authority_repeatability.stable).toBe(true); + expect(result.value.candidate_repeatability.stable).toBe(true); + expect(result.value.cross_side?.status).toBe("unchanged"); + }); + + it("blocks cross-side claims when a candidate has a flaky exit", async () => { + const base = parseProcessCapture(EMPTY_PROCESS_CAPTURE_EXAMPLE); + let candidateRun = 0; + const capturePort: ProcessCapturePort = { + capture: (scenario) => { + let capture = structuredClone(base); + if (scenario.executable === "/bin/candidate") { + candidateRun += 1; + if (candidateRun === 2) + capture = { + ...capture, + exit: { code: 1, signal: null, reason: "exited" }, + }; + } + return Promise.resolve(ok(capture)); + }, + }; + const result = await runPairedProcessExperiment(experiment(), policy, { + capturePort, + now: () => Date.parse("2026-01-01T00:00:01.000Z"), + }); + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.value.candidate_repeatability).toMatchObject({ + stable: false, + classification: "flaky_exit", + unstable_dimensions: ["exit"], + }); + expect(result.value.cross_side).toBeNull(); + expect(result.value.cross_side_blocked_reason).toMatch(/unstable/u); + }); + + it("distinguishes normalized timing variance from semantic output drift", () => { + const base = parseProcessCapture(EMPTY_PROCESS_CAPTURE_EXAMPLE); + const early = { + ...base, + frames: [{ sequence: 0, at_ms: 10, data: "ready" }], + }; + const late = { + ...base, + frames: [{ sequence: 0, at_ms: 20, data: "ready" }], + }; + expect( + analyzeProcessRepeatability([early, late], ["terminal"]), + ).toMatchObject({ + stable: false, + classification: "timing_variance", + unstable_dimensions: ["terminal"], + }); + const changed = { + ...base, + frames: [{ sequence: 0, at_ms: 20, data: "failed" }], + }; + expect( + analyzeProcessRepeatability([early, changed], ["terminal"]), + ).toMatchObject({ classification: "nondeterministic_output" }); + }); + + it("stops before capture when the experiment is cancelled", async () => { + const controller = new AbortController(); + controller.abort(); + const capturePort: ProcessCapturePort = { + capture: () => { + throw new Error("capture must not start after cancellation"); + }, + }; + + const result = await runPairedProcessExperiment(experiment(), policy, { + capturePort, + signal: controller.signal, + }); + + expect(result).toMatchObject({ + ok: false, + error: { _tag: "ProcessCaptureError", reason: "cancelled" }, + }); + }); + + it("blocks full cross-side output when only a subset was repeatable", async () => { + const capture = parseProcessCapture(EMPTY_PROCESS_CAPTURE_EXAMPLE); + const capturePort: ProcessCapturePort = { + capture: () => Promise.resolve(ok(structuredClone(capture))), + }; + const subset = processPairedExperimentSchema.parse({ + ...experiment(), + required_dimensions: ["exit"], + }); + + const result = await runPairedProcessExperiment(subset, policy, { + capturePort, + }); + + expect(result).toMatchObject({ + ok: true, + value: { + cross_side: null, + cross_side_blocked_reason: expect.stringMatching(/every returned/u), + }, + }); + }); + + it("compares the newest stable captures under the freshness policy", async () => { + const base = parseProcessCapture(EMPTY_PROCESS_CAPTURE_EXAMPLE); + let captureIndex = 0; + const capturePort: ProcessCapturePort = { + capture: () => { + const completedAt = + captureIndex === 0 + ? "2026-01-01T00:00:00.000Z" + : "2026-01-01T00:00:09.500Z"; + captureIndex += 1; + return Promise.resolve( + ok({ + ...structuredClone(base), + manifest: { + ...base.manifest, + started_at: completedAt, + completed_at: completedAt, + }, + }), + ); + }, + }; + const shortFreshness = processPairedExperimentSchema.parse({ + ...experiment(), + freshness_policy: { max_capture_age_ms: 1_000 }, + }); + + const result = await runPairedProcessExperiment(shortFreshness, policy, { + capturePort, + now: () => Date.parse("2026-01-01T00:00:10.000Z"), + }); + + expect(result).toMatchObject({ + ok: true, + value: { cross_side: { status: "unchanged" } }, + }); + }); +});