diff --git a/src/domain/processObservation.ts b/src/domain/processObservation.ts new file mode 100644 index 000000000..d823f2136 --- /dev/null +++ b/src/domain/processObservation.ts @@ -0,0 +1,211 @@ +import { z } from "zod"; + +import { jsonValueSchema, type JsonValue } from "./jsonValue.js"; +import type { ProcessCapture } from "./processCapture.js"; + +/** Process-capture observation families shared by live drivers and trace assertions. */ +export const processObservationSourceSchema = z.enum([ + "terminal_raw", + "terminal_rendered", + "interaction", + "lifecycle", + "process", + "filesystem", + "http", + "websocket", + "shim", + "replay_transition", +]); +export type ProcessObservationSource = z.infer< + typeof processObservationSourceSchema +>; + +/** Stable raw-record location within one Process Capture observation journal. */ +export const processObservationLocationSchema = z.strictObject({ + collection: z.enum([ + "frames", + "rendered_frames", + "interaction_events", + "lifecycle", + "process_samples", + "filesystem_checkpoints", + "shim_events", + "protocol_events", + "replay_transitions", + ]), + index: z.number().int().nonnegative(), + capture_order: z.number().int().nonnegative(), +}); +export type ProcessObservationLocation = z.infer< + typeof processObservationLocationSchema +>; + +/** Provider-neutral event offered to a reactive scenario or offline assertion. */ +export interface ProcessObservation { + readonly event_id: string; + readonly source: ProcessObservationSource; + readonly source_sequence: number; + readonly capture_order: number; + readonly captured_at_ms: number | null; + readonly subject_id: string | null; + readonly location: ProcessObservationLocation; + readonly payload: JsonValue; +} + +/** Input used by live producers to construct the same observation shape. */ +export interface ProcessObservationInput { + readonly source: ProcessObservationSource; + readonly source_sequence: number; + readonly captured_at_ms: number | null; + readonly subject_id: string | null; + readonly location: ProcessObservationLocation; + readonly payload: unknown; +} + +/** Remove only explicitly declared top-level volatile fields before matching. */ +export const comparableProcessObservationPayload = ( + value: unknown, + ignoredFields: readonly string[] = [], +): unknown => { + if ( + ignoredFields.length === 0 || + typeof value !== "object" || + value === null || + Array.isArray(value) + ) + return value; + return Object.fromEntries( + Object.entries(value).filter(([name]) => !ignoredFields.includes(name)), + ); +}; + +/** Construct a validated observation from one live producer record. */ +export const createProcessObservation = ( + input: ProcessObservationInput, +): ProcessObservation => ({ + event_id: `obs.${input.location.collection}.${String(input.location.index)}`, + source: input.source, + source_sequence: input.source_sequence, + capture_order: input.location.capture_order, + captured_at_ms: input.captured_at_ms, + subject_id: input.subject_id, + location: input.location, + payload: jsonValueSchema.parse(input.payload), +}); + +type ProjectedRecord = Omit< + ProcessObservation, + "event_id" | "source_sequence" | "capture_order" | "location" +>; + +const atMilliseconds = (value: unknown): number | null => { + if (typeof value !== "object" || value === null) return null; + for (const name of ["at_ms", "dispatched_at_ms"]) + if (name in value) { + const candidate = value[name as keyof typeof value]; + if (typeof candidate === "number" && Number.isSafeInteger(candidate)) + return candidate; + } + return null; +}; + +const subjectFor = ( + source: ProcessObservationSource, + value: JsonValue, +): string | null => { + if (typeof value !== "object" || value === null || Array.isArray(value)) + return null; + if (source === "process" && typeof value["pid"] === "number") + return `process:${String(value["pid"])}`; + if (source === "filesystem" && typeof value["name"] === "string") + return `checkpoint:${value["name"]}`; + return null; +}; + +const projectedRecord = ( + capture: ProcessCapture, + location: ProcessObservationLocation, +): ProjectedRecord | null => { + const projected = (() => { + switch (location.collection) { + case "frames": + return { + source: "terminal_raw" as const, + value: capture.frames[location.index], + }; + case "rendered_frames": + return { + source: "terminal_rendered" as const, + value: capture.rendered_frames[location.index], + }; + case "interaction_events": + return { + source: "interaction" as const, + value: capture.interaction_events[location.index], + }; + case "lifecycle": + return location.index === 0 + ? { + source: "lifecycle" as const, + value: { event: "exit", ...capture.exit }, + } + : location.index === 1 + ? { + source: "lifecycle" as const, + value: { event: "settlement", ...capture.settlement }, + } + : null; + case "process_samples": + return { + source: "process" as const, + value: capture.process_samples[location.index], + }; + case "filesystem_checkpoints": + return { + source: "filesystem" as const, + value: capture.filesystem_checkpoints[location.index], + }; + case "shim_events": + return { + source: "shim" as const, + value: capture.shim_events[location.index], + }; + case "protocol_events": { + const event = capture.protocol_events[location.index]; + return event === undefined + ? null + : { source: event.protocol, value: event }; + } + case "replay_transitions": + return { + source: "replay_transition" as const, + value: capture.replay_transitions[location.index], + }; + } + })(); + if (projected === null || projected.value === undefined) return null; + const payload = jsonValueSchema.parse(projected.value); + return { + source: projected.source, + captured_at_ms: atMilliseconds(payload), + subject_id: subjectFor(projected.source, payload), + payload, + }; +}; + +/** Project one validated capture-journal reference into the shared observation shape. */ +export const projectProcessObservation = ( + capture: ProcessCapture, + location: ProcessObservationLocation, +): ProcessObservation | null => { + const projected = projectedRecord(capture, location); + if (projected === null) return null; + return createProcessObservation({ + source: projected.source, + source_sequence: location.index, + captured_at_ms: projected.captured_at_ms, + subject_id: projected.subject_id, + location, + payload: projected.payload, + }); +}; diff --git a/src/domain/processReactiveCheckpointDataflow.ts b/src/domain/processReactiveCheckpointDataflow.ts new file mode 100644 index 000000000..8bdf3064f --- /dev/null +++ b/src/domain/processReactiveCheckpointDataflow.ts @@ -0,0 +1,41 @@ +import type { ProcessReactiveScenario } from "./processReactiveScenario.js"; + +/** Checkpoints guaranteed on every known path entering each reactive state. */ +export const definitelyAvailableCheckpoints = ( + scenario: ProcessReactiveScenario, +): ReadonlyMap> => { + const available = new Map>([ + [scenario.initial_state, new Set()], + ]); + let changed = true; + while (changed) { + changed = false; + for (const state of scenario.states) { + const before = available.get(state.id); + if (before === undefined) continue; + for (const transition of state.on) { + if (transition.target.kind !== "goto") continue; + const after = new Set(before); + for (const name of transition.actions.flatMap((action) => + action.type === "checkpoint" ? [action.name] : [], + )) + after.add(name); + const prior = available.get(transition.target.state); + const next = + prior === undefined + ? after + : new Set([...prior].filter((name) => after.has(name))); + if (prior !== undefined && setsEqual(prior, next)) continue; + available.set(transition.target.state, next); + changed = true; + } + } + } + return available; +}; + +const setsEqual = ( + left: ReadonlySet, + right: ReadonlySet, +): boolean => + left.size === right.size && [...left].every((value) => right.has(value)); diff --git a/src/domain/processReactiveMatching.ts b/src/domain/processReactiveMatching.ts new file mode 100644 index 000000000..dd121164e --- /dev/null +++ b/src/domain/processReactiveMatching.ts @@ -0,0 +1,302 @@ +import canonicalize from "canonicalize"; + +import type { ProcessObservation } from "./processObservation.js"; +import { comparableProcessObservationPayload } from "./processObservation.js"; +import { + PROCESS_REACTIVE_LIMITS, + type ProcessReactiveFrontier, + type ProcessReactiveTrigger, +} from "./processReactiveScenario.js"; +import type { ProcessReactiveSnapshot } from "./processReactiveRuntime.js"; + +/** Internal bounded trigger-match result used by the pure reducer. */ +export interface ProcessReactiveTriggerMatch { + readonly matched: boolean; + readonly eventIds: readonly string[]; + readonly consumeIds: readonly string[]; + readonly lastOrder: number; + readonly overflow: boolean; +} + +interface EvaluationBudget { + remaining: number; +} + +interface TriggerMatchContext { + readonly snapshot: ProcessReactiveSnapshot; + readonly budget: EvaluationBudget; + readonly afterOrder: number; + readonly upperOrder: number | null; +} + +const spendEvaluationWork = ( + budget: EvaluationBudget, + units: number, +): boolean => { + if (units > budget.remaining) return false; + budget.remaining -= units; + return true; +}; + +const noMatch = (overflow = false): ProcessReactiveTriggerMatch => ({ + matched: false, + eventIds: [], + consumeIds: [], + lastOrder: -1, + overflow, +}); + +const frontierOrder = ( + frontier: ProcessReactiveFrontier, + snapshot: ProcessReactiveSnapshot, +): number | null => { + switch (frontier.kind) { + case "scenario_start": + return -1; + case "state_entry": + return snapshot.state_entry_capture_order - 1; + case "checkpoint": + return ( + snapshot.checkpoints.findLast(({ name }) => name === frontier.name) + ?.capture_order ?? null + ); + case "event_id": + return ( + snapshot.observations.find( + ({ event_id }) => event_id === frontier.event_id, + )?.capture_order ?? null + ); + } +}; + +const eligibleObservations = ( + frontier: ProcessReactiveFrontier, + context: TriggerMatchContext, +): readonly ProcessObservation[] => { + const frontierValue = frontierOrder(frontier, context.snapshot); + if (frontierValue === null) return []; + const lowerBound = Math.max(frontierValue, context.afterOrder); + const consumed = new Set(context.snapshot.consumed_event_ids); + return context.snapshot.observations.filter( + ({ capture_order, event_id }) => + capture_order > lowerBound && + (context.upperOrder === null || capture_order <= context.upperOrder) && + !consumed.has(event_id), + ); +}; + +const canonicalJson = (value: unknown): string => { + const serialized = canonicalize(value); + if (serialized === undefined) + throw new TypeError("Reactive scenario value is not canonical JSON"); + return serialized; +}; + +const matchEvent = ( + trigger: Extract, + context: TriggerMatchContext, +): ProcessReactiveTriggerMatch => { + const expected = canonicalJson( + comparableProcessObservationPayload(trigger.exact, trigger.ignore_fields), + ); + const matches: ProcessObservation[] = []; + for (const observation of eligibleObservations(trigger.since, context)) { + if (!spendEvaluationWork(context.budget, 1)) return noMatch(true); + if ( + observation.source === trigger.source && + canonicalJson( + comparableProcessObservationPayload( + observation.payload, + trigger.ignore_fields, + ), + ) === expected + ) + matches.push(observation); + } + if ( + matches.length < trigger.cardinality.min || + matches.length > trigger.cardinality.max + ) + return noMatch(); + const eventIds = matches.map(({ event_id }) => event_id); + return { + matched: true, + eventIds, + consumeIds: trigger.consume ? eventIds : [], + lastOrder: matches.at(-1)?.capture_order ?? context.afterOrder, + overflow: false, + }; +}; + +const occurrenceRange = ( + haystack: string, + needle: string, + occurrence: number, +): { readonly start: number; readonly end: number } | null => { + let offset = 0; + let start = -1; + for (let found = 0; found < occurrence; found += 1) { + start = haystack.indexOf(needle, offset); + if (start < 0) return null; + offset = start + needle.length; + } + return { start, end: offset }; +}; + +const matchTerminalText = ( + trigger: Extract, + context: TriggerMatchContext, +): ProcessReactiveTriggerMatch => { + const chunks: { observation: ProcessObservation; text: string }[] = []; + for (const observation of eligibleObservations(trigger.since, context)) { + if (!spendEvaluationWork(context.budget, 1)) return noMatch(true); + if ( + observation.source === "terminal_raw" && + typeof observation.payload === "object" && + observation.payload !== null && + !Array.isArray(observation.payload) && + typeof observation.payload["data"] === "string" + ) + chunks.push({ observation, text: observation.payload["data"] }); + } + let rawBytes = 0; + for (const { text } of chunks) { + if (text.length > PROCESS_REACTIVE_LIMITS.terminalMatchBytes) + return noMatch(true); + const bytes = Buffer.byteLength(text); + rawBytes += bytes; + if ( + rawBytes > PROCESS_REACTIVE_LIMITS.terminalMatchBytes || + !spendEvaluationWork(context.budget, bytes) + ) + return noMatch(true); + } + const range = occurrenceRange( + chunks.map(({ text }) => text).join(""), + trigger.literal, + trigger.occurrence, + ); + if (range === null) return noMatch(); + let firstChunk = 0; + let lastChunk = 0; + let characters = 0; + for (const [index, chunk] of chunks.entries()) { + characters += chunk.text.length; + if (characters <= range.start) firstChunk = index + 1; + if (characters >= range.end) { + lastChunk = index; + break; + } + } + const matched = chunks + .slice(firstChunk, lastChunk + 1) + .map(({ observation }) => observation); + const eventIds = matched.map(({ event_id }) => event_id); + return { + matched: true, + eventIds, + consumeIds: trigger.consume ? eventIds : [], + lastOrder: matched.at(-1)?.capture_order ?? context.afterOrder, + overflow: false, + }; +}; + +const mergeMatches = ( + matches: readonly ProcessReactiveTriggerMatch[], +): ProcessReactiveTriggerMatch => ({ + matched: matches.every(({ matched }) => matched), + eventIds: [...new Set(matches.flatMap(({ eventIds }) => eventIds))], + consumeIds: [...new Set(matches.flatMap(({ consumeIds }) => consumeIds))], + lastOrder: Math.max(-1, ...matches.map(({ lastOrder }) => lastOrder)), + overflow: matches.some(({ overflow }) => overflow), +}); + +const matchEarliestPrefix = ( + trigger: ProcessReactiveTrigger, + context: TriggerMatchContext, +): ProcessReactiveTriggerMatch => { + for (const observation of context.snapshot.observations) { + if (observation.capture_order <= context.afterOrder) continue; + if ( + context.upperOrder !== null && + observation.capture_order > context.upperOrder + ) + break; + const matched = matchTrigger(trigger, { + ...context, + upperOrder: observation.capture_order, + }); + if (matched.matched || matched.overflow) return matched; + } + return noMatch(); +}; + +const matchTrigger = ( + trigger: ProcessReactiveTrigger, + context: TriggerMatchContext, +): ProcessReactiveTriggerMatch => { + if (trigger.kind === "event") return matchEvent(trigger, context); + if (trigger.kind === "terminal_text") + return matchTerminalText(trigger, context); + if (trigger.kind === "all") + return mergeMatches( + trigger.triggers.map((child) => matchTrigger(child, context)), + ); + if (trigger.kind === "any") { + const evaluated = trigger.triggers.map((child) => + matchTrigger(child, context), + ); + const matches = evaluated.filter(({ matched }) => matched); + return matches.length === 0 + ? noMatch(evaluated.some(({ overflow }) => overflow)) + : mergeMatches(matches); + } + if (trigger.kind === "sequence") { + const matches: ProcessReactiveTriggerMatch[] = []; + let cursor = context.afterOrder; + for (const child of trigger.triggers) { + const matched = matchEarliestPrefix(child, { + ...context, + afterOrder: cursor, + }); + if (!matched.matched) return noMatch(matched.overflow); + matches.push(matched); + cursor = matched.lastOrder; + } + return mergeMatches(matches); + } + const matches: ProcessReactiveTriggerMatch[] = []; + let cursor = context.afterOrder; + while (matches.length < trigger.max) { + const matched = matchEarliestPrefix(trigger.trigger, { + ...context, + afterOrder: cursor, + }); + if (!matched.matched || matched.lastOrder <= cursor) break; + matches.push(matched); + cursor = matched.lastOrder; + } + if (matches.length === trigger.max) { + const excess = matchEarliestPrefix(trigger.trigger, { + ...context, + afterOrder: cursor, + }); + if (excess.matched || excess.overflow) return noMatch(excess.overflow); + } + if (matches.length < trigger.min) + return noMatch(matches.some(({ overflow }) => overflow)); + return mergeMatches(matches); +}; + +/** Match one trigger using a shared deterministic work budget. */ +export const matchProcessReactiveTrigger = ( + trigger: ProcessReactiveTrigger, + snapshot: ProcessReactiveSnapshot, + budget: EvaluationBudget, +): ProcessReactiveTriggerMatch => + matchTrigger(trigger, { + snapshot, + budget, + afterOrder: -1, + upperOrder: null, + }); diff --git a/src/domain/processReactiveRuntime.ts b/src/domain/processReactiveRuntime.ts new file mode 100644 index 000000000..deb1c155a --- /dev/null +++ b/src/domain/processReactiveRuntime.ts @@ -0,0 +1,263 @@ +import type { ProcessObservation } from "./processObservation.js"; +import { matchProcessReactiveTrigger } from "./processReactiveMatching.js"; +import { + PROCESS_REACTIVE_LIMITS, + type ProcessReactiveAction, + type ProcessReactiveScenario, +} from "./processReactiveScenario.js"; +export { commitProcessReactiveProposal } from "./processReactiveTransition.js"; + +/** Terminal and failure outcomes produced by the pure reactive reducer. */ +export type ProcessReactiveOutcome = + | "passed" + | "predicate_timeout" + | "scenario_deadline" + | "ambiguous_match" + | "action_rejected" + | "target_lost" + | "capture_incomplete" + | "cancelled" + | "cleanup_failed"; + +/** One deterministic scenario transition suitable for later Evidence projection. */ +export interface ProcessReactiveTransitionRecord { + readonly sequence: number; + readonly transition_id: string; + readonly state_before: string; + readonly state_after: string | null; + readonly outcome: ProcessReactiveOutcome | null; + readonly trigger_event_ids: readonly string[]; + readonly action_event_ids: readonly string[]; + readonly action_types: readonly ProcessReactiveAction["type"][]; +} + +interface Counter { + readonly id: string; + readonly count: number; +} + +interface ReactiveCheckpoint { + readonly name: string; + readonly event_id: string; + readonly capture_order: number; +} + +/** Immutable reducer state owned by one validated reactive scenario run. */ +export interface ProcessReactiveSnapshot { + readonly status: "running" | "finished"; + readonly outcome: ProcessReactiveOutcome | null; + readonly active_state: string; + readonly state_entry_capture_order: number; + readonly observations: readonly ProcessObservation[]; + readonly consumed_event_ids: readonly string[]; + readonly transition_uses: readonly Counter[]; + readonly state_visits: readonly Counter[]; + readonly checkpoints: readonly ReactiveCheckpoint[]; + readonly transitions: readonly ProcessReactiveTransitionRecord[]; +} + +/** External fact offered to the pure reducer. */ +export type ProcessReactiveInput = + | { readonly kind: "observation"; readonly observation: ProcessObservation } + | { + readonly kind: "state_deadline"; + readonly state_id: string; + readonly state_entry_capture_order: number; + } + | { readonly kind: "scenario_deadline" } + | { readonly kind: "cancelled" } + | { readonly kind: "cleanup_failed" }; + +/** Per-transition matching summary returned without caller payloads. */ +export interface ProcessReactiveEvaluation { + readonly transition_id: string; + readonly status: "matched" | "unmatched" | "exhausted"; + readonly matched_event_ids: readonly string[]; +} + +/** Host acknowledgement for one proposed declarative effect. */ +export type ProcessReactiveEffectResult = + | { + readonly status: "succeeded"; + readonly observation: ProcessObservation; + } + | { readonly status: "rejected" } + | { readonly status: "target_lost" }; + +/** Immutable transition proposal awaiting host effect results. */ +export interface ProcessReactiveTransitionProposal { + readonly transition_id: string; + readonly transition_identity: string; + readonly observation_event_id: string; + readonly trigger_event_ids: readonly string[]; + readonly consume_event_ids: readonly string[]; +} + +/** Pure scenario decision; effects remain declarative for an authorized host. */ +export type ProcessReactiveDecision = + | { + readonly kind: "waiting" | "already_finished"; + readonly snapshot: ProcessReactiveSnapshot; + readonly evaluations: readonly ProcessReactiveEvaluation[]; + } + | { + readonly kind: "proposal"; + readonly snapshot: ProcessReactiveSnapshot; + readonly evaluations: readonly ProcessReactiveEvaluation[]; + readonly effects: readonly ProcessReactiveAction[]; + readonly proposal: ProcessReactiveTransitionProposal; + } + | { + readonly kind: "transition"; + readonly snapshot: ProcessReactiveSnapshot; + readonly evaluations: readonly ProcessReactiveEvaluation[]; + readonly record: ProcessReactiveTransitionRecord; + } + | { + readonly kind: "finished"; + readonly snapshot: ProcessReactiveSnapshot; + readonly evaluations: readonly ProcessReactiveEvaluation[]; + readonly outcome: ProcessReactiveOutcome; + }; + +const counterValue = (counters: readonly Counter[], id: string): number => + counters.find((counter) => counter.id === id)?.count ?? 0; + +/** Create the initial immutable reducer state for one validated scenario. */ +export const createProcessReactiveSnapshot = ( + scenario: ProcessReactiveScenario, +): ProcessReactiveSnapshot => ({ + status: "running", + outcome: null, + active_state: scenario.initial_state, + state_entry_capture_order: 0, + observations: [], + consumed_event_ids: [], + transition_uses: [], + state_visits: [{ id: scenario.initial_state, count: 1 }], + checkpoints: [], + transitions: [], +}); + +const finishedDecision = ( + snapshot: ProcessReactiveSnapshot, + outcome: ProcessReactiveOutcome, + evaluations: readonly ProcessReactiveEvaluation[] = [], +): ProcessReactiveDecision => ({ + kind: "finished", + outcome, + evaluations, + snapshot: { ...snapshot, status: "finished", outcome }, +}); + +const reduceControlInput = ( + snapshot: ProcessReactiveSnapshot, + input: ProcessReactiveInput, +): ProcessReactiveDecision | null => { + if (input.kind === "cleanup_failed") + return finishedDecision(snapshot, "cleanup_failed"); + if (snapshot.status === "finished") + return { kind: "already_finished", snapshot, evaluations: [] }; + if (input.kind === "observation") return null; + if (input.kind === "state_deadline") + return input.state_id === snapshot.active_state && + input.state_entry_capture_order === snapshot.state_entry_capture_order + ? finishedDecision(snapshot, "predicate_timeout") + : { kind: "waiting", snapshot, evaluations: [] }; + if (input.kind === "scenario_deadline") + return finishedDecision(snapshot, "scenario_deadline"); + return finishedDecision(snapshot, "cancelled"); +}; + +/** Advance one validated scenario using only an observation or explicit deadline. */ +export const reduceProcessReactiveScenario = ( + scenario: ProcessReactiveScenario, + snapshot: ProcessReactiveSnapshot, + input: ProcessReactiveInput, +): ProcessReactiveDecision => { + const controlDecision = reduceControlInput(snapshot, input); + if (controlDecision !== null) return controlDecision; + if (input.kind !== "observation") + return finishedDecision(snapshot, "capture_incomplete"); + const prior = snapshot.observations.at(-1); + if ( + (prior !== undefined && + input.observation.capture_order <= prior.capture_order) || + snapshot.observations.some( + ({ event_id }) => event_id === input.observation.event_id, + ) + ) + return finishedDecision(snapshot, "capture_incomplete"); + if ( + snapshot.observations.length >= PROCESS_REACTIVE_LIMITS.retainedObservations + ) + return finishedDecision(snapshot, "capture_incomplete"); + const observed = { + ...snapshot, + observations: [...snapshot.observations, input.observation], + }; + const state = scenario.states.find(({ id }) => id === observed.active_state); + if (state === undefined) return finishedDecision(observed, "target_lost"); + const budget = { remaining: PROCESS_REACTIVE_LIMITS.evaluationWork }; + const matches = state.on.map((transition) => ({ + transition, + match: matchProcessReactiveTrigger(transition.when, observed, budget), + uses: counterValue(observed.transition_uses, transition.id), + })); + if (matches.some(({ match }) => match.overflow)) + return finishedDecision(observed, "capture_incomplete"); + const evaluations: ProcessReactiveEvaluation[] = matches.map( + ({ transition, match, uses }) => ({ + transition_id: transition.id, + status: + uses >= transition.max_uses + ? "exhausted" + : match.matched + ? "matched" + : "unmatched", + matched_event_ids: match.eventIds, + }), + ); + const eligible = matches.filter( + ({ transition, match, uses }) => + match.matched && uses < transition.max_uses, + ); + if (eligible.length === 0) + return { kind: "waiting", snapshot: observed, evaluations }; + const minimumPriority = Math.min( + ...eligible.map(({ transition }) => transition.priority), + ); + const winners = eligible.filter( + ({ transition }) => transition.priority === minimumPriority, + ); + if (winners.length !== 1) + return finishedDecision(observed, "ambiguous_match", evaluations); + const winner = winners[0]; + if (winner === undefined) + return finishedDecision(observed, "capture_incomplete", evaluations); + const targetDeclaration = winner.transition.target; + if (targetDeclaration.kind === "goto") { + const target = scenario.states.find( + ({ id }) => id === targetDeclaration.state, + ); + if ( + target === undefined || + counterValue(observed.state_visits, target.id) >= target.max_visits + ) + return finishedDecision(observed, "capture_incomplete", evaluations); + } + return { + kind: "proposal", + snapshot: observed, + evaluations, + effects: winner.transition.actions, + proposal: { + transition_id: winner.transition.id, + transition_identity: canonicalize(winner.transition) ?? "", + observation_event_id: input.observation.event_id, + trigger_event_ids: winner.match.eventIds, + consume_event_ids: winner.match.consumeIds, + }, + }; +}; +import canonicalize from "canonicalize"; diff --git a/src/domain/processReactiveScenario.ts b/src/domain/processReactiveScenario.ts new file mode 100644 index 000000000..0014faf71 --- /dev/null +++ b/src/domain/processReactiveScenario.ts @@ -0,0 +1,500 @@ +import { z } from "zod"; + +import { jsonValueSchema } from "./jsonValue.js"; +import { processObservationSourceSchema } from "./processObservation.js"; +import { definitelyAvailableCheckpoints } from "./processReactiveCheckpointDataflow.js"; +import { preflightProcessReactiveScenario } from "./processReactiveScenarioPreflight.js"; + +/** Fixed public bounds for Process Reactive Scenario v1. */ +export const PROCESS_REACTIVE_LIMITS = { + states: 32, + transitions: 128, + triggerDepth: 6, + predicates: 64, + actions: 256, + actionsPerTransition: 16, + childrenPerComposite: 16, + stateVisits: 256, + transitionUses: 256, + repeat: 64, + retainedObservations: 256, + terminalMatchBytes: 65_536, + evaluationWork: 8_192, + jsonDepth: 20, + jsonNodes: 1_024, + runtimeMs: 300_000, +} as const; + +const identifierSchema = z.string().regex(/^[A-Za-z][A-Za-z0-9._-]{0,63}$/u); +const checkpointNameSchema = z + .string() + .regex(/^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$/u); +const positiveRuntime = z + .number() + .int() + .positive() + .max(PROCESS_REACTIVE_LIMITS.runtimeMs); + +const frontierSchema = z.discriminatedUnion("kind", [ + z.strictObject({ kind: z.literal("scenario_start") }), + z.strictObject({ kind: z.literal("state_entry") }), + z.strictObject({ kind: z.literal("checkpoint"), name: checkpointNameSchema }), + z.strictObject({ kind: z.literal("event_id"), event_id: identifierSchema }), +]); +export type ProcessReactiveFrontier = z.infer; + +const ignoredFieldSchema = z.enum([ + "sequence", + "at_ms", + "scheduled_at_ms", + "dispatched_at_ms", + "elapsed_ms", +]); + +interface EventTrigger { + readonly kind: "event"; + readonly source: z.infer; + readonly exact: z.infer; + readonly ignore_fields: readonly z.infer[]; + readonly since: ProcessReactiveFrontier; + readonly consume: boolean; + readonly cardinality: { readonly min: number; readonly max: number }; +} + +interface TerminalTextTrigger { + readonly kind: "terminal_text"; + readonly view: "decoded"; + readonly encoding: "utf8"; + readonly literal: string; + readonly case_sensitive: true; + readonly control_sequences: "include"; + readonly occurrence: number; + readonly since: ProcessReactiveFrontier; + readonly consume: boolean; +} + +type TriggerGroup = + | { + readonly kind: "all"; + readonly triggers: readonly ProcessReactiveTrigger[]; + } + | { + readonly kind: "any"; + readonly triggers: readonly ProcessReactiveTrigger[]; + } + | { + readonly kind: "sequence"; + readonly triggers: readonly ProcessReactiveTrigger[]; + }; + +interface RepeatTrigger { + readonly kind: "repeat"; + readonly trigger: ProcessReactiveTrigger; + readonly min: number; + readonly max: number; +} + +/** Declarative, bounded predicate tree evaluated against process observations. */ +export type ProcessReactiveTrigger = + | EventTrigger + | TerminalTextTrigger + | TriggerGroup + | RepeatTrigger; + +const processReactiveTriggerSchema: z.ZodType = z.lazy( + () => + z.discriminatedUnion("kind", [ + z.strictObject({ + kind: z.literal("event"), + source: processObservationSourceSchema, + exact: jsonValueSchema, + ignore_fields: z + .array(ignoredFieldSchema) + .max(5) + .refine((values) => new Set(values).size === values.length, { + message: "ignored event fields must be unique", + }), + since: frontierSchema, + consume: z.boolean(), + cardinality: z + .strictObject({ + min: z.number().int().positive().max(256), + max: z.number().int().positive().max(256), + }) + .refine(({ min, max }) => min <= max, { + message: "cardinality min must not exceed max", + }), + }), + z.strictObject({ + kind: z.literal("terminal_text"), + view: z.literal("decoded"), + encoding: z.literal("utf8"), + literal: z.string().min(1).max(10_000), + case_sensitive: z.literal(true), + control_sequences: z.literal("include"), + occurrence: z.number().int().positive().max(1_000), + since: frontierSchema, + consume: z.boolean(), + }), + z.strictObject({ + kind: z.literal("all"), + triggers: z + .array(processReactiveTriggerSchema) + .min(2) + .max(PROCESS_REACTIVE_LIMITS.childrenPerComposite), + }), + z.strictObject({ + kind: z.literal("any"), + triggers: z + .array(processReactiveTriggerSchema) + .min(2) + .max(PROCESS_REACTIVE_LIMITS.childrenPerComposite), + }), + z.strictObject({ + kind: z.literal("sequence"), + triggers: z + .array(processReactiveTriggerSchema) + .min(1) + .max(PROCESS_REACTIVE_LIMITS.childrenPerComposite), + }), + z + .strictObject({ + kind: z.literal("repeat"), + trigger: processReactiveTriggerSchema, + min: z.number().int().positive().max(PROCESS_REACTIVE_LIMITS.repeat), + max: z.number().int().positive().max(PROCESS_REACTIVE_LIMITS.repeat), + }) + .refine(({ min, max }) => min <= max, { + message: "repeat min must not exceed max", + }), + ]), +); + +const signalTargetSchema = z.discriminatedUnion("kind", [ + z.strictObject({ kind: z.literal("root") }), + z.strictObject({ kind: z.literal("process_group") }), + z.strictObject({ + kind: z.literal("subject_id"), + subject_id: identifierSchema, + }), +]); + +const processReactiveActionSchema = z.discriminatedUnion("type", [ + z.strictObject({ + type: z.literal("send_input"), + data: z.string().max(1_000_000), + sensitive: z.boolean(), + }), + z.strictObject({ + type: z.literal("resize"), + columns: z.number().int().min(1).max(1_000), + rows: z.number().int().min(1).max(1_000), + }), + z.strictObject({ + type: z.literal("send_signal"), + target: signalTargetSchema, + signal: z.enum(["SIGINT", "SIGTERM", "SIGKILL"]), + }), + z.strictObject({ type: z.literal("checkpoint"), name: checkpointNameSchema }), +]); +export type ProcessReactiveAction = z.infer; + +const transitionTargetSchema = z.discriminatedUnion("kind", [ + z.strictObject({ kind: z.literal("goto"), state: identifierSchema }), + z.strictObject({ kind: z.literal("finish"), outcome: z.literal("passed") }), +]); + +const transitionSchema = z.strictObject({ + id: identifierSchema, + priority: z.number().int().min(0).max(1_000), + max_uses: z + .number() + .int() + .positive() + .max(PROCESS_REACTIVE_LIMITS.transitionUses), + when: processReactiveTriggerSchema, + actions: z + .array(processReactiveActionSchema) + .max(PROCESS_REACTIVE_LIMITS.actionsPerTransition), + target: transitionTargetSchema, +}); + +const stateSchema = z.strictObject({ + id: identifierSchema, + max_visits: z + .number() + .int() + .positive() + .max(PROCESS_REACTIVE_LIMITS.stateVisits), + deadline_ms: positiveRuntime, + on: z.array(transitionSchema).min(1).max(128), +}); + +const scenarioShapeSchema = z.strictObject({ + version: z.literal(1), + initial_state: identifierSchema, + deadline_ms: positiveRuntime, + states: z.array(stateSchema).min(1).max(PROCESS_REACTIVE_LIMITS.states), +}); + +/** Parsed Process Reactive Scenario v1 declaration. */ +export type ProcessReactiveScenario = z.infer; + +type TriggerMeasurements = { + readonly depth: number; + readonly predicates: number; +}; + +const measureTrigger = ( + trigger: ProcessReactiveTrigger, +): TriggerMeasurements => { + if (trigger.kind === "event" || trigger.kind === "terminal_text") + return { depth: 1, predicates: 1 }; + const children = + trigger.kind === "repeat" ? [trigger.trigger] : trigger.triggers; + const measured = children.map(measureTrigger); + return { + depth: 1 + Math.max(...measured.map(({ depth }) => depth)), + predicates: measured.reduce( + (total, { predicates }) => total + predicates, + 0, + ), + }; +}; + +const collectCheckpointFrontiers = ( + trigger: ProcessReactiveTrigger, +): readonly string[] => { + if (trigger.kind === "event" || trigger.kind === "terminal_text") + return trigger.since.kind === "checkpoint" ? [trigger.since.name] : []; + const children = + trigger.kind === "repeat" ? [trigger.trigger] : trigger.triggers; + return children.flatMap(collectCheckpointFrontiers); +}; + +const validateTriggerSemantics = ( + trigger: ProcessReactiveTrigger, + path: PropertyKey[], + context: z.RefinementCtx, +): void => { + if (trigger.kind === "event") { + const exact = trigger.exact; + if ( + trigger.ignore_fields.length > 0 && + (typeof exact !== "object" || exact === null || Array.isArray(exact)) + ) + context.addIssue({ + code: "custom", + message: "ignored fields require an exact object payload", + path: [...path, "ignore_fields"], + }); + if ( + typeof exact === "object" && + exact !== null && + !Array.isArray(exact) && + trigger.ignore_fields.some((field) => field in exact) + ) + context.addIssue({ + code: "custom", + message: "exact payload must omit every ignored field", + path: [...path, "exact"], + }); + return; + } + if (trigger.kind === "terminal_text") return; + if (trigger.kind === "repeat") { + validateTriggerSemantics(trigger.trigger, [...path, "trigger"], context); + return; + } + for (const [index, child] of trigger.triggers.entries()) + validateTriggerSemantics(child, [...path, "triggers", index], context); +}; + +type GraphValidation = { + readonly knownStates: ReadonlySet; + readonly transitionIds: Set; + readonly checkpoints: Set; + readonly checkpointReferences: Array<{ + readonly name: string; + readonly stateId: string; + readonly path: PropertyKey[]; + }>; + readonly adjacency: ReadonlyMap>; + transitions: number; + predicates: number; + actions: number; +}; + +type ReactiveTransition = + ProcessReactiveScenario["states"][number]["on"][number]; + +const validateTransition = (input: { + readonly stateId: string; + readonly transition: ReactiveTransition; + readonly path: PropertyKey[]; + readonly validation: GraphValidation; + readonly context: z.RefinementCtx; +}): void => { + const { transition, validation, path, context } = input; + validation.transitions += 1; + validation.actions += transition.actions.length; + const measured = measureTrigger(transition.when); + validation.predicates += measured.predicates; + validateTriggerSemantics(transition.when, [...path, "when"], context); + if (measured.depth > PROCESS_REACTIVE_LIMITS.triggerDepth) + context.addIssue({ + code: "custom", + message: "trigger nesting exceeds the v1 limit", + path: [...path, "when"], + }); + if (validation.transitionIds.has(transition.id)) + context.addIssue({ + code: "custom", + message: "transition ids must be unique", + path: [...path, "id"], + }); + validation.transitionIds.add(transition.id); + for (const action of transition.actions) { + if (action.type !== "checkpoint") continue; + if (validation.checkpoints.has(action.name)) + context.addIssue({ + code: "custom", + message: "checkpoint action names must be unique", + path: [...path, "actions"], + }); + validation.checkpoints.add(action.name); + } + for (const name of collectCheckpointFrontiers(transition.when)) + validation.checkpointReferences.push({ + name, + stateId: input.stateId, + path: [...path, "when"], + }); + if (transition.target.kind !== "goto") return; + if (!validation.knownStates.has(transition.target.state)) { + context.addIssue({ + code: "custom", + message: "transition target state is not declared", + path: [...path, "target", "state"], + }); + return; + } + validation.adjacency.get(input.stateId)?.add(transition.target.state); +}; + +const validateTotalsAndReferences = ( + scenario: ProcessReactiveScenario, + validation: GraphValidation, + context: z.RefinementCtx, +): void => { + for (const [count, limit, message] of [ + [ + validation.transitions, + PROCESS_REACTIVE_LIMITS.transitions, + "scenario has too many transitions", + ], + [ + validation.predicates, + PROCESS_REACTIVE_LIMITS.predicates, + "scenario has too many predicates", + ], + [ + validation.actions, + PROCESS_REACTIVE_LIMITS.actions, + "scenario has too many actions", + ], + ] as const) + if (count > limit) + context.addIssue({ code: "custom", message, path: ["states"] }); + const available = definitelyAvailableCheckpoints(scenario); + for (const reference of validation.checkpointReferences) + if (!available.get(reference.stateId)?.has(reference.name)) + context.addIssue({ + code: "custom", + message: + "checkpoint frontier is not definitely available before this transition", + path: reference.path, + }); +}; + +const validateReachability = ( + scenario: ProcessReactiveScenario, + stateIds: readonly string[], + validation: GraphValidation, + context: z.RefinementCtx, +): void => { + const reachable = new Set(); + const pending = validation.knownStates.has(scenario.initial_state) + ? [scenario.initial_state] + : []; + while (pending.length > 0) { + const current = pending.pop(); + if (current === undefined || reachable.has(current)) continue; + reachable.add(current); + pending.push(...(validation.adjacency.get(current) ?? [])); + } + for (const [index, id] of stateIds.entries()) + if (!reachable.has(id)) + context.addIssue({ + code: "custom", + message: "state is unreachable from the initial state", + path: ["states", index], + }); +}; + +const validateScenarioGraph = ( + scenario: ProcessReactiveScenario, + context: z.RefinementCtx, +): void => { + const stateIds = scenario.states.map(({ id }) => id); + const knownStates = new Set(stateIds); + if (knownStates.size !== stateIds.length) + context.addIssue({ + code: "custom", + message: "state ids must be unique", + path: ["states"], + }); + if (!knownStates.has(scenario.initial_state)) + context.addIssue({ + code: "custom", + message: "initial state is not declared", + path: ["initial_state"], + }); + const validation: GraphValidation = { + knownStates, + transitionIds: new Set(), + checkpoints: new Set(), + checkpointReferences: [], + adjacency: new Map(stateIds.map((id) => [id, new Set()])), + transitions: 0, + predicates: 0, + actions: 0, + }; + for (const [stateIndex, state] of scenario.states.entries()) { + if (state.deadline_ms > scenario.deadline_ms) + context.addIssue({ + code: "custom", + message: "state deadline must not exceed scenario deadline", + path: ["states", stateIndex, "deadline_ms"], + }); + for (const [transitionIndex, transition] of state.on.entries()) + validateTransition({ + stateId: state.id, + transition, + path: ["states", stateIndex, "on", transitionIndex], + validation, + context, + }); + } + validateTotalsAndReferences(scenario, validation, context); + validateReachability(scenario, stateIds, validation, context); +}; + +/** Strict boundary schema for a bounded declarative reactive scenario. */ +export const processReactiveScenarioSchema = z.preprocess( + (input, context) => + preflightProcessReactiveScenario(input, context, PROCESS_REACTIVE_LIMITS), + scenarioShapeSchema.superRefine((scenario, context) => + validateScenarioGraph(scenario, context), + ), +); diff --git a/src/domain/processReactiveScenarioPreflight.ts b/src/domain/processReactiveScenarioPreflight.ts new file mode 100644 index 000000000..db867fb75 --- /dev/null +++ b/src/domain/processReactiveScenarioPreflight.ts @@ -0,0 +1,165 @@ +import { z } from "zod"; + +interface ReactivePreflightLimits { + readonly states: number; + readonly transitions: number; + readonly predicates: number; + readonly triggerDepth: number; + readonly childrenPerComposite: number; + readonly jsonDepth: number; + readonly jsonNodes: number; +} + +type PendingTrigger = { readonly value: unknown; readonly depth: number }; + +interface TriggerTraversal { + readonly pending: PendingTrigger[]; + readonly context: z.RefinementCtx; + readonly limits: ReactivePreflightLimits; +} + +const isUnknownRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const reject = (context: z.RefinementCtx, message: string): false => { + context.addIssue({ code: "custom", message }); + return false; +}; + +const inspectJsonValue = ( + value: unknown, + context: z.RefinementCtx, + limits: ReactivePreflightLimits, +): boolean => { + const pending: PendingTrigger[] = [{ value, depth: 1 }]; + const seen = new WeakSet(); + let nodes = 0; + while (pending.length > 0) { + const current = pending.pop(); + if (current === undefined) break; + nodes += 1; + if (nodes > limits.jsonNodes) + return reject(context, "event exact JSON exceeds the v1 node limit"); + if (current.depth > limits.jsonDepth) + return reject(context, "event exact JSON exceeds the v1 depth limit"); + if (typeof current.value !== "object" || current.value === null) continue; + if (seen.has(current.value)) + return reject(context, "event exact JSON must not contain cycles"); + seen.add(current.value); + if (Array.isArray(current.value)) { + if (current.value.length + nodes > limits.jsonNodes) + return reject(context, "event exact JSON exceeds the v1 node limit"); + for (const child of current.value) + pending.push({ value: child, depth: current.depth + 1 }); + continue; + } + if (!isUnknownRecord(current.value)) continue; + for (const key in current.value) { + if (!Object.hasOwn(current.value, key)) continue; + if (nodes + pending.length >= limits.jsonNodes) + return reject(context, "event exact JSON exceeds the v1 node limit"); + pending.push({ + value: current.value[key], + depth: current.depth + 1, + }); + } + } + return true; +}; + +const collectTriggerRoots = ( + states: readonly unknown[], + context: z.RefinementCtx, + limits: ReactivePreflightLimits, +): PendingTrigger[] | null => { + const pending: PendingTrigger[] = []; + let transitions = 0; + for (const stateValue of states) { + if (!isUnknownRecord(stateValue) || !Array.isArray(stateValue["on"])) + continue; + transitions += stateValue["on"].length; + if (transitions > limits.transitions) { + reject(context, "scenario has too many transitions"); + return null; + } + for (const transitionValue of stateValue["on"]) + if (isUnknownRecord(transitionValue)) + pending.push({ value: transitionValue["when"], depth: 1 }); + } + return pending; +}; + +const enqueueTriggerChildren = ( + trigger: Record, + depth: number, + traversal: TriggerTraversal, +): boolean => { + if (trigger["kind"] === "repeat") { + traversal.pending.push({ value: trigger["trigger"], depth: depth + 1 }); + return true; + } + if ( + trigger["kind"] !== "all" && + trigger["kind"] !== "any" && + trigger["kind"] !== "sequence" + ) + return true; + const children = trigger["triggers"]; + if (!Array.isArray(children)) return true; + if (children.length > traversal.limits.childrenPerComposite) + return reject(traversal.context, "trigger group has too many children"); + for (const child of children) + traversal.pending.push({ value: child, depth: depth + 1 }); + return true; +}; + +const inspectTriggerTree = ( + pending: PendingTrigger[], + context: z.RefinementCtx, + limits: ReactivePreflightLimits, +): boolean => { + let nodes = 0; + const maximumNodes = limits.predicates * limits.triggerDepth; + while (pending.length > 0) { + const current = pending.pop(); + if (current === undefined) break; + nodes += 1; + if (nodes > maximumNodes) + return reject(context, "trigger tree exceeds the v1 limit"); + if (current.depth > limits.triggerDepth) + return reject(context, "trigger nesting exceeds the v1 limit"); + if ( + isUnknownRecord(current.value) && + current.value["kind"] === "event" && + !inspectJsonValue(current.value["exact"], context, limits) + ) + return false; + if ( + isUnknownRecord(current.value) && + !enqueueTriggerChildren(current.value, current.depth, { + pending, + context, + limits, + }) + ) + return false; + } + return true; +}; + +/** Reject oversized unknown trees before the recursive Zod schema visits them. */ +export const preflightProcessReactiveScenario = ( + input: unknown, + context: z.RefinementCtx, + limits: ReactivePreflightLimits, +): unknown => { + if (!isUnknownRecord(input) || !Array.isArray(input["states"])) return input; + if (input["states"].length > limits.states) { + reject(context, "scenario has too many states"); + return z.NEVER; + } + const pending = collectTriggerRoots(input["states"], context, limits); + if (pending === null || !inspectTriggerTree(pending, context, limits)) + return z.NEVER; + return input; +}; diff --git a/src/domain/processReactiveTransition.ts b/src/domain/processReactiveTransition.ts new file mode 100644 index 000000000..759dd89a4 --- /dev/null +++ b/src/domain/processReactiveTransition.ts @@ -0,0 +1,250 @@ +import type { ProcessObservation } from "./processObservation.js"; +import type { + ProcessReactiveDecision, + ProcessReactiveEffectResult, + ProcessReactiveEvaluation, + ProcessReactiveSnapshot, + ProcessReactiveTransitionRecord, +} from "./processReactiveRuntime.js"; +import type { ProcessReactiveScenario } from "./processReactiveScenario.js"; + +type ReactiveTransition = + ProcessReactiveScenario["states"][number]["on"][number]; + +const counterValue = ( + counters: readonly { readonly id: string; readonly count: number }[], + id: string, +): number => counters.find((counter) => counter.id === id)?.count ?? 0; + +const incrementCounter = ( + counters: readonly { readonly id: string; readonly count: number }[], + id: string, +) => { + let found = false; + const updated = counters.map((counter) => { + if (counter.id !== id) return counter; + found = true; + return { id, count: counter.count + 1 }; + }); + return found ? updated : [...updated, { id, count: 1 }]; +}; + +/** Commit one uniquely selected transition after its declarative effects succeed. */ +const applyProcessReactiveTransition = (input: { + readonly scenario: ProcessReactiveScenario; + readonly snapshot: ProcessReactiveSnapshot; + readonly observation: ProcessObservation; + readonly transition: ReactiveTransition; + readonly triggerEventIds: readonly string[]; + readonly consumeEventIds: readonly string[]; + readonly effectResults: readonly Extract< + ProcessReactiveEffectResult, + { readonly status: "succeeded" } + >[]; + readonly evaluations: readonly ProcessReactiveEvaluation[]; +}): ProcessReactiveDecision => { + const outcome = + input.transition.target.kind === "finish" + ? input.transition.target.outcome + : null; + const stateAfter = + input.transition.target.kind === "goto" + ? input.transition.target.state + : null; + const record: ProcessReactiveTransitionRecord = { + sequence: input.snapshot.transitions.length, + transition_id: input.transition.id, + state_before: input.snapshot.active_state, + state_after: stateAfter, + outcome, + trigger_event_ids: input.triggerEventIds, + action_event_ids: input.effectResults.map( + ({ observation }) => observation.event_id, + ), + action_types: input.transition.actions.map(({ type }) => type), + }; + const consumed = new Set(input.snapshot.consumed_event_ids); + for (const eventId of input.consumeEventIds) consumed.add(eventId); + const checkpoints = [...input.snapshot.checkpoints]; + for (const [index, action] of input.transition.actions.entries()) + if (action.type === "checkpoint") + checkpoints.push({ + name: action.name, + event_id: input.effectResults[index]!.observation.event_id, + capture_order: input.effectResults[index]!.observation.capture_order, + }); + const completedCaptureOrder = Math.max( + input.observation.capture_order, + ...input.effectResults.map(({ observation }) => observation.capture_order), + ); + let next: ProcessReactiveSnapshot = { + ...input.snapshot, + status: outcome === null ? "running" : "finished", + outcome, + active_state: stateAfter ?? input.snapshot.active_state, + state_entry_capture_order: + stateAfter === null + ? input.snapshot.state_entry_capture_order + : completedCaptureOrder + 1, + consumed_event_ids: [...consumed], + transition_uses: incrementCounter( + input.snapshot.transition_uses, + input.transition.id, + ), + state_visits: + stateAfter === null + ? input.snapshot.state_visits + : incrementCounter(input.snapshot.state_visits, stateAfter), + checkpoints, + transitions: [...input.snapshot.transitions, record], + }; + if (stateAfter !== null) { + const nextState = input.scenario.states.find(({ id }) => id === stateAfter); + if ( + nextState === undefined || + counterValue(next.state_visits, stateAfter) > nextState.max_visits + ) + next = { ...next, status: "finished", outcome: "capture_incomplete" }; + } + return next.outcome === "capture_incomplete" + ? { + kind: "finished", + snapshot: next, + evaluations: input.evaluations, + outcome: next.outcome, + } + : { + kind: "transition", + snapshot: next, + evaluations: input.evaluations, + record, + }; +}; + +const effectObservationMatchesAction = ( + action: ProcessReactiveScenario["states"][number]["on"][number]["actions"][number], + observation: ProcessObservation, +): boolean => { + const payload = + typeof observation.payload === "object" && + observation.payload !== null && + !Array.isArray(observation.payload) + ? observation.payload + : null; + const journalIdentityMatches = + observation.source_sequence === observation.location.index && + observation.capture_order === observation.location.capture_order && + observation.event_id === + `obs.${observation.location.collection}.${String(observation.location.index)}`; + if (!journalIdentityMatches || payload === null) return false; + if (action.type === "checkpoint") + return ( + observation.source === "filesystem" && + observation.location.collection === "filesystem_checkpoints" && + payload["name"] === action.name + ); + if ( + observation.source !== "interaction" || + observation.location.collection !== "interaction_events" || + payload["outcome"] !== "dispatched" + ) + return false; + if (action.type === "send_input") + return ( + payload["type"] === "input" && + payload["data"] === + (action.sensitive + ? `` + : action.data) + ); + if (action.type === "resize") + return ( + payload["type"] === "resize" && + payload["data"] === `${String(action.columns)}x${String(action.rows)}` + ); + return payload["type"] === "signal" && payload["data"] === action.signal; +}; + +const failedProposal = ( + proposed: Extract, + outcome: "action_rejected" | "target_lost", +): ProcessReactiveDecision => ({ + kind: "finished", + outcome, + evaluations: proposed.evaluations, + snapshot: { ...proposed.snapshot, status: "finished", outcome }, +}); + +/** Commit a proposal only after the host reports every effect as successful. */ +export const commitProcessReactiveProposal = ( + scenario: ProcessReactiveScenario, + proposed: Extract, + results: readonly ProcessReactiveEffectResult[], +): ProcessReactiveDecision => { + if (results.some(({ status }) => status === "target_lost")) + return failedProposal(proposed, "target_lost"); + const state = scenario.states.find( + ({ id }) => id === proposed.snapshot.active_state, + ); + const transition = state?.on.find( + ({ id }) => id === proposed.proposal.transition_id, + ); + const observation = proposed.snapshot.observations.find( + ({ event_id }) => event_id === proposed.proposal.observation_event_id, + ); + if ( + transition === undefined || + observation === undefined || + canonicalize(transition) !== proposed.proposal.transition_identity || + canonicalize(proposed.effects) !== canonicalize(transition.actions) + ) + return failedProposal(proposed, "target_lost"); + if ( + results.length !== transition.actions.length || + results.some(({ status }) => status === "rejected") + ) + return failedProposal(proposed, "action_rejected"); + const succeeded = results.filter( + ( + result, + ): result is Extract< + ProcessReactiveEffectResult, + { readonly status: "succeeded" } + > => result.status === "succeeded", + ); + let previousOrder = + proposed.snapshot.observations.find( + ({ event_id }) => event_id === proposed.proposal.observation_event_id, + )?.capture_order ?? -1; + const knownEventIds = new Set([ + ...proposed.snapshot.observations.map(({ event_id }) => event_id), + ...proposed.snapshot.transitions.flatMap( + ({ action_event_ids }) => action_event_ids, + ), + ]); + for (const [index, result] of succeeded.entries()) { + const action = transition.actions[index]; + const effectObservation = result.observation; + if ( + action === undefined || + !effectObservationMatchesAction(action, effectObservation) || + !Number.isSafeInteger(effectObservation.capture_order) || + effectObservation.capture_order <= previousOrder || + knownEventIds.has(effectObservation.event_id) + ) + return failedProposal(proposed, "action_rejected"); + knownEventIds.add(effectObservation.event_id); + previousOrder = effectObservation.capture_order; + } + return applyProcessReactiveTransition({ + scenario, + snapshot: proposed.snapshot, + observation, + transition, + triggerEventIds: proposed.proposal.trigger_event_ids, + consumeEventIds: proposed.proposal.consume_event_ids, + effectResults: succeeded, + evaluations: proposed.evaluations, + }); +}; +import canonicalize from "canonicalize"; diff --git a/src/domain/processTraceEvaluation.ts b/src/domain/processTraceEvaluation.ts index 10539e83b..31da9336f 100644 --- a/src/domain/processTraceEvaluation.ts +++ b/src/domain/processTraceEvaluation.ts @@ -1,7 +1,12 @@ import { z } from "zod"; -import { jsonValueSchema, type JsonValue } from "./jsonValue.js"; +import type { JsonValue } from "./jsonValue.js"; import type { ProcessCapture } from "./processCapture.js"; +import { + processObservationLocationSchema, + projectProcessObservation, + type ProcessObservationLocation, +} from "./processObservation.js"; import { canonicalTraceJson, comparableTracePayload, @@ -12,36 +17,9 @@ import { const identifierSchema = z.string().regex(/^[A-Za-z][A-Za-z0-9._-]{0,63}$/u); -export type ProcessTraceLocation = { - readonly collection: - | "frames" - | "rendered_frames" - | "interaction_events" - | "lifecycle" - | "process_samples" - | "filesystem_checkpoints" - | "shim_events" - | "protocol_events" - | "replay_transitions"; - readonly index: number; - readonly capture_order: number; -}; +export type ProcessTraceLocation = ProcessObservationLocation; -const locationSchema = z.strictObject({ - collection: z.enum([ - "frames", - "rendered_frames", - "interaction_events", - "lifecycle", - "process_samples", - "filesystem_checkpoints", - "shim_events", - "protocol_events", - "replay_transitions", - ]), - index: z.number().int().nonnegative(), - capture_order: z.number().int().nonnegative(), -}); +const locationSchema = processObservationLocationSchema; const matchedEventSchema = z.strictObject({ event_id: identifierSchema, @@ -88,71 +66,6 @@ type TraceRecord = { readonly location: ProcessTraceLocation; }; -const recordAt = ( - capture: ProcessCapture, - location: ProcessTraceLocation, -): Omit | null => { - const value = (() => { - switch (location.collection) { - case "frames": - return { - source: "terminal_raw" as const, - value: capture.frames[location.index], - }; - case "rendered_frames": - return { - source: "terminal_rendered" as const, - value: capture.rendered_frames[location.index], - }; - case "interaction_events": - return { - source: "interaction" as const, - value: capture.interaction_events[location.index], - }; - case "lifecycle": - return location.index === 0 - ? { - source: "lifecycle" as const, - value: { event: "exit", ...capture.exit }, - } - : location.index === 1 - ? { - source: "lifecycle" as const, - value: { event: "settlement", ...capture.settlement }, - } - : null; - case "process_samples": - return { - source: "process" as const, - value: capture.process_samples[location.index], - }; - case "filesystem_checkpoints": - return { - source: "filesystem" as const, - value: capture.filesystem_checkpoints[location.index], - }; - case "shim_events": - return { - source: "shim" as const, - value: capture.shim_events[location.index], - }; - case "protocol_events": { - const event = capture.protocol_events[location.index]; - return event === undefined - ? null - : { source: event.protocol, value: event }; - } - case "replay_transitions": - return { - source: "replay_transition" as const, - value: capture.replay_transitions[location.index], - }; - } - })(); - if (value === null || value.value === undefined) return null; - return { source: value.source, payload: jsonValueSchema.parse(value.value) }; -}; - const scopesFor = ( sources: ReadonlySet, ): ReadonlySet => @@ -244,9 +157,13 @@ const collectRecords = ( ): readonly TraceRecord[] => { const records: TraceRecord[] = []; for (const location of capture.event_journal ?? []) { - const record = recordAt(capture, location); + const record = projectProcessObservation(capture, location); if (record !== null && sources.has(record.source)) - records.push({ ...record, location }); + records.push({ + source: record.source, + payload: record.payload, + location, + }); } return records; }; diff --git a/src/domain/processTraceSpecification.ts b/src/domain/processTraceSpecification.ts index b15b8a017..e6243910e 100644 --- a/src/domain/processTraceSpecification.ts +++ b/src/domain/processTraceSpecification.ts @@ -2,23 +2,17 @@ import canonicalize from "canonicalize"; import { z } from "zod"; import { jsonValueSchema } from "./jsonValue.js"; +import { + comparableProcessObservationPayload, + processObservationSourceSchema, + type ProcessObservationSource, +} from "./processObservation.js"; const identifierSchema = z.string().regex(/^[A-Za-z][A-Za-z0-9._-]{0,63}$/u); /** Process observation families admitted by a declared trace specification. */ -export const processTraceSourceSchema = z.enum([ - "terminal_raw", - "terminal_rendered", - "interaction", - "lifecycle", - "process", - "filesystem", - "http", - "websocket", - "shim", - "replay_transition", -]); -export type ProcessTraceSource = z.infer; +export const processTraceSourceSchema = processObservationSourceSchema; +export type ProcessTraceSource = ProcessObservationSource; const cardinalitySchema = z.discriminatedUnion("kind", [ z.strictObject({ kind: z.literal("required") }), @@ -134,18 +128,7 @@ export const canonicalTraceJson = (value: unknown): string => { export const comparableTracePayload = ( value: unknown, ignoredFields: readonly string[] = [], -): unknown => { - if ( - ignoredFields.length === 0 || - typeof value !== "object" || - value === null || - Array.isArray(value) - ) - return value; - return Object.fromEntries( - Object.entries(value).filter(([name]) => !ignoredFields.includes(name)), - ); -}; +): unknown => comparableProcessObservationPayload(value, ignoredFields); export const processTraceCardinalityBounds = ( cardinality: ProcessTraceSpecification["events"][number]["cardinality"], diff --git a/tests/processObservation.test.ts b/tests/processObservation.test.ts new file mode 100644 index 000000000..92b53e9c2 --- /dev/null +++ b/tests/processObservation.test.ts @@ -0,0 +1,156 @@ +import { describe, expect, it } from "vitest"; + +import { EMPTY_PROCESS_CAPTURE_EXAMPLE } from "../src/contracts/processCaptureExample.js"; +import { + processCaptureSchema, + type ProcessCapture, +} from "../src/domain/processCapture.js"; +import { + createProcessObservation, + projectProcessObservation, +} from "../src/domain/processObservation.js"; + +const emptyCapture = processCaptureSchema.parse(EMPTY_PROCESS_CAPTURE_EXAMPLE); + +const capture: ProcessCapture = { + ...emptyCapture, + frames: [{ sequence: 0, at_ms: 1, data: "Ready" }], + rendered_frames: [ + { + sequence: 0, + at_ms: 2, + columns: 80, + rows: 24, + cursor_x: 0, + cursor_y: 0, + active_buffer: "normal", + lines: ["Ready"], + serialized_state: "Ready", + }, + ], + interaction_events: [ + { + sequence: 0, + scheduled_at_ms: 2, + dispatched_at_ms: 3, + type: "signal", + data: "SIGINT", + outcome: "dispatched", + }, + ], + process_samples: [ + { + at_ms: 4, + pid: 2, + parent_pid: 1, + command: "worker", + process_group_id: 1, + session_id: 1, + }, + ], + shim_events: [ + { + sequence: 0, + at_ms: 5, + command: "helper", + route_index: 0, + arguments: [], + working_directory: "/fixture", + outcome: "matched", + }, + ], + protocol_events: [ + { + sequence: 0, + at_ms: 6, + protocol: "http", + direction: "request", + method: "GET", + path: "/done", + data: "", + outcome: "matched", + }, + ], + replay_transitions: [ + { + sequence: 0, + at_ms: 7, + transition_id: "done", + state_before: "open", + state_after: "closed", + sensitive_aliases: [], + }, + ], +}; + +describe("process observation vocabulary", () => { + it("projects every capture collection without timestamp sorting", () => { + const locations = [ + { collection: "frames", index: 0, capture_order: 4 }, + { collection: "rendered_frames", index: 0, capture_order: 3 }, + { collection: "interaction_events", index: 0, capture_order: 2 }, + { collection: "lifecycle", index: 0, capture_order: 8 }, + { collection: "lifecycle", index: 1, capture_order: 9 }, + { collection: "process_samples", index: 0, capture_order: 1 }, + { collection: "filesystem_checkpoints", index: 0, capture_order: 0 }, + { collection: "shim_events", index: 0, capture_order: 5 }, + { collection: "protocol_events", index: 0, capture_order: 6 }, + { collection: "replay_transitions", index: 0, capture_order: 7 }, + ] as const; + const observations = locations.map((location) => + projectProcessObservation(capture, location), + ); + + expect(observations.map((value) => value?.source)).toEqual([ + "terminal_raw", + "terminal_rendered", + "interaction", + "lifecycle", + "lifecycle", + "process", + "filesystem", + "shim", + "http", + "replay_transition", + ]); + expect(observations.map((value) => value?.capture_order)).toEqual( + locations.map(({ capture_order }) => capture_order), + ); + expect(observations[5]).toMatchObject({ + event_id: "obs.process_samples.0", + source_sequence: 0, + captured_at_ms: 4, + subject_id: "process:2", + }); + expect(observations[3]?.captured_at_ms).toBeNull(); + expect(observations[4]?.captured_at_ms).toBeNull(); + }); + + it("uses the same constructor for live and post-hoc observations", () => { + const location = { + collection: "frames" as const, + index: 0, + capture_order: 4, + }; + const projected = projectProcessObservation(capture, location); + const live = createProcessObservation({ + source: "terminal_raw", + source_sequence: 0, + captured_at_ms: 1, + subject_id: null, + location, + payload: capture.frames[0], + }); + expect(live).toEqual(projected); + }); + + it("returns no observation for a missing raw record", () => { + expect( + projectProcessObservation(capture, { + collection: "frames", + index: 1, + capture_order: 10, + }), + ).toBeNull(); + }); +}); diff --git a/tests/processReactiveRuntime.test.ts b/tests/processReactiveRuntime.test.ts new file mode 100644 index 000000000..5a570075b --- /dev/null +++ b/tests/processReactiveRuntime.test.ts @@ -0,0 +1,642 @@ +import { describe, expect, it } from "vitest"; + +import { + createProcessObservation, + type ProcessObservationSource, +} from "../src/domain/processObservation.js"; +import { + commitProcessReactiveProposal, + createProcessReactiveSnapshot, + reduceProcessReactiveScenario, +} from "../src/domain/processReactiveRuntime.js"; +import { + PROCESS_REACTIVE_LIMITS, + processReactiveScenarioSchema, + type ProcessReactiveAction, + type ProcessReactiveScenario, +} from "../src/domain/processReactiveScenario.js"; + +const terminalTrigger = (literal = "Ready") => ({ + kind: "terminal_text" as const, + view: "decoded" as const, + encoding: "utf8" as const, + literal, + case_sensitive: true, + control_sequences: "include" as const, + occurrence: 1, + since: { kind: "scenario_start" as const }, + consume: false, +}); + +const scenarioWith = ( + transitions: readonly unknown[], +): ProcessReactiveScenario => + processReactiveScenarioSchema.parse({ + version: 1, + initial_state: "starting", + deadline_ms: 30_000, + states: [ + { + id: "starting", + max_visits: 4, + deadline_ms: 5_000, + on: transitions, + }, + ], + }); + +const finish = ( + id: string, + when: unknown, + priority = 100, + actions: readonly unknown[] = [], +) => ({ + id, + priority, + max_uses: 4, + when, + actions, + target: { kind: "finish", outcome: "passed" }, +}); + +const collectionFor = (source: ProcessObservationSource) => { + switch (source) { + case "terminal_raw": + return "frames" as const; + case "terminal_rendered": + return "rendered_frames" as const; + case "process": + return "process_samples" as const; + case "filesystem": + return "filesystem_checkpoints" as const; + case "shim": + return "shim_events" as const; + case "http": + case "websocket": + return "protocol_events" as const; + case "replay_transition": + return "replay_transitions" as const; + case "interaction": + return "interaction_events" as const; + case "lifecycle": + return "lifecycle" as const; + } +}; + +const observation = ( + source: ProcessObservationSource, + order: number, + payload: unknown, +) => + createProcessObservation({ + source, + source_sequence: order, + captured_at_ms: order, + subject_id: null, + location: { + collection: collectionFor(source), + index: order, + capture_order: order, + }, + payload, + }); + +const succeededEffect = (action: ProcessReactiveAction, order: number) => ({ + status: "succeeded" as const, + observation: + action.type === "checkpoint" + ? observation("filesystem", order, { name: action.name }) + : observation("interaction", order, { + type: + action.type === "send_input" + ? "input" + : action.type === "resize" + ? "resize" + : "signal", + data: + action.type === "send_input" + ? action.sensitive + ? `` + : action.data + : action.type === "resize" + ? `${String(action.columns)}x${String(action.rows)}` + : action.signal, + outcome: "dispatched", + }), +}); + +const offer = ( + scenario: ProcessReactiveScenario, + snapshot: ReturnType, + value: ReturnType, +) => { + const proposed = reduceProcessReactiveScenario(scenario, snapshot, { + kind: "observation", + observation: value, + }); + return proposed.kind === "proposal" + ? commitProcessReactiveProposal( + scenario, + proposed, + proposed.effects.map((effect, index) => ({ + ...succeededEffect( + effect, + (proposed.snapshot.observations.at(-1)?.capture_order ?? -1) + + index + + 1, + ), + })), + ) + : proposed; +}; + +describe("process reactive runtime", () => { + it("matches decoded terminal text across PTY chunks and emits declarative effects", () => { + const scenario = scenarioWith([ + finish("ready", terminalTrigger(), 100, [ + { type: "send_signal", target: { kind: "root" }, signal: "SIGINT" }, + { type: "checkpoint", name: "ready" }, + ]), + ]); + const first = offer( + scenario, + createProcessReactiveSnapshot(scenario), + observation("terminal_raw", 0, { + sequence: 0, + at_ms: 0, + data: "Re", + }), + ); + expect(first.kind).toBe("waiting"); + const second = offer( + scenario, + first.snapshot, + observation("terminal_raw", 1, { + sequence: 1, + at_ms: 1, + data: "ady\n", + }), + ); + expect(second.kind).toBe("transition"); + if (second.kind !== "transition") return; + expect(second.record).toEqual({ + sequence: 0, + transition_id: "ready", + state_before: "starting", + state_after: null, + outcome: "passed", + trigger_event_ids: ["obs.frames.0", "obs.frames.1"], + action_event_ids: [ + "obs.interaction_events.2", + "obs.filesystem_checkpoints.3", + ], + action_types: ["send_signal", "checkpoint"], + }); + }); + + it("reports only the minimal terminal chunk span containing the occurrence", () => { + const scenario = scenarioWith([finish("ready", terminalTrigger())]); + let snapshot = createProcessReactiveSnapshot(scenario); + for (const [order, data] of ["unrelated\n", "Re"].entries()) { + const decision = offer( + scenario, + snapshot, + observation("terminal_raw", order, { data }), + ); + expect(decision.kind).toBe("waiting"); + snapshot = decision.snapshot; + } + const matched = offer( + scenario, + snapshot, + observation("terminal_raw", 2, { data: "ady" }), + ); + expect(matched.kind).toBe("transition"); + if (matched.kind === "transition") + expect(matched.record.trigger_event_ids).toEqual([ + "obs.frames.1", + "obs.frames.2", + ]); + }); + + it("does not commit state, checkpoints, consumption, or history before effects succeed", () => { + const scenario = scenarioWith([ + finish("ready", { ...terminalTrigger(), consume: true }, 100, [ + { type: "checkpoint", name: "ready" }, + { type: "send_signal", target: { kind: "root" }, signal: "SIGINT" }, + ]), + ]); + const proposed = reduceProcessReactiveScenario( + scenario, + createProcessReactiveSnapshot(scenario), + { + kind: "observation", + observation: observation("terminal_raw", 0, { data: "Ready" }), + }, + ); + expect(proposed.kind).toBe("proposal"); + if (proposed.kind !== "proposal") return; + expect(proposed.snapshot).toMatchObject({ + status: "running", + checkpoints: [], + consumed_event_ids: [], + transitions: [], + }); + const rejected = commitProcessReactiveProposal(scenario, proposed, [ + { + status: "succeeded", + observation: observation("filesystem", 1, { name: "ready" }), + }, + { status: "rejected" }, + ]); + expect(rejected).toMatchObject({ + kind: "finished", + outcome: "action_rejected", + snapshot: { + checkpoints: [], + consumed_event_ids: [], + transitions: [], + }, + }); + expect( + commitProcessReactiveProposal(scenario, proposed, [ + { status: "target_lost" }, + { + status: "succeeded", + observation: observation("interaction", 2, { + type: "signal", + data: "SIGINT", + outcome: "dispatched", + }), + }, + ]), + ).toMatchObject({ kind: "finished", outcome: "target_lost" }); + + const committed = commitProcessReactiveProposal(scenario, proposed, [ + { + status: "succeeded", + observation: observation("filesystem", 1, { name: "ready" }), + }, + { + status: "succeeded", + observation: observation("interaction", 2, { + type: "signal", + data: "SIGINT", + outcome: "dispatched", + }), + }, + ]); + expect(committed.snapshot).toMatchObject({ + checkpoints: [ + { + name: "ready", + event_id: "obs.filesystem_checkpoints.1", + capture_order: 1, + }, + ], + }); + expect( + commitProcessReactiveProposal(scenario, proposed, [ + { + status: "succeeded", + observation: observation("filesystem", 1, { name: "other" }), + }, + { + status: "succeeded", + observation: observation("interaction", 2, { + type: "signal", + data: "SIGINT", + outcome: "dispatched", + }), + }, + ]), + ).toMatchObject({ kind: "finished", outcome: "action_rejected" }); + const changedScenario = scenarioWith([ + finish("ready", { ...terminalTrigger(), consume: true }, 100, [ + { type: "checkpoint", name: "changed" }, + { type: "send_signal", target: { kind: "root" }, signal: "SIGINT" }, + ]), + ]); + expect( + commitProcessReactiveProposal(changedScenario, proposed, [ + succeededEffect({ type: "checkpoint", name: "ready" }, 1), + succeededEffect( + { + type: "send_signal", + target: { kind: "root" }, + signal: "SIGINT", + }, + 2, + ), + ]), + ).toMatchObject({ kind: "finished", outcome: "target_lost" }); + expect( + commitProcessReactiveProposal(scenario, proposed, [ + { + status: "succeeded", + observation: observation("filesystem", 1, { name: "ready" }), + }, + { + status: "succeeded", + observation: { + ...observation("filesystem", 2, { name: "ready" }), + event_id: "obs.filesystem_checkpoints.1", + }, + }, + ]), + ).toMatchObject({ kind: "finished", outcome: "action_rejected" }); + }); + + it("preserves overflow when no any branch matches", () => { + const scenario = scenarioWith([ + finish("bounded_any", { + kind: "any", + triggers: [ + terminalTrigger("never"), + { + kind: "event", + source: "shim", + exact: { name: "also-never" }, + ignore_fields: [], + since: { kind: "scenario_start" }, + consume: false, + cardinality: { min: 1, max: 1 }, + }, + ], + }), + ]); + const decision = reduceProcessReactiveScenario( + scenario, + createProcessReactiveSnapshot(scenario), + { + kind: "observation", + observation: observation("terminal_raw", 0, { + data: "x".repeat(PROCESS_REACTIVE_LIMITS.evaluationWork + 1), + }), + }, + ); + expect(decision).toMatchObject({ + kind: "finished", + outcome: "capture_incomplete", + }); + }); + + it("fails closed at the deterministic per-reduction work budget", () => { + const transitions = Array.from( + { length: PROCESS_REACTIVE_LIMITS.predicates }, + (_, index) => + finish(`event_${String(index)}`, { + kind: "event", + source: "shim", + exact: { name: `expected_${String(index)}` }, + ignore_fields: ["sequence"], + since: { kind: "scenario_start" }, + consume: false, + cardinality: { min: 1, max: 1 }, + }), + ); + const scenario = scenarioWith(transitions); + const retained = Array.from({ length: 128 }, (_, order) => + observation("shim", order, { name: "other", sequence: order }), + ); + const snapshot = { + ...createProcessReactiveSnapshot(scenario), + observations: retained, + }; + const decision = offer( + scenario, + snapshot, + observation("shim", retained.length, { + name: "other", + sequence: retained.length, + }), + ); + expect(decision).toMatchObject({ + kind: "finished", + outcome: "capture_incomplete", + }); + }); + + it("selects a unique priority and reports equal-priority ambiguity", () => { + const preferred = scenarioWith([ + finish("fallback", terminalTrigger(), 20), + finish("preferred", terminalTrigger(), 10), + ]); + const selected = offer( + preferred, + createProcessReactiveSnapshot(preferred), + observation("terminal_raw", 0, { data: "Ready" }), + ); + expect(selected.kind).toBe("transition"); + if (selected.kind === "transition") + expect(selected.record.transition_id).toBe("preferred"); + + for (const transitions of [ + [finish("one", terminalTrigger()), finish("two", terminalTrigger())], + [finish("two", terminalTrigger()), finish("one", terminalTrigger())], + ]) { + const ambiguous = scenarioWith(transitions); + const decision = offer( + ambiguous, + createProcessReactiveSnapshot(ambiguous), + observation("terminal_raw", 0, { data: "Ready" }), + ); + expect(decision).toMatchObject({ + kind: "finished", + outcome: "ambiguous_match", + }); + } + }); + + it("evaluates ordered and repeated predicates without array-order arbitration", () => { + const event = (name: string) => ({ + kind: "event" as const, + source: "shim" as const, + exact: { name }, + ignore_fields: ["sequence" as const], + since: { kind: "scenario_start" as const }, + consume: true, + cardinality: { min: 1, max: 1 }, + }); + const scenario = scenarioWith([ + finish("sequence", { + kind: "sequence", + triggers: [ + event("start"), + { kind: "repeat", trigger: event("tick"), min: 2, max: 2 }, + event("done"), + ], + }), + ]); + let snapshot = createProcessReactiveSnapshot(scenario); + for (const [order, name] of ["start", "tick", "tick"].entries()) { + const decision = offer( + scenario, + snapshot, + observation("shim", order, { name, sequence: order }), + ); + expect(decision.kind).toBe("waiting"); + snapshot = decision.snapshot; + } + const decision = offer( + scenario, + snapshot, + observation("shim", 3, { name: "done", sequence: 3 }), + ); + expect(decision.kind).toBe("transition"); + if (decision.kind === "transition") + expect(decision.record.trigger_event_ids).toEqual([ + "obs.shim_events.0", + "obs.shim_events.1", + "obs.shim_events.2", + "obs.shim_events.3", + ]); + }); + + it("applies state-entry frontiers and explicit consumption", () => { + const secondTrigger = { + ...terminalTrigger(), + since: { kind: "state_entry" as const }, + consume: true, + }; + const scenario = processReactiveScenarioSchema.parse({ + version: 1, + initial_state: "one", + deadline_ms: 30_000, + states: [ + { + id: "one", + max_visits: 1, + deadline_ms: 5_000, + on: [ + { + ...finish("advance", terminalTrigger()), + target: { kind: "goto", state: "two" }, + }, + ], + }, + { + id: "two", + max_visits: 1, + deadline_ms: 5_000, + on: [finish("finish", secondTrigger)], + }, + ], + }); + const advanced = offer( + scenario, + createProcessReactiveSnapshot(scenario), + observation("terminal_raw", 0, { data: "Ready" }), + ); + expect(advanced.kind).toBe("transition"); + const waiting = offer( + scenario, + advanced.snapshot, + observation("shim", 1, { name: "unrelated" }), + ); + expect(waiting.kind).toBe("waiting"); + const finished = offer( + scenario, + waiting.snapshot, + observation("terminal_raw", 2, { data: "Ready" }), + ); + expect(finished.kind).toBe("transition"); + expect(finished.snapshot.consumed_event_ids).toEqual(["obs.frames.2"]); + }); + + it("rejects an exhausted state visit before proposing effects", () => { + const scenario = processReactiveScenarioSchema.parse({ + version: 1, + initial_state: "loop", + deadline_ms: 30_000, + states: [ + { + id: "loop", + max_visits: 1, + deadline_ms: 5_000, + on: [ + { + ...finish("again", terminalTrigger(), 100, [ + { + type: "send_signal", + target: { kind: "root" }, + signal: "SIGKILL", + }, + ]), + target: { kind: "goto", state: "loop" }, + }, + ], + }, + ], + }); + const decision = reduceProcessReactiveScenario( + scenario, + createProcessReactiveSnapshot(scenario), + { + kind: "observation", + observation: observation("terminal_raw", 0, { data: "Ready" }), + }, + ); + expect(decision).toMatchObject({ + kind: "finished", + outcome: "capture_incomplete", + }); + expect(decision).not.toHaveProperty("effects"); + }); + + it("classifies deadlines and non-monotonic observation input", () => { + const scenario = scenarioWith([finish("ready", terminalTrigger())]); + const initial = createProcessReactiveSnapshot(scenario); + expect( + reduceProcessReactiveScenario(scenario, initial, { + kind: "state_deadline", + state_id: "starting", + state_entry_capture_order: 0, + }), + ).toMatchObject({ kind: "finished", outcome: "predicate_timeout" }); + expect( + reduceProcessReactiveScenario(scenario, initial, { + kind: "state_deadline", + state_id: "previous", + state_entry_capture_order: 0, + }), + ).toMatchObject({ kind: "waiting", snapshot: initial }); + expect( + reduceProcessReactiveScenario(scenario, initial, { + kind: "scenario_deadline", + }), + ).toMatchObject({ kind: "finished", outcome: "scenario_deadline" }); + expect( + reduceProcessReactiveScenario(scenario, initial, { kind: "cancelled" }), + ).toMatchObject({ kind: "finished", outcome: "cancelled" }); + expect( + reduceProcessReactiveScenario(scenario, initial, { + kind: "cleanup_failed", + }), + ).toMatchObject({ kind: "finished", outcome: "cleanup_failed" }); + const passed = offer( + scenario, + initial, + observation("terminal_raw", 0, { data: "Ready" }), + ); + expect( + reduceProcessReactiveScenario(scenario, passed.snapshot, { + kind: "cleanup_failed", + }), + ).toMatchObject({ kind: "finished", outcome: "cleanup_failed" }); + const first = offer( + scenario, + initial, + observation("shim", 1, { name: "later" }), + ); + const invalid = offer( + scenario, + first.snapshot, + observation("shim", 0, { name: "earlier" }), + ); + expect(invalid).toMatchObject({ + kind: "finished", + outcome: "capture_incomplete", + }); + }); +}); diff --git a/tests/processReactiveScenario.test.ts b/tests/processReactiveScenario.test.ts new file mode 100644 index 000000000..3a37d09a6 --- /dev/null +++ b/tests/processReactiveScenario.test.ts @@ -0,0 +1,382 @@ +import { describe, expect, it } from "vitest"; + +import { + PROCESS_REACTIVE_LIMITS, + processReactiveScenarioSchema, +} from "../src/domain/processReactiveScenario.js"; + +const terminalTrigger = () => ({ + kind: "terminal_text" as const, + view: "decoded" as const, + encoding: "utf8" as const, + literal: "Ready", + case_sensitive: true, + control_sequences: "include" as const, + occurrence: 1, + since: { kind: "scenario_start" as const }, + consume: false, +}); + +const finishTransition = () => ({ + id: "finish", + priority: 100, + max_uses: 1, + when: terminalTrigger(), + actions: [{ type: "checkpoint" as const, name: "ready" }], + target: { kind: "finish" as const, outcome: "passed" as const }, +}); + +const baseScenario = () => ({ + version: 1 as const, + initial_state: "starting", + deadline_ms: 30_000, + states: [ + { + id: "starting", + max_visits: 1, + deadline_ms: 5_000, + on: [finishTransition()], + }, + ], +}); + +describe("process reactive scenario schema", () => { + it("parses the bounded v1 terminal scenario", () => { + expect(processReactiveScenarioSchema.parse(baseScenario())).toMatchObject({ + version: 1, + initial_state: "starting", + }); + expect(PROCESS_REACTIVE_LIMITS.triggerDepth).toBe(6); + }); + + it.each([ + ["unknown major version", { ...baseScenario(), version: 2 }], + [ + "missing state bound", + { + ...baseScenario(), + states: [ + { id: "starting", deadline_ms: 5_000, on: [finishTransition()] }, + ], + }, + ], + [ + "missing transition bound", + { + ...baseScenario(), + states: [ + { + id: "starting", + max_visits: 1, + deadline_ms: 5_000, + on: [{ ...finishTransition(), max_uses: undefined }], + }, + ], + }, + ], + [ + "unsupported trigger", + { + ...baseScenario(), + states: [ + { + id: "starting", + max_visits: 1, + deadline_ms: 5_000, + on: [{ ...finishTransition(), when: { kind: "callback" } }], + }, + ], + }, + ], + [ + "zero-minimum event absence predicate", + { + ...baseScenario(), + states: [ + { + ...baseScenario().states[0], + on: [ + { + ...finishTransition(), + when: { + kind: "event", + source: "shim", + exact: { name: "missing" }, + ignore_fields: [], + since: { kind: "scenario_start" }, + consume: false, + cardinality: { min: 0, max: 1 }, + }, + }, + ], + }, + ], + }, + ], + [ + "unsupported terminal interpretation", + { + ...baseScenario(), + states: [ + { + ...baseScenario().states[0], + on: [ + { + ...finishTransition(), + when: { + ...terminalTrigger(), + control_sequences: "strip_ansi", + }, + }, + ], + }, + ], + }, + ], + ])("rejects %s", (_name, input) => { + expect(processReactiveScenarioSchema.safeParse(input).success).toBe(false); + }); + + it("rejects unknown targets, unreachable states, and duplicate transition ids", () => { + const input = { + ...baseScenario(), + states: [ + { + id: "starting", + max_visits: 1, + deadline_ms: 5_000, + on: [ + { + ...finishTransition(), + target: { kind: "goto" as const, state: "missing" }, + }, + ], + }, + { + id: "unreachable", + max_visits: 1, + deadline_ms: 5_000, + on: [finishTransition()], + }, + ], + }; + const result = processReactiveScenarioSchema.safeParse(input); + expect(result.success).toBe(false); + if (!result.success) + expect(result.error.issues.map(({ message }) => message)).toEqual( + expect.arrayContaining([ + "transition target state is not declared", + "state is unreachable from the initial state", + "transition ids must be unique", + ]), + ); + }); + + it("accepts only explicitly bounded cycles", () => { + const cycle = { + version: 1 as const, + initial_state: "one", + deadline_ms: 10_000, + states: [ + { + id: "one", + max_visits: 2, + deadline_ms: 2_000, + on: [ + { + ...finishTransition(), + id: "to_two", + max_uses: 2, + actions: [], + target: { kind: "goto" as const, state: "two" }, + }, + ], + }, + { + id: "two", + max_visits: 2, + deadline_ms: 2_000, + on: [ + { + ...finishTransition(), + id: "to_one", + max_uses: 2, + actions: [], + target: { kind: "goto" as const, state: "one" }, + }, + ], + }, + ], + }; + expect(processReactiveScenarioSchema.safeParse(cycle).success).toBe(true); + const unbounded = { + ...cycle, + states: cycle.states.map((state) => ({ + ...state, + max_visits: undefined, + })), + }; + expect(processReactiveScenarioSchema.safeParse(unbounded).success).toBe( + false, + ); + }); + + it("rejects impossible checkpoint frontiers and volatile exact fields", () => { + const exact = { + kind: "event" as const, + source: "http" as const, + exact: { sequence: 0, path: "/ready" }, + ignore_fields: ["sequence" as const], + since: { kind: "checkpoint" as const, name: "missing" }, + consume: false, + cardinality: { min: 1, max: 1 }, + }; + const input = { + ...baseScenario(), + states: [ + { + ...baseScenario().states[0], + on: [{ ...finishTransition(), when: exact, actions: [] }], + }, + ], + }; + const result = processReactiveScenarioSchema.safeParse(input); + expect(result.success).toBe(false); + if (!result.success) + expect(result.error.issues.map(({ message }) => message)).toEqual( + expect.arrayContaining([ + "exact payload must omit every ignored field", + "checkpoint frontier is not definitely available before this transition", + ]), + ); + }); + + it("rejects checkpoint frontiers produced only by self or later transitions", () => { + const checkpointTrigger = (name: string) => ({ + ...terminalTrigger(), + since: { kind: "checkpoint" as const, name }, + }); + const input = { + version: 1 as const, + initial_state: "one", + deadline_ms: 10_000, + states: [ + { + id: "one", + max_visits: 1, + deadline_ms: 2_000, + on: [ + { + ...finishTransition(), + id: "self", + when: checkpointTrigger("self_checkpoint"), + actions: [ + { type: "checkpoint" as const, name: "self_checkpoint" }, + ], + target: { kind: "goto" as const, state: "two" }, + }, + ], + }, + { + id: "two", + max_visits: 1, + deadline_ms: 2_000, + on: [ + { + ...finishTransition(), + id: "later", + when: checkpointTrigger("later_checkpoint"), + actions: [ + { type: "checkpoint" as const, name: "later_checkpoint" }, + ], + }, + ], + }, + ], + }; + const result = processReactiveScenarioSchema.safeParse(input); + expect(result.success).toBe(false); + if (!result.success) + expect( + result.error.issues.filter(({ message }) => + message.includes("not definitely available"), + ), + ).toHaveLength(2); + }); + + it("rejects trigger trees beyond the declared depth", () => { + let trigger: unknown = terminalTrigger(); + for ( + let depth = 0; + depth < PROCESS_REACTIVE_LIMITS.triggerDepth; + depth += 1 + ) + trigger = { kind: "repeat", trigger, min: 1, max: 1 }; + const input = { + ...baseScenario(), + states: [ + { + ...baseScenario().states[0], + on: [{ ...finishTransition(), when: trigger }], + }, + ], + }; + const result = processReactiveScenarioSchema.safeParse(input); + expect(result.success).toBe(false); + if (!result.success) + expect(result.error.issues.map(({ message }) => message)).toContain( + "trigger nesting exceeds the v1 limit", + ); + + let hostile: unknown = terminalTrigger(); + for (let depth = 0; depth < 10_000; depth += 1) + hostile = { kind: "repeat", trigger: hostile, min: 1, max: 1 }; + const hostileInput = { + ...input, + states: [ + { + ...input.states[0], + on: [{ ...finishTransition(), when: hostile }], + }, + ], + }; + expect(() => + processReactiveScenarioSchema.safeParse(hostileInput), + ).not.toThrow(); + expect(processReactiveScenarioSchema.safeParse(hostileInput).success).toBe( + false, + ); + + let deepJson: unknown = "leaf"; + for (let depth = 0; depth < 10_000; depth += 1) + deepJson = { child: deepJson }; + const deepJsonInput = { + ...input, + states: [ + { + ...input.states[0], + on: [ + { + ...finishTransition(), + when: { + kind: "event", + source: "shim", + exact: deepJson, + ignore_fields: [], + since: { kind: "scenario_start" }, + consume: false, + cardinality: { min: 1, max: 1 }, + }, + }, + ], + }, + ], + }; + expect(() => + processReactiveScenarioSchema.safeParse(deepJsonInput), + ).not.toThrow(); + expect(processReactiveScenarioSchema.safeParse(deepJsonInput).success).toBe( + false, + ); + }); +});