Make snapshot persistence JSON-safe

.changeset/calm-machines-travel.md

@@ -0,0 +1,23 @@
+---
+"@typeonce/effect-machine": minor
+---
+
+Make `Machine.encodeSnapshot` return a canonical JSON representation or fail
+with `MachineSchemaEncodeError`. Encoded state values, completion outputs, and
+history values are now typed as `Schema.Json`; rich schema values use their
+canonical JSON codecs, while cycles and other non-JSON values fail at the
+machine boundary instead of causing a later serialization crash.
+Declared `Schema.Void` and `Schema.Undefined` completion outputs now use their
+canonical `null` encoding; an omitted output is reserved for final states that
+do not declare an output schema.
+
+`ClusterMachine.make` now requires JSON-encoded state, completion-output, and
+public input-event schemas. Keep process-local capabilities in services,
+adapters, or internal events, and give transported values an explicit JSON
+codec. Cluster snapshot encoding failures are reported as
+`SnapshotEncodeFailure` without advancing the checkpoint.
+
+`MachineTest.observedGraph` continues to support process-local state. Its node
+`encoded` field is now optional: portable snapshots retain their canonical JSON
+form, while non-portable snapshots use local structural identity and omit it.
+Snapshot encoding failures no longer appear in the operation's error channel.

README.md

@@ -620,6 +620,14 @@ const decoded = yield * Machine.decodeSnapshot(machine, encoded)
 const ref = yield * Machine.resume(machine, decoded)
 ```
 
+Decoded snapshots are local runtime values and may contain class instances or
+other process-local data. `encodeSnapshot` is the persistence boundary: it uses
+each declared schema's canonical JSON codec and succeeds only when every active
+state value, completion output, and history value is JSON. Rich values such as
+dates and bigints use their schema-defined JSON representation; cyclic or
+non-JSON values fail with `MachineSchemaEncodeError` instead of escaping to a
+later `JSON.stringify` crash.
+
 Resumption restores logical state, values, completion, and history metadata.
 It creates a fresh runtime: active invokes restart, timers restart at their
 full duration, and prior fibers, subscriptions, queues, and child runtimes are

docs/agent-guide.md

@@ -1276,6 +1276,22 @@ const decoded = yield* Machine.decodeSnapshot(machine, encoded)
 const ref = yield* Machine.resume(machine, decoded)
 ```
 
+`Machine.Snapshot` is the decoded, process-local representation. It may retain
+`Schema.Class` instances and capabilities that cannot cross a JSON boundary.
+`Machine.EncodedSnapshot` is different: successful `encodeSnapshot` calls
+guarantee canonical `Schema.Json` for every active value, completion output,
+and history value. Schema codecs convert rich values such as `Date`, `bigint`,
+and `undefined` to their declared JSON forms. Cycles, functions, symbols, and
+opaque values without a JSON representation fail with the typed
+`MachineSchemaEncodeError`; they never become a later `JSON.stringify` defect.
+
+Keep DOM nodes, open handles, services, and similar capabilities in an Effect
+service or UI adapter. Local events may carry those values when they stay inside
+one process. Cluster public input events, persisted state, and completion
+outputs are transport protocols and must instead declare JSON-compatible
+encoded forms; use an explicit transform or `Schema.toCodecJson` where the
+canonical codec is the intended wire contract.
+
 Pass only a decoded `Machine.Snapshot` to `resume`; encoded or arbitrary
 transport data belongs at `decodeSnapshot`. Resumption validates and normalizes
 the logical snapshot again, then publishes it as the fresh runtime's first

src/Machine.ts

@@ -1878,6 +1878,9 @@ export type RuntimeSnapshot<State, Error = never, Output = never> =
  * one stream may contain unrelated root, child-machine, and `Logic` protocols.
  * The record structure remains a closed discriminated union, while typed
  * application observation continues through `changes` and `emissions`.
+ * Records retain decoded local events and snapshots and are not themselves a
+ * stable JSON export format. Telemetry exporters must project process-local
+ * values into an explicit portable representation.
  *
  * @category models
  * @since 0.13.0
