Add causal runtime verification laws (#88)

Author: SandroMaglione

.changeset/bright-runtimes-verify.md

@@ -0,0 +1,7 @@
+---
+"@typeonce/effect-machine": minor
+---
+
+Add reusable runtime invariants, law-oriented causal command verification, and
+an explicit planner/runtime agreement check. Causal probe microsteps now retain
+stable public snapshots even when the optimized runtime reuses internal state.

README.md

@@ -811,6 +811,56 @@ bursts or retains outstanding mailbox work. Its model steps continue to use
 `formatRuntimeTranscript` names are deprecated aliases for the enqueue-oriented
 runner and formatter because their delivery semantics were not visible.
 
+### Runtime invariants and planner agreement
+
+Planner invariants and runtime invariants are deliberately separate. Runtime
+laws inspect causal command evidence, explicit asynchronous observations, and
+runtime status without requiring a duplicate reference model:
+
+```ts
+const invariant = MachineTest.runtimeInvariants(machine)
+const laws = [
+  invariant.snapshot("count never becomes negative", ({ snapshot }) => snapshot.state.value.count >= 0),
+  invariant.command(
+    "every accepted add is processed",
+    ({ command, result }) =>
+      command._tag !== "Send" || command.event._tag !== "Add" ||
+      result._tag === "SendProcessed"
+  )
+]
+
+const transcript = yield * MachineTest.verifyCausalCommands(
+  probe,
+  commands,
+  { invariants: laws }
+)
+```
+
+Use the existing `runCausalCommands` when a simplified application model
+provides exact expected results. Its returned transcript implements the same
+model-independent evidence interface, so reusable runtime laws compose with
+it directly:
+
+```ts
+const transcript = yield * MachineTest.runCausalCommands(probe, commands, model)
+
+yield * MachineTest.assertRuntimeInvariants(machine, transcript, laws)
+yield * MachineTest.assertPlannerRuntimeAgreement(machine, transcript)
+```
+
+`checkRuntimeInvariants` returns an aggregate report; `assertRuntimeInvariants`
+fails with every predicate and non-vacuity violation. Snapshot laws observe the
+initial and post-command snapshots by default. Select `"awaited"`, `"all"`, or
+`"final"` explicitly when a law targets observations retained by
+`probe.await.until` or only the final runtime snapshot.
+
+`assertPlannerRuntimeAgreement` is an explicit consistency check, not an
+application oracle. For each processed send it freshly plans from the receipt's
+`before` snapshot and compares handled/change flags, the public next snapshots,
+completion, command counts, emitted events, and public microstep evidence. It
+does not prove that the planner implements the intended business rules; use a
+reference model and runtime invariants for that.
+
 ## Current limits
 
 Declarative first-class guards are not part of the current API. Ordinary

docs/agent-guide.md

@@ -840,6 +840,37 @@ policies observe public snapshots but do not turn send acceptance into causal
 completion. Do not use the deprecated `runRuntimeCommands` name in new code;
 it is an alias for enqueue behavior and hides that important distinction.
 
+For semantic laws over live execution, bind runtime invariant constructors to
+the machine and use the law-oriented causal verifier:
+
+```ts
+const invariant = MachineTest.runtimeInvariants(machine)
+
+const laws = [
+  invariant.snapshot("balance never becomes negative", ({ snapshot }) =>
+    snapshot.state.value.balance >= 0
+  ),
+  invariant.command("stopped sends are rejected", ({ previous, result }) =>
+    previous?.result._tag !== "Stopped" || result._tag === "SendRejected"
+  )
+]
+
+yield* MachineTest.verifyCausalCommands(probe, commands, { invariants: laws })
+```
+
+Do not create a dummy model merely to run runtime laws. Continue to use
+`runCausalCommands` when an independent simplified model supplies exact
+expected results, then apply the same laws to its returned transcript with
+`assertRuntimeInvariants`. Conditional laws that must execute should declare
+`require.minObservations` so irrelevant generated commands cannot pass them
+vacuously.
+
+Use `assertPlannerRuntimeAgreement(machine, transcript)` only to check the
+managed runtime boundary against a fresh pure plan. It is not an independent
+business oracle and is intentionally an explicit operation rather than a
+generic conformance mode. Combine it with application runtime laws or a
+reference model when correctness of the expected behavior matters.
+
 ## Common compiler errors
 
 ### `initial` is not callable

src/internal/machine/executionPlan.ts

@@ -194,6 +194,20 @@ interface OwnedIndexedState {
   readonly completedOrder: ReadonlyArray<number>
 }
 
+const copyOwnedIndexedState = (state: OwnedIndexedState): OwnedIndexedState => ({
+  active: state.active.slice(),
+  activeLeaves: state.activeLeaves.slice(),
+  values: state.values.slice(),
+  completed: state.completed.slice(),
+  outputs: state.outputs.slice(),
+  completedOrder: state.completedOrder.slice()
+})
+
+const retainIndexedMicrostep = (
+  step: ExecutionMicrostep<OwnedIndexedState>,
+  retain: boolean
+): ExecutionMicrostep<OwnedIndexedState> => retain ? { ...step, next: copyOwnedIndexedState(step.next) } : step
+
 const updateOwnedIndexedValue = (
   state: OwnedIndexedState,
   index: number,
@@ -607,7 +621,8 @@ const planIndexedFlatState = (
   machine: Machine.Any,
   descriptor: IndexedExecutionDescriptor,
   configuration: OwnedIndexedState,
-  decoded: { readonly _tag: PropertyKey }
+  decoded: { readonly _tag: PropertyKey },
+  retainMicrosteps: boolean
 ): ExecutionMacrostep<OwnedIndexedState> => {
   let current = configuration
   let event: any = decoded
@@ -716,7 +731,7 @@ const planIndexedFlatState = (
         changed
       }
       current = next
-      ;(microsteps ??= []).push(step)
+      ;(microsteps ??= []).push(retainIndexedMicrostep(step, retainMicrosteps))
       if (transitionResult.commands.length > 0) {
         ;(commands ??= []).push(...transitionResult.commands)
       }
@@ -751,11 +766,12 @@ const planIndexedState = (
   machine: Machine.Any,
   descriptor: IndexedExecutionDescriptor,
   configuration: OwnedIndexedState,
-  input: unknown
+  input: unknown,
+  retainMicrosteps: boolean
 ): ExecutionMacrostep<OwnedIndexedState> => {
   const decoded = decodeEventSync(machine, input)
   if (descriptor.flat) {
-    return planIndexedFlatState(machine, descriptor, configuration, decoded)
+    return planIndexedFlatState(machine, descriptor, configuration, decoded, retainMicrosteps)
   }
   if (descriptor.finalIndices.some((index) => configuration.active[index] === 1)) {
     const active = activeConfigurationFromIndexedState(descriptor, configuration)
@@ -794,7 +810,7 @@ const planIndexedState = (
   const commands = [...first.commands]
   const raisedEvents = [...first.raisedEvents]
   const emittedEvents = [...first.emittedEvents]
-  const microsteps = [first]
+  const microsteps = [retainIndexedMicrostep(first, retainMicrosteps)]
   let raisedIndex = 0
   let iterations = 0
 
@@ -851,7 +867,7 @@ const planIndexedState = (
     commands.push(...step.commands)
     raisedEvents.push(...step.raisedEvents)
     emittedEvents.push(...step.emittedEvents)
-    microsteps.push(step)
+    microsteps.push(retainIndexedMicrostep(step, retainMicrosteps))
   }
 }
 
@@ -861,7 +877,8 @@ export interface CompiledExecutionPlan {
   readonly snapshot: (state: unknown) => Machine.Snapshot<any>
   readonly plan: (
     state: unknown,
-    event: unknown
+    event: unknown,
+    retainMicrosteps?: boolean
   ) => ExecutionMacrostep
   readonly initial?: (
     args: ReadonlyArray<unknown>
@@ -891,7 +908,8 @@ const makeIndexedExecutionPlan = (
   fromConfiguration: (configuration) => ownedIndexedStateFromActive(indexed, configuration),
   toConfiguration: (state) => activeConfigurationFromIndexedState(indexed, state as OwnedIndexedState),
   snapshot: (state) => snapshotFromIndexedState(indexed, state as OwnedIndexedState),
-  plan: (state, event) => planIndexedState(machine, indexed, state as OwnedIndexedState, event),
+  plan: (state, event, retainMicrosteps = false) =>
+    planIndexedState(machine, indexed, state as OwnedIndexedState, event, retainMicrosteps),
   initial: (args) => {
     const inputArgs = machine.input === undefined
       ? args

src/internal/machine/process.ts

@@ -48,6 +48,38 @@ const runSequentialDiscard = <E, R>(
     ? effects[0]!
     : Effect.all(effects, { discard: true })
 
+const acknowledgedPlan = (
+  planned: {
+    readonly next: unknown
+    readonly commands: ReadonlyArray<unknown>
+    readonly emittedEvents: ReadonlyArray<unknown>
+    readonly microsteps: ReadonlyArray<{
+      readonly next: unknown
+      readonly event: unknown
+      readonly transitions?: ReadonlyArray<unknown>
+      readonly commands: ReadonlyArray<unknown>
+      readonly raisedEvents: ReadonlyArray<unknown>
+      readonly emittedEvents: ReadonlyArray<unknown>
+      readonly exitPaths: ReadonlyArray<string>
+      readonly entryPaths: ReadonlyArray<string>
+      readonly changed: boolean
+    }>
+    readonly done: boolean
+    readonly output: unknown
+  },
+  snapshot: (state: unknown) => unknown
+): unknown => ({
+  next: snapshot(planned.next),
+  commands: planned.commands,
+  emittedEvents: planned.emittedEvents,
+  microsteps: planned.microsteps.map((microstep) => {
+    const { transitions: _, ...evidence } = microstep
+    return { ...evidence, next: snapshot(microstep.next) }
+  }),
+  done: planned.done,
+  output: planned.output
+})
+
 const invokeCapabilityCache = new WeakMap<Machine.Any, boolean>()
 
 const hasInvokeCapability = (machine: Machine.Any): boolean => {
@@ -98,7 +130,8 @@ const makeChildlessCompiledDrain = (
       try {
         planned = executionPlan.plan(
           configuration ?? executionPlan.fromConfiguration(Configuration.normalizeConfigurationSync(machine, current)),
-          event
+          event,
+          acknowledged
         )
       } catch (error) {
         return error instanceof InfiniteTransitionError || error instanceof MachineSchemaDecodeError
@@ -108,7 +141,9 @@ const makeChildlessCompiledDrain = (
       configuration = planned.next
       context.executionState = configuration
       if (planned.microsteps.length === 0) {
-        if (acknowledged) context.completeMessage({ before, plan: planned, after: current })
+        if (acknowledged) {
+          context.completeMessage({ before, plan: acknowledgedPlan(planned, executionPlan.snapshot), after: current })
+        }
         return loop
       }
 
@@ -129,7 +164,9 @@ const makeChildlessCompiledDrain = (
       }
       const continueAfterCommit = (): Effect.Effect<Option.Option<any>, any, any> => {
         const continued = Effect.suspend(() => {
-          if (acknowledged) context.completeMessage({ before, plan: planned, after: next })
+          if (acknowledged) {
+            context.completeMessage({ before, plan: acknowledgedPlan(planned, executionPlan.snapshot), after: next })
+          }
           return planned.done ? Effect.succeed(Option.some(planned.output)) : loop
         })
         return afterCommit === undefined ? continued : afterCommit.pipe(Effect.andThen(continued))
@@ -217,7 +254,8 @@ const makeInvokingCompiledDrain = (
         planned = executionPlan.plan(
           configuration ??
             executionPlan.fromConfiguration(Configuration.normalizeConfigurationSync(machine, current)),
-          event
+          event,
+          acknowledged
         )
       } catch (error) {
         return error instanceof InfiniteTransitionError || error instanceof MachineSchemaDecodeError
@@ -226,7 +264,9 @@ const makeInvokingCompiledDrain = (
       }
       configuration = planned.next
       if (planned.microsteps.length === 0) {
-        if (acknowledged) context.completeMessage({ before, plan: planned, after: current })
+        if (acknowledged) {
+          context.completeMessage({ before, plan: acknowledgedPlan(planned, executionPlan.snapshot), after: current })
+        }
         return loop
       }
 
@@ -285,7 +325,9 @@ const makeInvokingCompiledDrain = (
       }
       const continueAfterCommit = (): Effect.Effect<Option.Option<any>, any, any> => {
         const continued = Effect.suspend(() => {
-          if (acknowledged) context.completeMessage({ before, plan: planned, after: next })
+          if (acknowledged) {
+            context.completeMessage({ before, plan: acknowledgedPlan(planned, executionPlan.snapshot), after: next })
+          }
           return planned.done ? Effect.succeed(Option.some(planned.output)) : loop
         })
         return afterCommit.length === 0
@@ -522,7 +564,17 @@ const makeProcessLogic: <
                 }
               }
 
-              if (acknowledged) completeMessage({ before, plan: planned, after: current })
+              if (acknowledged) {
+                completeMessage({
+                  before,
+                  plan: acknowledgedPlan(
+                    planned,
+                    (state) =>
+                      Configuration.snapshotFromConfiguration(machine, state as Configuration.ActiveConfiguration)
+                  ),
+                  after: current
+                })
+              }
 
               if (terminal === undefined) {
                 pendingMessage = yield* pollMessage
@@ -638,7 +690,17 @@ const makeProcessLogic: <
                 }
               }
 
-              if (acknowledged) completeMessage({ before, plan: planned, after: current })
+              if (acknowledged) {
+                completeMessage({
+                  before,
+                  plan: acknowledgedPlan(
+                    planned,
+                    (state) =>
+                      Configuration.snapshotFromConfiguration(machine, state as Configuration.ActiveConfiguration)
+                  ),
+                  after: current
+                })
+              }
 
               if (terminal === undefined) {
                 pendingMessage = yield* pollMessage