feat(process): add bounded replay state machines

This commit is contained in:
morluto 2026-07-20 12:17:14 +00:00
parent e9899c003b
commit dc0c2e5ac3
4 changed files with 1519 additions and 0 deletions

376
src/domain/replayMachine.ts Normal file
View File

@ -0,0 +1,376 @@
import { MultiDirectedGraph } from "graphology";
import { singleSource } from "graphology-shortest-path";
import canonicalize from "canonicalize";
import { z } from "zod";
const replayIdentifierSchema = z
.string()
.regex(/^[A-Za-z][A-Za-z0-9_-]{0,63}$/u);
const replayPathSchema = z
.array(z.union([z.string().max(256), z.number().int().nonnegative()]))
.max(32);
const sensitiveHeaderNames = new Set([
"authorization",
"cookie",
"proxy-authorization",
"set-cookie",
]);
const replayHeaderMatchersSchema = z
.record(z.string().min(1).max(256), z.string().max(8_192))
.superRefine((headers, context) => {
const names = Object.keys(headers).map((name) => name.toLowerCase());
if (new Set(names).size !== names.length)
context.addIssue({
code: "custom",
message: "replay header matchers must be unique ignoring case",
input: headers,
});
if (names.some((name) => sensitiveHeaderNames.has(name)))
context.addIssue({
code: "custom",
message:
"replay header matchers cannot persist authorization or cookie values; capture and guard through a secret alias instead",
input: headers,
});
})
.transform((headers) =>
Object.fromEntries(
Object.entries(headers).map(([name, value]) => [
name.toLowerCase(),
value,
]),
),
);
const replayValueSourceSchema = z.discriminatedUnion("source", [
z.strictObject({
source: z.literal("request_header"),
name: z
.string()
.min(1)
.transform((name) => name.toLowerCase()),
}),
z.strictObject({ source: z.literal("request_json"), path: replayPathSchema }),
z.strictObject({
source: z.literal("websocket_json"),
path: replayPathSchema,
}),
z.strictObject({ source: z.literal("action_json"), path: replayPathSchema }),
]);
const replayTriggerSchema = z.discriminatedUnion("protocol", [
z.strictObject({
protocol: z.literal("http"),
method: z
.string()
.min(1)
.max(16)
.transform((method) => method.toUpperCase()),
path: z.string().startsWith("/"),
headers: replayHeaderMatchersSchema.default({}),
body: z.string().max(1_000_000).nullable().default(null),
}),
z.strictObject({
protocol: z.literal("websocket_connect"),
path: z.string().startsWith("/"),
headers: replayHeaderMatchersSchema.default({}),
}),
z.strictObject({
protocol: z.literal("websocket_message"),
path: z.string().startsWith("/"),
body: z.string().max(1_000_000).nullable().default(null),
}),
]);
const replayActionSchema = z.discriminatedUnion("type", [
z.strictObject({
type: z.literal("http_response"),
status: z.number().int().min(100).max(599),
headers: z.record(z.string(), z.string()).default({}),
body: z.string().max(1_000_000),
}),
z.strictObject({
type: z.literal("websocket_send"),
data: z.string().max(1_000_000),
}),
z.strictObject({ type: z.literal("disconnect") }),
z.strictObject({
type: z.literal("delay"),
duration_ms: z.number().int().nonnegative().max(30_000),
}),
]);
const replayTransitionSchema = z.strictObject({
id: replayIdentifierSchema,
from: replayIdentifierSchema,
to: replayIdentifierSchema,
priority: z.number().int().min(0).max(1_000).default(100),
trigger: replayTriggerSchema,
guards: z
.array(
z.strictObject({
variable: replayIdentifierSchema,
value: replayValueSourceSchema,
}),
)
.max(32)
.default([]),
captures: z
.array(
z.strictObject({
variable: replayIdentifierSchema,
value: replayValueSourceSchema,
sensitive: z.boolean().default(false),
}),
)
.max(32)
.default([]),
actions: z.array(replayActionSchema).min(1).max(32),
max_uses: z.number().int().min(1).max(10_000),
});
const transitionSignature = (
transition: z.infer<typeof replayTransitionSchema>,
): string => {
const signature = canonicalize({
from: transition.from,
priority: transition.priority,
trigger: transition.trigger,
});
if (signature === undefined)
throw new TypeError("Replay transition signature is not canonical JSON");
return signature;
};
const replayMachineBaseSchema = z.strictObject({
initial_state: replayIdentifierSchema,
states: z
.array(
z.strictObject({
name: replayIdentifierSchema,
terminal: z.boolean().default(false),
max_visits: z.number().int().min(1).max(100_000).default(100_000),
}),
)
.min(1)
.max(256),
transitions: z.array(replayTransitionSchema).min(1).max(2_000),
max_transitions: z.number().int().min(1).max(100_000),
limits: z
.strictObject({
connections: z.number().int().min(1).max(1_000).default(100),
messages: z.number().int().min(1).max(100_000).default(10_000),
bytes: z.number().int().min(1).max(10_000_000).default(1_000_000),
duration_ms: z.number().int().min(1).max(300_000).default(30_000),
})
.default({
connections: 100,
messages: 10_000,
bytes: 1_000_000,
duration_ms: 30_000,
}),
});
type ReplayMachineInput = z.infer<typeof replayMachineBaseSchema>;
type ReplayMachineContext = z.core.$RefinementCtx<ReplayMachineInput>;
const validateReplayIdentities = (
machine: ReplayMachineInput,
context: ReplayMachineContext,
): ReadonlySet<string> => {
const names = machine.states.map(({ name }) => name);
const nameSet = new Set(names);
if (nameSet.size !== names.length)
context.addIssue({
code: "custom",
message: "replay machine state names must be unique",
path: ["states"],
input: machine,
});
if (!nameSet.has(machine.initial_state))
context.addIssue({
code: "custom",
message: "replay machine initial state is not declared",
path: ["initial_state"],
input: machine,
});
const transitionIds = machine.transitions.map(({ id }) => id);
if (new Set(transitionIds).size !== transitionIds.length)
context.addIssue({
code: "custom",
message: "replay machine transition IDs must be unique",
path: ["transitions"],
input: machine,
});
const signatures = machine.transitions.map(transitionSignature);
if (new Set(signatures).size !== signatures.length)
context.addIssue({
code: "custom",
message:
"replay machine transitions with the same state, trigger, and priority are ambiguous",
path: ["transitions"],
input: machine,
});
return nameSet;
};
const validateReplayVariables = (
machine: ReplayMachineInput,
context: ReplayMachineContext,
): void => {
const capturedVariables = new Map<string, boolean>();
for (const [transitionIndex, transition] of machine.transitions.entries()) {
for (const capture of transition.captures) {
const sensitivity = capturedVariables.get(capture.variable);
if (sensitivity !== undefined && sensitivity !== capture.sensitive)
context.addIssue({
code: "custom",
message:
"a replay variable cannot change secret classification between captures",
path: ["transitions", transitionIndex, "captures"],
input: machine,
});
capturedVariables.set(capture.variable, capture.sensitive);
}
}
const availableByState = new Map<string, Set<string>>([
[machine.initial_state, new Set()],
]);
let changed = true;
while (changed) {
changed = false;
for (const transition of machine.transitions) {
const available = availableByState.get(transition.from);
if (
available === undefined ||
transition.guards.some(({ variable }) => !available.has(variable))
)
continue;
const nextAvailable = new Set(available);
for (const { variable } of transition.captures)
nextAvailable.add(variable);
const existing = availableByState.get(transition.to);
if (existing === undefined) {
availableByState.set(transition.to, nextAvailable);
changed = true;
continue;
}
for (const variable of nextAvailable)
if (!existing.has(variable)) {
existing.add(variable);
changed = true;
}
}
}
for (const [transitionIndex, transition] of machine.transitions.entries())
for (const [guardIndex, guard] of transition.guards.entries())
if (!availableByState.get(transition.from)?.has(guard.variable))
context.addIssue({
code: "custom",
message:
"replay guard variable is not captured on an executable path before use",
path: ["transitions", transitionIndex, "guards", guardIndex],
input: machine,
});
};
const validateReplayActions = (
machine: ReplayMachineInput,
context: ReplayMachineContext,
): void => {
for (const [index, transition] of machine.transitions.entries()) {
const httpResponses = transition.actions.filter(
({ type }) => type === "http_response",
).length;
const websocketSends = transition.actions.some(
({ type }) => type === "websocket_send",
);
const invalid =
transition.trigger.protocol === "http"
? httpResponses !== 1 || websocketSends
: httpResponses !== 0;
if (invalid)
context.addIssue({
code: "custom",
message:
"replay actions must match the transition protocol and HTTP transitions require exactly one response",
path: ["transitions", index, "actions"],
input: machine,
});
}
};
const validateReplayGraph = (
machine: ReplayMachineInput,
names: ReadonlySet<string>,
context: ReplayMachineContext,
): void => {
const graph = new MultiDirectedGraph();
for (const name of names) graph.addNode(name);
for (const [index, transition] of machine.transitions.entries()) {
if (!names.has(transition.from) || !names.has(transition.to)) {
context.addIssue({
code: "custom",
message: "replay machine transition references an unknown state",
path: ["transitions", index],
input: machine,
});
continue;
}
graph.addDirectedEdgeWithKey(
`${transition.id}:${String(index)}`,
transition.from,
transition.to,
);
}
if (!graph.hasNode(machine.initial_state)) return;
if (!machine.states.some(({ terminal }) => terminal))
context.addIssue({
code: "custom",
message: "replay machine must declare at least one terminal state",
path: ["states"],
input: machine,
});
const reachable = new Set(
Object.keys(singleSource(graph, machine.initial_state)),
);
reachable.add(machine.initial_state);
for (const [index, state] of machine.states.entries()) {
if (!reachable.has(state.name))
context.addIssue({
code: "custom",
message: "replay machine state is unreachable from the initial state",
path: ["states", index],
input: machine,
});
if (state.terminal && graph.outDegree(state.name) > 0)
context.addIssue({
code: "custom",
message:
"terminal replay machine state cannot have outgoing transitions",
path: ["states", index],
input: machine,
});
if (!state.terminal && graph.outDegree(state.name) === 0)
context.addIssue({
code: "custom",
message: "non-terminal replay machine state has no outgoing transition",
path: ["states", index],
input: machine,
});
}
};
/** Portable finite replay machine with no executable callbacks. */
export const replayMachineSchema = z
.object(replayMachineBaseSchema.shape)
.strict()
.superRefine((machine, context) => {
const names = validateReplayIdentities(machine, context);
validateReplayVariables(machine, context);
validateReplayActions(machine, context);
validateReplayGraph(machine, names, context);
});
export type ReplayMachine = z.infer<typeof replayMachineSchema>;