@@ -3963,36 +3966,39 @@ export declare namespace Machine {
    */
   export interface EncodedSnapshotState {
     readonly path: string
-    readonly value?: unknown
+    readonly value?: Schema.Json
   }
 
   /**
    * Encoded output for one completed state path in a normalized machine
-   * snapshot. An omitted output represents `undefined`.
+   * snapshot. An omitted output means the final state declares no output
+   * schema; a declared `Schema.Void` or `Schema.Undefined` output encodes as
+   * canonical JSON `null`.
    *
    * @category models
    * @since 0.4.0
    */
   export interface EncodedSnapshotCompletion {
     readonly path: string
-    readonly output?: unknown
+    readonly output?: Schema.Json
   }
 
   /** Encoded values and paths retained by one history pseudo-state. */
   export interface EncodedSnapshotHistoryEntry {
     readonly mode: "shallow" | "deep"
     readonly active: ReadonlyArray<string>
-    readonly values: Readonly<Record<string, unknown>>
+    readonly values: Readonly<Record<string, Schema.Json>>
   }
 
   /**
    * Normalized data representation of a machine snapshot.
    *
    * **Details**
    *
-   * Active state and completion values use the encoded representations of
-   * their declared schemas. Runtime process state such as children, fibers,
-   * scopes, queues, and subscriptions is not included.
+   * Active state and completion values use the canonical JSON representations
+   * derived from their declared schemas. A successfully encoded snapshot is
+   * safe to pass to JSON-backed persistence and transport. Runtime process state
+   * such as children, fibers, scopes, queues, and subscriptions is not included.
    *
    * @category models
    * @since 0.4.0
@@ -8018,18 +8024,21 @@ export const emittedEvents: {
  *
  * **Details**
  *
- * Each active state value and completed output is encoded with the schema
- * declared for its state path. The result contains no process-local runtime
- * state.
+ * Each active state value and completed output is encoded with the canonical
+ * JSON codec derived from the schema declared for its state path. Success
+ * guarantees that every state, output, and history value is `Schema.Json`.
+ * Non-JSON values, including cyclic process-local capabilities, fail with
+ * {@link MachineSchemaEncodeError} at their declared boundary.
  *
  * **Gotchas**
  *
  * The encoded snapshot does not contain the machine definition, machine
  * version, running children, invoked process state, services, or subscriptions.
  * Store machine identity and migration metadata alongside the result when the
- * snapshot crosses deployment versions. Schema encoding does not by itself
- * guarantee JSON-compatible values; schemas used with JSON-backed storage must
- * have JSON-compatible encoded representations.
+ * snapshot crosses deployment versions. Opaque declarations without a JSON
+ * codec can encode only when their current value is already JSON-compatible.
+ * Define an explicit JSON codec or keep process-local capabilities outside the
+ * logical snapshot.
  *
  * **Example**
  *

src/internal/machine/cluster.ts

@@ -72,6 +72,7 @@ export const RejectionReason = Schema.Literals([
   "InvalidCheckpoint",
   "UnsupportedProcessLocal",
   "TransitionFailure",
+  "SnapshotEncodeFailure",
   "PersistenceFailure",
   "EmissionFailure"
 ])
@@ -341,7 +342,9 @@ export const make = <
             emitted.push(...planned.emittedEvents as any)
           }
 
-          const encoded = yield* internalMachine.encodeSnapshot(rootMachine, current)
+          const encoded = yield* internalMachine.encodeSnapshot(rootMachine, current).pipe(
+            Effect.mapError((error) => reject("SnapshotEncodeFailure", String(error.cause)))
+          )
           if (emitted.length > 0 && layerOptions?.enqueue === undefined) {
             return yield* fail("EmissionFailure", "No durable enqueue handler was configured")
           }

src/internal/machine/serialization.ts

@@ -31,22 +31,48 @@ const EncodedSnapshotSchema = Schema.Struct({
   _tag: Schema.Literal("MachineSnapshot"),
   active: Schema.Array(Schema.Struct({
     path: Schema.String,
-    value: Schema.optional(Schema.Unknown)
+    value: Schema.optionalKey(Schema.Json)
   })),
-  completed: Schema.optional(Schema.Array(Schema.Struct({
+  completed: Schema.optionalKey(Schema.Array(Schema.Struct({
     path: Schema.String,
-    output: Schema.optional(Schema.Unknown)
+    output: Schema.optionalKey(Schema.Json)
   }))),
-  history: Schema.optional(Schema.Record(
+  history: Schema.optionalKey(Schema.Record(
     Schema.String,
     Schema.Struct({
       mode: Schema.Literals(["shallow", "deep"]),
       active: Schema.Array(Schema.String),
-      values: Schema.Record(Schema.String, Schema.Unknown)
+      values: Schema.Record(Schema.String, Schema.Json)
     })
   ))
 })
 
+const jsonCodecCache = new WeakMap<object, Schema.Top>()
+
+const getJsonCodec = (schema: Schema.Top): Schema.Top => {
+  const key = schema as object
+  const cached = jsonCodecCache.get(key)
+  if (cached !== undefined) return cached
+  const codec = Schema.toCodecJson(schema)
+  jsonCodecCache.set(key, codec)
+  return codec
+}
+
+const encodeError = (
+  machine: Machine.Any,
+  options: {
+    readonly boundary: "state" | "output" | "history"
+    readonly state: string
+  },
+  cause: Schema.SchemaError | Cause.Cause<unknown>
+): MachineSchemaEncodeError =>
+  new MachineSchemaEncodeError({
+    machineId: machine.id,
+    boundary: options.boundary,
+    state: options.state,
+    cause
+  })
+
 const encodeBoundary = (
   machine: Machine.Any,
   schema: Schema.Top,
@@ -55,16 +81,14 @@ const encodeBoundary = (
     readonly boundary: "state" | "output" | "history"
     readonly state: string
   }
-): Effect.Effect<unknown, MachineSchemaEncodeError, unknown> =>
-  Schema.encodeUnknownEffect(schema)(value).pipe(
-    Effect.mapError((cause) =>
-      new MachineSchemaEncodeError({
-        machineId: machine.id,
-        boundary: options.boundary,
-        state: options.state,
-        cause
-      })
-    )
+): Effect.Effect<Schema.Json, MachineSchemaEncodeError, unknown> =>
+  Effect.try({
+    try: () => getJsonCodec(schema),
+    catch: (cause) => encodeError(machine, options, Cause.die(cause))
+  }).pipe(
+    Effect.flatMap((codec) => Schema.encodeUnknownEffect(codec)(value)),
+    Effect.flatMap(Schema.decodeUnknownEffect(Schema.Json)),
+    Effect.mapError((cause) => cause instanceof MachineSchemaEncodeError ? cause : encodeError(machine, options, cause))
   )
 
 const decodeEncodedBoundary = (
@@ -76,22 +100,34 @@ const decodeEncodedBoundary = (
     readonly state: string
   }
 ): Effect.Effect<unknown, MachineSchemaDecodeError, unknown> =>
-  Schema.decodeUnknownEffect(schema)(value).pipe(
-    Effect.mapError((cause) =>
+  Effect.try({
+    try: () => getJsonCodec(schema),
+    catch: (cause) =>
       new MachineSchemaDecodeError({
         machineId: machine.id,
         boundary: options.boundary,
         state: options.state,
-        cause
+        cause: Cause.die(cause)
       })
+  }).pipe(
+    Effect.flatMap((codec) => Schema.decodeUnknownEffect(codec)(value)),
+    Effect.mapError((cause) =>
+      cause instanceof MachineSchemaDecodeError ?
+        cause :
+        new MachineSchemaDecodeError({
+          machineId: machine.id,
+          boundary: options.boundary,
+          state: options.state,
+          cause
+        })
     )
   )
 
 const getCompletionSchema = (
   machine: Machine.Any,
   configuration: ActiveConfiguration,
   path: string
-): Schema.Top => {
+): Schema.Top | undefined => {
   const node = getNode(machine, path)
   if (node.type === "compound") {
     const child = getActiveChildPath(machine, configuration, path)
@@ -100,7 +136,7 @@ const getCompletionSchema = (
     }
     return getCompletionSchema(machine, configuration, child)
   }
-  return node.output ?? Schema.Void
+  return node.output
 }
 
 /** Defensively validates and normalizes an in-memory logical snapshot. Unlike
@@ -132,9 +168,17 @@ export const normalizeSnapshotEffect = <const States extends Machine.StateSchema
         throw new Error(`Machine snapshot contains invalid completion "${path}"`)
       }
       completionPaths.add(path)
+      const schema = getCompletionSchema(machine, configuration, path)
+      if (schema === undefined) {
+        if (completion.output !== undefined) {
+          throw new Error(`Machine snapshot contains an output for state "${path}" without an output schema`)
+        }
+        outputs.set(path, undefined)
+        continue
+      }
       outputs.set(
         path,
-        yield* decodeBoundary(machine, getCompletionSchema(machine, configuration, path), completion.output, {
+        yield* decodeBoundary(machine, schema, completion.output, {
           boundary: "output",
           state: path
         })
@@ -242,9 +286,17 @@ export const encodeSnapshot = (
       if (!configuration.active.has(path) || !isActiveFinalNode(machine, configuration, path)) {
         throw new Error(`Machine encoded snapshot contains invalid completion "${path}"`)
       }
+      const schema = getCompletionSchema(machine, configuration, path)
+      if (schema === undefined) {
+        if (output !== undefined) {
+          throw new Error(`Machine snapshot contains an output for state "${path}" without an output schema`)
+        }
+        completed.push({ path })
+        continue
+      }
       const encodedOutput = yield* encodeBoundary(
         machine,
-        getCompletionSchema(machine, configuration, path),
+        schema,
         output,
         {
           boundary: "output",
@@ -289,7 +341,7 @@ export const encodeSnapshot = (
           })
         )
       }
-      const encodedValues: Record<string, unknown> = {}
+      const encodedValues: Record<string, Schema.Json> = {}
       for (const path of record.active) {
         const stateNode = machine.stateNodes.byPath.get(path)
         if (
@@ -338,12 +390,21 @@ export const encodeSnapshot = (
       }
     }
 
-    return {
+    const encoded: Machine.EncodedSnapshot = {
       _tag: "MachineSnapshot" as const,
       active,
       ...(completed.length === 0 ? {} : { completed }),
       ...(Object.keys(history).length === 0 ? {} : { history })
     }
+    return yield* Schema.decodeUnknownEffect(EncodedSnapshotSchema)(encoded).pipe(
+      Effect.mapError((cause) =>
+        new MachineSchemaEncodeError({
+          machineId: machine.id,
+          boundary: "configuration",
+          cause
+        })
+      )
+    )
   }).pipe(Effect.catchCause((cause) => failEncodeCause(machine, cause)))
 
 export const decodeSnapshot = (
@@ -501,11 +562,25 @@ export const decodeSnapshot = (
         throw new Error(`Machine encoded snapshot contains invalid completion "${completion.path}"`)
       }
       completionPaths.add(completion.path)
+      const schema = getCompletionSchema(machine, configuration, completion.path)
+      const hasOutput = Object.prototype.hasOwnProperty.call(completion, "output")
+      if (schema === undefined) {
+        if (hasOutput) {
+          throw new Error(
+            `Machine encoded snapshot contains an output for state "${completion.path}" without an output schema`
+          )
+        }
+        completions.push({ path: completion.path, output: undefined })
+        continue
+      }
+      if (!hasOutput) {
+        throw new Error(`Machine encoded snapshot omits output for state "${completion.path}"`)
+      }
       completions.push({
         path: completion.path,
         output: yield* decodeEncodedBoundary(
           machine,
-          getCompletionSchema(machine, configuration, completion.path),
+          schema,
           completion.output,
           {
             boundary: "output",