View File

@ -0,0 +1,318 @@
import canonicalize from "canonicalize";
import type { ReplayMachine } from "./replayMachine.js";
type ReplayTransition = ReplayMachine["transitions"][number];
type ReplayValueSource = ReplayTransition["captures"][number]["value"];
/** One bounded inbound protocol event offered to a replay machine. */
export interface ReplayMachineEvent {
readonly protocol: "http" | "websocket_connect" | "websocket_message";
readonly connection: "not_applicable" | "initial" | "reconnect";
readonly at_ms: number;
readonly method: string | null;
readonly path: string;
readonly headers: Readonly<Record<string, string>>;
readonly body: string;
}
/** Persistable transition journal entry containing aliases, never secrets. */
export interface ReplayTransitionRecord {
readonly sequence: number;
readonly transition_id: string;
readonly state_before: string;
readonly state_after: string;
readonly sensitive_aliases: readonly string[];
}
/** Result of dispatching one event through a replay machine. */
export type ReplayMachineDecision =
| {
readonly outcome: "matched";
readonly actions: ReplayTransition["actions"];
readonly transition: ReplayTransitionRecord;
}
| {
readonly outcome:
| "unmatched"
| "invalid_state"
| "guard_failed"
| "transition_exhausted"
| "invalid_capture"
| "unexpected_reconnect"
| "limit_exhausted";
readonly actions: readonly [];
readonly transition: null;
};
const parseJson = (value: string): unknown => {
try {
return JSON.parse(value) as unknown;
} catch {
return undefined;
}
};
const atPath = (
value: unknown,
path: readonly (string | number)[],
): unknown => {
let current = value;
for (const segment of path) {
if (typeof segment === "number") {
if (!Array.isArray(current)) return undefined;
current = current[segment];
continue;
}
if (
typeof current !== "object" ||
current === null ||
!Object.hasOwn(current, segment)
)
return undefined;
current = (current as Readonly<Record<string, unknown>>)[segment];
}
return current;
};
const actionJsonBody = (transition: ReplayTransition): string | undefined => {
for (const action of transition.actions)
if (action.type === "http_response") return action.body;
else if (action.type === "websocket_send") return action.data;
return undefined;
};
const readValue = (
source: ReplayValueSource,
event: ReplayMachineEvent,
transition: ReplayTransition,
): unknown => {
switch (source.source) {
case "request_header":
return event.headers[source.name.toLowerCase()];
case "request_json":
return atPath(parseJson(event.body), source.path);
case "websocket_json":
return atPath(parseJson(event.body), source.path);
case "action_json": {
const body = actionJsonBody(transition);
return body === undefined
? undefined
: atPath(parseJson(body), source.path);
}
}
};
const triggerMatches = (
transition: ReplayTransition,
event: ReplayMachineEvent,
): boolean => {
const trigger = transition.trigger;
if (trigger.protocol !== event.protocol || trigger.path !== event.path)
return false;
if (
"headers" in trigger &&
Object.entries(trigger.headers).some(
([name, value]) => event.headers[name.toLowerCase()] !== value,
)
)
return false;
if (trigger.protocol === "http")
return (
trigger.method === event.method &&
(trigger.body === null || trigger.body === event.body)
);
if (trigger.protocol === "websocket_message")
return trigger.body === null || trigger.body === event.body;
return true;
};
const isJsonCompatible = (value: unknown): boolean => {
if (value === null || typeof value === "string" || typeof value === "boolean")
return true;
if (typeof value === "number") return Number.isFinite(value);
if (Array.isArray(value)) return value.every(isJsonCompatible);
if (typeof value !== "object") return false;
return Object.values(value).every(isJsonCompatible);
};
const replayValuesEqual = (left: unknown, right: unknown): boolean =>
isJsonCompatible(left) &&
isJsonCompatible(right) &&
(Object.is(left, right) || canonicalize(left) === canonicalize(right));
/** Stateful evaluator for one validated replay-machine instance. */
export class ReplayMachineRuntime {
readonly #variables = new Map<string, unknown>();
readonly #sensitiveValues = new Map<string, Set<string>>();
readonly #uses = new Map<string, number>();
readonly #stateVisits = new Map<string, number>();
readonly #timeline: ReplayTransitionRecord[] = [];
#state: string;
#connections = 0;
#messages = 0;
#bytes = 0;
#lastAtMs = 0;
constructor(readonly machine: ReplayMachine) {
this.#state = machine.initial_state;
this.#stateVisits.set(machine.initial_state, 1);
}
/** Current machine state after all accepted transitions. */
get state(): string {
return this.#state;
}
/** Ordered, secret-free transition records. */
get timeline(): readonly ReplayTransitionRecord[] {
return this.#timeline;
}
/** Replace captured secret values with stable aliases before persistence. */
redact(value: string): string {
let redacted = value;
const secrets = [...this.#sensitiveValues]
.flatMap(([variable, values]) =>
[...values].map((secret) => ({ variable, secret })),
)
.sort(
(left, right) =>
right.secret.length - left.secret.length ||
left.variable.localeCompare(right.variable),
);
for (const { variable, secret } of secrets)
redacted = redacted.replaceAll(secret, `<secret:${variable}>`);
return redacted;
}
/** Evaluate and, when admitted, commit one protocol event. */
dispatch(event: ReplayMachineEvent): ReplayMachineDecision {
const normalizedEvent = {
...event,
headers: Object.fromEntries(
Object.entries(event.headers).map(([name, value]) => [
name.toLowerCase(),
value,
]),
),
};
if (!this.#admitEvent(normalizedEvent))
return this.#refusal("limit_exhausted");
const matching = this.machine.transitions
.map((transition, declarationOrder) => ({
transition,
declarationOrder,
}))
.filter(({ transition }) => triggerMatches(transition, normalizedEvent));
const eligible = matching
.filter(({ transition }) => transition.from === this.#state)
.sort(
(left, right) =>
left.transition.priority - right.transition.priority ||
left.declarationOrder - right.declarationOrder,
);
if (eligible.length === 0)
return this.#refusal(
matching.length === 0
? "unmatched"
: normalizedEvent.protocol === "websocket_connect" &&
normalizedEvent.connection === "reconnect"
? "unexpected_reconnect"
: "invalid_state",
);
const guarded = eligible.filter(({ transition }) =>
transition.guards.every(
(guard) =>
this.#variables.has(guard.variable) &&
replayValuesEqual(
this.#variables.get(guard.variable),
readValue(guard.value, normalizedEvent, transition),
),
),
);
if (guarded.length === 0) return this.#refusal("guard_failed");
if (this.#timeline.length >= this.machine.max_transitions)
return this.#refusal("transition_exhausted");
const usable = guarded.find(
({ transition }) =>
(this.#uses.get(transition.id) ?? 0) < transition.max_uses,
);
if (usable === undefined) return this.#refusal("transition_exhausted");
const transition = usable.transition;
const used = this.#uses.get(transition.id) ?? 0;
const nextState = this.machine.states.find(
({ name }) => name === transition.to,
);
const nextVisits = (this.#stateVisits.get(transition.to) ?? 0) + 1;
if (nextState === undefined || nextVisits > nextState.max_visits)
return this.#refusal("limit_exhausted");
const captures = transition.captures.map((capture) => ({
...capture,
captured: readValue(capture.value, normalizedEvent, transition),
}));
if (
captures.some(
({ captured, sensitive }) =>
!isJsonCompatible(captured) ||
(sensitive && typeof captured !== "string"),
)
)
return this.#refusal("invalid_capture");
for (const capture of captures) {
this.#variables.set(capture.variable, capture.captured);
if (capture.sensitive && typeof capture.captured === "string") {
const values = this.#sensitiveValues.get(capture.variable) ?? new Set();
if (capture.captured.length > 0) values.add(capture.captured);
this.#sensitiveValues.set(capture.variable, values);
}
}
const record = {
sequence: this.#timeline.length,
transition_id: transition.id,
state_before: this.#state,
state_after: transition.to,
sensitive_aliases: captures
.filter(({ sensitive }) => sensitive)
.map(({ variable }) => variable)
.sort(),
} satisfies ReplayTransitionRecord;
this.#uses.set(transition.id, used + 1);
this.#stateVisits.set(transition.to, nextVisits);
this.#state = transition.to;
this.#timeline.push(record);
return {
outcome: "matched",
actions: transition.actions,
transition: record,
};
}
#admitEvent(event: ReplayMachineEvent): boolean {
const connections =
this.#connections + (event.protocol === "websocket_connect" ? 1 : 0);
const messages =
this.#messages + (event.protocol === "websocket_message" ? 1 : 0);
const bytes = this.#bytes + Buffer.byteLength(event.body);
if (
!Number.isSafeInteger(event.at_ms) ||
event.at_ms < this.#lastAtMs ||
event.at_ms > this.machine.limits.duration_ms ||
connections > this.machine.limits.connections ||
messages > this.machine.limits.messages ||
bytes > this.machine.limits.bytes
)
return false;
this.#connections = connections;
this.#messages = messages;
this.#bytes = bytes;
this.#lastAtMs = event.at_ms;
return true;
}
#refusal(
outcome: Exclude<ReplayMachineDecision["outcome"], "matched">,
): ReplayMachineDecision {
return { outcome, actions: [], transition: null };
}
}

244
tests/replayMachine.test.ts Normal file
View File

@ -0,0 +1,244 @@
import { describe, expect, it } from "vitest";
import { replayMachineSchema } from "../src/domain/replayMachine.js";
const loginMachine = () => ({
initial_state: "waiting_for_login",
states: [
{ name: "waiting_for_login" },
{ name: "waiting_for_api" },
{ name: "complete", terminal: true },
],
transitions: [
{
id: "login",
from: "waiting_for_login",
to: "waiting_for_api",
priority: 10,
trigger: { protocol: "http", method: "POST", path: "/login" },
captures: [
{
variable: "token",
value: { source: "request_json", path: ["token"] },
sensitive: true,
},
],
actions: [{ type: "http_response", status: 204, headers: {}, body: "" }],
max_uses: 1,
},
{
id: "api",
from: "waiting_for_api",
to: "complete",
priority: 10,
trigger: { protocol: "http", method: "GET", path: "/api" },
guards: [
{
variable: "token",
value: { source: "request_header", name: "authorization" },
},
],
actions: [
{
type: "http_response",
status: 200,
headers: {},
body: "done",
},
],
max_uses: 1,
},
],
max_transitions: 2,
});
describe("replay machine", () => {
it("accepts a bounded, reachable, data-only login machine", () => {
expect(replayMachineSchema.parse(loginMachine())).toMatchObject({
initial_state: "waiting_for_login",
max_transitions: 2,
});
});
it.each([
[
"unreachable state",
(machine: ReturnType<typeof loginMachine>) =>
machine.states.push({ name: "orphan", terminal: false }),
"unreachable",
],
[
"ambiguous transition",
(machine: ReturnType<typeof loginMachine>) => {
const transition = machine.transitions[0];
if (transition === undefined) throw new TypeError("missing transition");
machine.transitions.push({ ...transition, id: "login_2" });
},
"ambiguous",
],
[
"outgoing terminal transition",
(machine: ReturnType<typeof loginMachine>) => {
const transition = machine.transitions[0];
if (transition === undefined) throw new TypeError("missing transition");
machine.transitions.push({
...transition,
id: "restart",
from: "complete",
to: "waiting_for_login",
trigger: { protocol: "http", method: "POST", path: "/restart" },
});
},
"terminal",
],
[
"unknown guard variable",
(machine: ReturnType<typeof loginMachine>) => {
const guard = machine.transitions[1]?.guards?.[0];
if (guard === undefined) throw new TypeError("missing guard");
guard.variable = "missing";
},
"executable path before use",
],
[
"missing terminal state",
(machine: ReturnType<typeof loginMachine>) => {
const state = machine.states[2];
if (state === undefined) throw new TypeError("missing state");
state.terminal = false;
},
"terminal state",
],
])("rejects %s", (_name, mutate, message) => {
const machine = loginMachine();
mutate(machine);
expect(() => replayMachineSchema.parse(machine)).toThrow(message);
});
it("canonicalizes methods and headers while refusing persisted credential matchers", () => {
const canonical = {
...loginMachine(),
transitions: loginMachine().transitions.map((transition, index) =>
index === 0
? {
...transition,
trigger: {
protocol: "http",
method: "post",
path: "/login",
headers: { "X-Request-ID": "fixture" },
},
}
: transition,
),
};
expect(
replayMachineSchema.parse(canonical).transitions[0]?.trigger,
).toMatchObject({
method: "POST",
headers: { "x-request-id": "fixture" },
});
const credentialMatcher = {
...loginMachine(),
transitions: loginMachine().transitions.map((transition, index) =>
index === 0
? {
...transition,
trigger: {
protocol: "http",
method: "POST",
path: "/login",
headers: { Authorization: "test-auth-token" },
},
}
: transition,
),
};
expect(() => replayMachineSchema.parse(credentialMatcher)).toThrow(
/secret alias/u,
);
});
it("rejects unknown fields at nested machine boundaries", () => {
const candidate = loginMachine();
const action = candidate.transitions[0]?.actions[0];
if (action === undefined) throw new TypeError("missing action");
expect(() =>
replayMachineSchema.parse({
...candidate,
transitions: [
{
...candidate.transitions[0],
actions: [{ ...action, unexpected: true }],
},
candidate.transitions[1],
],
}),
).toThrow(/Unrecognized key/u);
});
it("reports duplicate transition IDs as structured validation", () => {
const candidate = loginMachine();
const first = candidate.transitions[0];
const second = candidate.transitions[1];
if (first === undefined || second === undefined)
throw new TypeError("missing transition");
expect(
replayMachineSchema.safeParse({
...candidate,
transitions: [first, { ...second, id: first.id }],
}),
).toMatchObject({ success: false });
});
it("rejects guards whose capture is not available before the transition", () => {
const candidate = loginMachine();
const first = candidate.transitions[0];
if (first === undefined) throw new TypeError("missing transition");
first.guards = [
{
variable: "token",
value: { source: "request_header", name: "authorization" },
},
];
expect(() => replayMachineSchema.parse(candidate)).toThrow(
/executable path before use/u,
);
});
it("rejects ambiguous header triggers regardless of key order", () => {
const candidate = loginMachine();
const first = candidate.transitions[0];
if (first === undefined) throw new TypeError("missing transition");
const withHeaders = {
...first,
trigger: {
protocol: "http" as const,
method: "POST",
path: "/login",
headers: { accept: "json", "x-mode": "test" },
},
};
expect(() =>
replayMachineSchema.parse({
...candidate,
transitions: [
withHeaders,
{
...withHeaders,
id: "login_duplicate",
trigger: {
...withHeaders.trigger,
headers: { "x-mode": "test", accept: "json" },
},
},
candidate.transitions[1],
],
}),
).toThrow(/ambiguous/u);
});
});

View File

@ -0,0 +1,581 @@
import { describe, expect, it } from "vitest";
import { replayMachineSchema } from "../src/domain/replayMachine.js";
import { ReplayMachineRuntime } from "../src/domain/replayMachineRuntime.js";
const machine = replayMachineSchema.parse({
initial_state: "login",
states: [
{ name: "login" },
{ name: "api" },
{ name: "complete", terminal: true },
],
transitions: [
{
id: "capture_token",
from: "login",
to: "api",
priority: 10,
trigger: { protocol: "http", method: "POST", path: "/login" },
captures: [
{
variable: "session_value",
value: { source: "request_json", path: ["session_value"] },
sensitive: true,
},
],
actions: [{ type: "http_response", status: 204, headers: {}, body: "" }],
max_uses: 1,
},
{
id: "authorize_api",
from: "api",
to: "complete",
priority: 10,
trigger: { protocol: "http", method: "GET", path: "/api" },
guards: [
{
variable: "session_value",
value: { source: "request_header", name: "authorization" },
},
],
actions: [
{ type: "http_response", status: 200, headers: {}, body: "done" },
],
max_uses: 1,
},
],
max_transitions: 2,
});
const event = (
method: string,
path: string,
body = "",
headers: Readonly<Record<string, string>> = {},
) => ({
protocol: "http" as const,
connection: "not_applicable" as const,
at_ms: 0,
method,
path,
body,
headers,
});
describe("replay machine runtime", () => {
it("captures secret values in memory and journals aliases only", () => {
const runtime = new ReplayMachineRuntime(machine);
expect(
runtime.dispatch(
event("POST", "/login", '{"session_value":"test-auth-token"}'),
),
).toMatchObject({
outcome: "matched",
transition: {
transition_id: "capture_token",
sensitive_aliases: ["session_value"],
},
});
const expectedRedaction = ["Bearer <", "secret", ":session_value>"].join(
"",
);
expect(runtime.redact("Bearer test-auth-token")).toBe(expectedRedaction);
expect(
runtime.dispatch(
event("GET", "/api", "", { Authorization: "test-auth-token" }),
),
).toMatchObject({ outcome: "matched" });
expect(JSON.stringify(runtime.timeline)).not.toContain("test-auth-token");
expect(runtime.state).toBe("complete");
});
it("classifies wrong-state, guard, capture, and exhaustion failures", () => {
expect(
new ReplayMachineRuntime(machine).dispatch(event("GET", "/api")),
).toMatchObject({ outcome: "invalid_state" });
expect(
new ReplayMachineRuntime(machine).dispatch(event("POST", "/login", "{}")),
).toMatchObject({ outcome: "invalid_capture" });
const runtime = new ReplayMachineRuntime(machine);
runtime.dispatch(
event("POST", "/login", '{"session_value":"test-auth-token"}'),
);
expect(
runtime.dispatch(
event("GET", "/api", "", {
authorization: "test-token-placeholder",
}),
),
).toMatchObject({
outcome: "guard_failed",
});
const bounded = new ReplayMachineRuntime(
replayMachineSchema.parse({
initial_state: "active",
states: [{ name: "active" }, { name: "done", terminal: true }],
transitions: [
{
id: "once",
from: "active",
to: "active",
trigger: { protocol: "http", method: "GET", path: "/once" },
actions: [
{ type: "http_response", status: 200, headers: {}, body: "ok" },
],
max_uses: 1,
},
{
id: "finish",
from: "active",
to: "done",
trigger: { protocol: "http", method: "GET", path: "/done" },
actions: [
{ type: "http_response", status: 200, headers: {}, body: "ok" },
],
max_uses: 1,
},
],
max_transitions: 2,
}),
);
expect(bounded.dispatch(event("GET", "/once"))).toMatchObject({
outcome: "matched",
});
expect(bounded.dispatch(event("GET", "/once"))).toMatchObject({
outcome: "transition_exhausted",
});
});
it("matches bounded headers and bodies and distinguishes wrong-state reconnects", () => {
const runtime = new ReplayMachineRuntime(
replayMachineSchema.parse({
initial_state: "connected",
states: [{ name: "connected" }, { name: "done", terminal: true }],
transitions: [
{
id: "finish",
from: "connected",
to: "done",
trigger: {
protocol: "http",
method: "POST",
path: "/finish",
headers: { "content-type": "application/json" },
body: '{"done":true}',
},
actions: [
{ type: "http_response", status: 204, headers: {}, body: "" },
],
max_uses: 1,
},
{
id: "initial_socket",
from: "connected",
to: "connected",
trigger: { protocol: "websocket_connect", path: "/ws" },
actions: [{ type: "disconnect" }],
max_uses: 1,
},
],
max_transitions: 2,
}),
);
expect(runtime.dispatch(event("POST", "/finish", "wrong"))).toMatchObject({
outcome: "unmatched",
});
expect(
runtime.dispatch({
protocol: "websocket_connect",
connection: "initial",
at_ms: 0,
method: null,
path: "/ws",
headers: {},
body: "",
}),
).toMatchObject({ outcome: "matched" });
expect(
runtime.dispatch({
protocol: "websocket_connect",
connection: "reconnect",
at_ms: 0,
method: null,
path: "/ws",
headers: {},
body: "",
}),
).toMatchObject({ outcome: "transition_exhausted" });
expect(
runtime.dispatch(
event("POST", "/finish", '{"done":true}', {
"content-type": "application/json",
}),
),
).toMatchObject({ outcome: "matched" });
expect(
runtime.dispatch({
protocol: "websocket_connect",
connection: "reconnect",
at_ms: 0,
method: null,
path: "/ws",
headers: {},
body: "",
}),
).toMatchObject({ outcome: "unexpected_reconnect" });
});
it("enforces state, connection, message, byte, and duration limits", () => {
const limitedMachine = replayMachineSchema.parse({
initial_state: "active",
states: [
{ name: "active", max_visits: 1 },
{ name: "done", terminal: true },
],
transitions: [
{
id: "loop",
from: "active",
to: "active",
trigger: { protocol: "websocket_message", path: "/ws" },
actions: [{ type: "websocket_send", data: "ack" }],
max_uses: 2,
},
{
id: "finish",
from: "active",
to: "done",
trigger: { protocol: "http", method: "GET", path: "/done" },
actions: [
{ type: "http_response", status: 200, headers: {}, body: "ok" },
],
max_uses: 1,
},
],
max_transitions: 3,
limits: { connections: 1, messages: 1, bytes: 100, duration_ms: 10 },
});
const stateLimited = new ReplayMachineRuntime(limitedMachine);
expect(
stateLimited.dispatch({
protocol: "websocket_message",
connection: "not_applicable",
at_ms: 0,
method: null,
path: "/ws",
headers: {},
body: "one",
}),
).toMatchObject({ outcome: "limit_exhausted" });
const messageLimited = new ReplayMachineRuntime(
replayMachineSchema.parse({
...limitedMachine,
states: limitedMachine.states.map((state) =>
state.name === "active" ? { ...state, max_visits: 3 } : state,
),
}),
);
const message = {
protocol: "websocket_message" as const,
connection: "not_applicable" as const,
at_ms: 0,
method: null,
path: "/ws",
headers: {},
body: "one",
};
expect(messageLimited.dispatch(message)).toMatchObject({
outcome: "matched",
});
expect(messageLimited.dispatch(message)).toMatchObject({
outcome: "limit_exhausted",
});
const durationLimited = new ReplayMachineRuntime(limitedMachine);
expect(
durationLimited.dispatch({ ...event("GET", "/done"), at_ms: 11 }),
).toMatchObject({ outcome: "limit_exhausted" });
expect(
new ReplayMachineRuntime(limitedMachine).dispatch({
...event("GET", "/done"),
at_ms: Number.NaN,
}),
).toMatchObject({ outcome: "limit_exhausted" });
const monotonic = new ReplayMachineRuntime(limitedMachine);
expect(
monotonic.dispatch({ ...event("GET", "/missing"), at_ms: 5 }),
).toMatchObject({ outcome: "unmatched" });
expect(
monotonic.dispatch({ ...event("GET", "/missing"), at_ms: 4 }),
).toMatchObject({ outcome: "limit_exhausted" });
const byteLimited = new ReplayMachineRuntime(
replayMachineSchema.parse({
...limitedMachine,
limits: { ...limitedMachine.limits, bytes: 4 },
}),
);
expect(
byteLimited.dispatch(event("GET", "/missing", "12345")),
).toMatchObject({ outcome: "limit_exhausted" });
const connectionLimited = new ReplayMachineRuntime(limitedMachine);
const connect = {
protocol: "websocket_connect" as const,
connection: "initial" as const,
at_ms: 0,
method: null,
path: "/unknown",
headers: {},
body: "",
};
expect(connectionLimited.dispatch(connect)).toMatchObject({
outcome: "unmatched",
});
expect(
connectionLimited.dispatch({ ...connect, connection: "reconnect" }),
).toMatchObject({ outcome: "limit_exhausted" });
});
it("compares captured JSON values canonically", () => {
const objectMachine = replayMachineSchema.parse({
initial_state: "capture",
states: [
{ name: "capture" },
{ name: "guard" },
{ name: "done", terminal: true },
],
transitions: [
{
id: "capture_profile",
from: "capture",
to: "guard",
trigger: { protocol: "http", method: "POST", path: "/capture" },
captures: [
{
variable: "profile",
value: { source: "request_json", path: ["profile"] },
},
],
actions: [
{ type: "http_response", status: 204, headers: {}, body: "" },
],
max_uses: 1,
},
{
id: "guard_profile",
from: "guard",
to: "done",
trigger: { protocol: "http", method: "POST", path: "/guard" },
guards: [
{
variable: "profile",
value: { source: "request_json", path: ["profile"] },
},
],
actions: [
{ type: "http_response", status: 204, headers: {}, body: "" },
],
max_uses: 1,
},
],
max_transitions: 2,
});
const runtime = new ReplayMachineRuntime(objectMachine);
expect(
runtime.dispatch(
event("POST", "/capture", '{"profile":{"name":"A","id":1}}'),
),
).toMatchObject({ outcome: "matched" });
expect(
runtime.dispatch(
event("POST", "/guard", '{"profile":{"id":1,"name":"A"}}'),
),
).toMatchObject({ outcome: "matched" });
});
it("does not traverse inherited properties in JSON capture paths", () => {
const inheritedPathMachine = replayMachineSchema.parse({
...machine,
transitions: machine.transitions.map((transition, index) =>
index === 0
? {
...transition,
captures: [
{
variable: "session_value",
value: {
source: "request_json" as const,
path: ["__proto__", "toString"],
},
sensitive: true,
},
],
}
: transition,
),
});
expect(
new ReplayMachineRuntime(inheritedPathMachine).dispatch(
event("POST", "/login", "{}"),
),
).toMatchObject({ outcome: "invalid_capture" });
});
it("rejects non-string sensitive captures and retains rotated values", () => {
expect(
new ReplayMachineRuntime(machine).dispatch(
event("POST", "/login", '{"session_value":1234}'),
),
).toMatchObject({ outcome: "invalid_capture" });
const rotatingMachine = replayMachineSchema.parse({
initial_state: "active",
states: [
{ name: "active", max_visits: 3 },
{ name: "done", terminal: true },
],
transitions: [
{
id: "rotate",
from: "active",
to: "active",
trigger: { protocol: "http", method: "POST", path: "/rotate" },
captures: [
{
variable: "session_value",
value: { source: "request_json", path: ["value"] },
sensitive: true,
},
],
actions: [
{ type: "http_response", status: 204, headers: {}, body: "" },
],
max_uses: 2,
},
{
id: "finish",
from: "active",
to: "done",
trigger: { protocol: "http", method: "GET", path: "/done" },
actions: [
{ type: "http_response", status: 204, headers: {}, body: "" },
],
max_uses: 1,
},
],
max_transitions: 3,
});
const rotating = new ReplayMachineRuntime(rotatingMachine);
rotating.dispatch(event("POST", "/rotate", '{"value":"alpha"}'));
rotating.dispatch(event("POST", "/rotate", '{"value":"beta"}'));
const alias = ["<", "secret", ":session_value>"].join("");
expect(rotating.redact("alpha beta")).toBe(`${alias} ${alias}`);
});
it("falls back to a lower-priority transition after a preferred one is exhausted", () => {
const fallbackMachine = replayMachineSchema.parse({
initial_state: "active",
states: [
{ name: "active", max_visits: 2 },
{ name: "done", terminal: true },
],
transitions: [
{
id: "preferred",
from: "active",
to: "active",
priority: 10,
trigger: { protocol: "http", method: "GET", path: "/next" },
actions: [
{ type: "http_response", status: 200, headers: {}, body: "again" },
],
max_uses: 1,
},
{
id: "fallback",
from: "active",
to: "done",
priority: 20,
trigger: { protocol: "http", method: "GET", path: "/next" },
actions: [
{ type: "http_response", status: 200, headers: {}, body: "done" },
],
max_uses: 1,
},
],
max_transitions: 2,
});
const runtime = new ReplayMachineRuntime(fallbackMachine);
expect(runtime.dispatch(event("GET", "/next"))).toMatchObject({
outcome: "matched",
transition: { transition_id: "preferred" },
});
expect(runtime.dispatch(event("GET", "/next"))).toMatchObject({
outcome: "matched",
transition: { transition_id: "fallback" },
});
});
it("rejects non-finite parsed JSON captures and guard values", () => {
const jsonMachine = replayMachineSchema.parse({
initial_state: "capture",
states: [
{ name: "capture" },
{ name: "guard" },
{ name: "done", terminal: true },
],
transitions: [
{
id: "capture_value",
from: "capture",
to: "guard",
trigger: { protocol: "http", method: "POST", path: "/capture" },
captures: [
{
variable: "value",
value: { source: "request_json", path: ["value"] },
},
],
actions: [
{ type: "http_response", status: 204, headers: {}, body: "" },
],
max_uses: 1,
},
{
id: "guard_value",
from: "guard",
to: "done",
trigger: { protocol: "http", method: "POST", path: "/guard" },
guards: [
{
variable: "value",
value: { source: "request_json", path: ["value"] },
},
],
actions: [
{ type: "http_response", status: 204, headers: {}, body: "" },
],
max_uses: 1,
},
],
max_transitions: 2,
});
expect(
new ReplayMachineRuntime(jsonMachine).dispatch(
event("POST", "/capture", '{"value":1e400}'),
),
).toMatchObject({ outcome: "invalid_capture" });
const guarded = new ReplayMachineRuntime(jsonMachine);
expect(
guarded.dispatch(event("POST", "/capture", '{"value":null}')),
).toMatchObject({ outcome: "matched" });
expect(
guarded.dispatch(event("POST", "/guard", '{"value":1e400}')),
).toMatchObject({ outcome: "guard_failed" });
});
});