From 9aa19d21dc2c7e69a25d2431da92c90d6cc6bddd Mon Sep 17 00:00:00 2001 From: SandroMaglione Date: Wed, 19 Aug 2026 19:19:23 +0200 Subject: [PATCH] Make snapshot persistence JSON-safe --- .changeset/calm-machines-travel.md | 23 +++ README.md | 8 + docs/agent-guide.md | 16 ++ src/Machine.ts | 35 +++-- src/internal/machine/cluster.ts | 5 +- src/internal/machine/serialization.ts | 125 ++++++++++++---- src/internal/testing/machine/verification.ts | 25 ++-- src/testing/MachineTest.ts | 15 +- src/unstable/cluster/ClusterMachine.ts | 52 ++++++- test/machine/Machine.test.ts | 25 ++++ test/machine/SnapshotCodecAdversarial.test.ts | 139 +++++++++++++++++- test/testing/Coverage.test.ts | 15 +- test/unstable/cluster/ClusterMachine.test.ts | 40 +++++ typetest/testing/Coverage.tst.ts | 3 +- .../unstable/cluster/ClusterMachine.tst.ts | 121 +++++++++++++++ 15 files changed, 579 insertions(+), 68 deletions(-) create mode 100644 .changeset/calm-machines-travel.md diff --git a/.changeset/calm-machines-travel.md b/.changeset/calm-machines-travel.md new file mode 100644 index 0000000..1a8baba --- /dev/null +++ b/.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. diff --git a/README.md b/README.md index 3636f4d..59b9bad 100644 --- a/README.md +++ b/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 diff --git a/docs/agent-guide.md b/docs/agent-guide.md index dfd2466..8bf50bf 100644 --- a/docs/agent-guide.md +++ b/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 diff --git a/src/Machine.ts b/src/Machine.ts index 196d28d..5b13753 100644 --- a/src/Machine.ts +++ b/src/Machine.ts @@ -1878,6 +1878,9 @@ export type RuntimeSnapshot = * 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,26 +3966,28 @@ 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 - readonly values: Readonly> + readonly values: Readonly> } /** @@ -3990,9 +3995,10 @@ export declare namespace Machine { * * **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** * diff --git a/src/internal/machine/cluster.ts b/src/internal/machine/cluster.ts index 9dc33e6..4965f2d 100644 --- a/src/internal/machine/cluster.ts +++ b/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") } diff --git a/src/internal/machine/serialization.ts b/src/internal/machine/serialization.ts index 8248cd0..9e99755 100644 --- a/src/internal/machine/serialization.ts +++ b/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() + +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 +): 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 => - Schema.encodeUnknownEffect(schema)(value).pipe( - Effect.mapError((cause) => - new MachineSchemaEncodeError({ - machineId: machine.id, - boundary: options.boundary, - state: options.state, - cause - }) - ) +): Effect.Effect => + 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,14 +100,26 @@ const decodeEncodedBoundary = ( readonly state: string } ): Effect.Effect => - 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 + }) ) ) @@ -91,7 +127,7 @@ 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 encodedValues: Record = {} 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", diff --git a/src/internal/testing/machine/verification.ts b/src/internal/testing/machine/verification.ts index 5a42355..689aca9 100644 --- a/src/internal/testing/machine/verification.ts +++ b/src/internal/testing/machine/verification.ts @@ -772,7 +772,7 @@ export const observedGraph: ( traceOrTraces: Trace | ReadonlyArray> ) => Effect.Effect< ObservedGraph, - Machine.MachineSchemaEncodeError, + never, Machine.Machine.SnapshotEncodingServices> > = Effect.fnUntraced(function*( machine: M, @@ -837,19 +837,24 @@ export const observedGraph: ( } }) - const encoded = yield* Effect.forEach( + const representations = yield* Effect.forEach( occurrences, ({ snapshot }) => - (Machine.encodeSnapshot as any)(machine, snapshot) as Effect.Effect< + ((Machine.encodeSnapshot as any)(machine, snapshot) as Effect.Effect< Machine.Machine.EncodedSnapshot, Machine.MachineSchemaEncodeError, Machine.Machine.SnapshotEncodingServices> - > + >).pipe( + Effect.match({ + onFailure: () => ({ identity: snapshot, encoded: undefined }), + onSuccess: (encoded) => ({ identity: encoded, encoded }) + }) + ) ) - const encodedIdentity = makeStructuralIdentityIndex() - const occurrenceIds = encoded.map(encodedIdentity) + const representationIdentity = makeStructuralIdentityIndex() + const occurrenceIds = representations.map(({ identity }) => representationIdentity(identity)) const grouped = new Map> startup: number event: number @@ -862,7 +867,7 @@ export const observedGraph: ( grouped.set( id, group = { - encoded: encoded[index]!, + encoded: representations[index]!.encoded, snapshot: occurrence.snapshot, startup: 0, event: 0, @@ -879,8 +884,8 @@ export const observedGraph: ( const node = Graph.addNode(mutable, { id, snapshot: group.snapshot, - encoded: group.encoded, - configuration: group.encoded.active.map(({ path }) => path) as unknown as ReadonlyArray>, + ...(group.encoded === undefined ? {} : { encoded: group.encoded }), + configuration: rawConfigurationPaths(machine, group.snapshot) as ReadonlyArray>, observations: { total: group.startup + group.event + group.microstep, startup: group.startup, diff --git a/src/testing/MachineTest.ts b/src/testing/MachineTest.ts index 0995387..d1532ef 100644 --- a/src/testing/MachineTest.ts +++ b/src/testing/MachineTest.ts @@ -1935,7 +1935,7 @@ export interface ObservedGraphNodeObservations { } /** - * One full encoded logical snapshot stored in the observed Effect graph. + * One decoded logical snapshot stored in the observed Effect graph. * * @category models * @since 0.4.0 @@ -1943,7 +1943,8 @@ export interface ObservedGraphNodeObservations { export interface ObservedGraphNode { readonly id: string readonly snapshot: Machine.Machine.Snapshot> - readonly encoded: Machine.Machine.EncodedSnapshot + /** Canonical JSON representation when this local snapshot is portable. */ + readonly encoded?: Machine.Machine.EncodedSnapshot readonly configuration: ReadonlyArray> readonly observations: ObservedGraphNodeObservations } @@ -2004,9 +2005,11 @@ export interface ObservedGraph { /** * Converts concrete planner traces into an observed logical-state graph. - * Nodes are deduplicated by the public snapshot encoding and every edge is a - * concrete startup or public-event macrostep. This intentionally does not - * claim to be a static or exhaustive graph of the machine. + * Portable nodes are deduplicated by the public snapshot encoding. If encoding + * fails, process-local values are instead compared by cycle-safe structural + * identity and `ObservedGraphNode.encoded` is omitted. Every edge is a concrete + * startup or public-event macrostep. This intentionally does not claim to be a + * static or exhaustive graph of the machine. * * @category verification * @since 0.4.0 @@ -2016,7 +2019,7 @@ export const observedGraph: ( traceOrTraces: Trace | ReadonlyArray> ) => Effect.Effect< ObservedGraph, - Machine.MachineSchemaEncodeError, + never, Machine.Machine.SnapshotEncodingServices> > = internal.observedGraph diff --git a/src/unstable/cluster/ClusterMachine.ts b/src/unstable/cluster/ClusterMachine.ts index ae756ce..0cede00 100644 --- a/src/unstable/cluster/ClusterMachine.ts +++ b/src/unstable/cluster/ClusterMachine.ts @@ -153,6 +153,7 @@ export const RejectionReason = Schema.Literals([ "InvalidCheckpoint", "UnsupportedProcessLocal", "TransitionFailure", + "SnapshotEncodeFailure", "PersistenceFailure", "EmissionFailure" ]) @@ -246,6 +247,51 @@ type MachineServices = type IsAny = 0 extends (1 & A) ? true : false +type IsNever = [A] extends [never] ? true : false + +type IsUnknown = IsAny extends true ? false : unknown extends A ? true : false + +type IsJsonEncoded = IsAny extends true ? false + : IsNever extends true ? false + : IsUnknown extends true ? false + : [S["Encoded"]] extends [Schema.Json] ? true + : false + +type NonJsonState = Machine.Machine.ValuedStateIdentifier extends + infer StateId + ? StateId extends Machine.Machine.ValuedStateIdentifier ? + IsJsonEncoded> extends true ? never : StateId + : never + : never + +type NonJsonOutput = Machine.Machine.DeclaredOutputState extends + infer StateId + ? StateId extends Machine.Machine.DeclaredOutputState ? + Machine.Machine.NodeByIdentifier extends { + readonly output: infer Output extends Schema.Top + } ? IsJsonEncoded extends true ? never : StateId + : never + : never + : never + +type NonJsonInputEvent> = { + readonly [Index in keyof Events]: Events[Index] extends infer EventSchema extends Machine.Machine.TaggedSchema + ? IsJsonEncoded extends true ? never : Machine.Machine.TagOf + : never +}[number] + +type EnsureJsonEncoded< + States extends Machine.Machine.StateSchemas, + InputEvents extends ReadonlyArray +> = [NonJsonState | NonJsonOutput | NonJsonInputEvent] extends [never] ? unknown + : { + readonly "~effect/ClusterMachine/NonJsonEncoded": { + readonly states: NonJsonState + readonly outputs: NonJsonOutput + readonly inputEvents: NonJsonInputEvent + } + } + type ExcludeCompatibleRuntime = Requirements extends Machine.Runtime.Requirement< infer RequiredEvents, infer RequiredEmits @@ -303,6 +349,9 @@ export const layerMemory: Layer.Layer = internal.layerMemory * rejected. Planning-time raised events remain part of the current macrostep. * Arbitrary action effects may run again after a crash before checkpoint * commit, so the bridge does not provide exactly-once external effects. + * State values, completion outputs, and public input events must declare + * JSON-compatible encoded representations. Local and internal event protocols + * remain unrestricted because they do not cross the Cluster boundary. * * **Example** * @@ -364,7 +413,8 @@ export const make: < ParentEvents > & EnsureExecutable - & Machine.Machine.RootCompatible, + & Machine.Machine.RootCompatible + & EnsureJsonEncoded, options: { readonly version: string }, diff --git a/test/machine/Machine.test.ts b/test/machine/Machine.test.ts index 34b02f8..7b13082 100644 --- a/test/machine/Machine.test.ts +++ b/test/machine/Machine.test.ts @@ -1933,6 +1933,31 @@ describe("Machine", () => { assert.deepStrictEqual(decoded.completed, [{ path: "all.left" as const, output: undefined }]) })) + it.effect("distinguishes an omitted output schema from an explicit void codec", () => + Effect.gen(function*() { + const states = Machine.states({ + done: { + schema: ParallelLeftDone, + type: "final", + output: Schema.Void + } + }) + const machine = Machine.make({ + states: states.states, + events: Machine.events(), + initial: (to) => to.done().resolve(({ target }) => target(new ParallelLeftDone({ id: "done" }))) + }).handle({ + done: { output: () => undefined } + }) + const planned = yield* Machine.planInitial(machine) + + const encoded = yield* Machine.encodeSnapshot(machine, planned.state) + const decoded = yield* Machine.decodeSnapshot(machine, JSON.parse(JSON.stringify(encoded))) + + assert.deepStrictEqual(encoded.completed, [{ path: "done" as const, output: null }]) + assert.deepStrictEqual(decoded.completed, [{ path: "done" as const, output: undefined }]) + })) + it.effect("rejects state values that cannot be encoded", () => Effect.gen(function*() { const states = Machine.states({ NonEmptyIdle }) diff --git a/test/machine/SnapshotCodecAdversarial.test.ts b/test/machine/SnapshotCodecAdversarial.test.ts index 270d350..eaf834a 100644 --- a/test/machine/SnapshotCodecAdversarial.test.ts +++ b/test/machine/SnapshotCodecAdversarial.test.ts @@ -103,7 +103,8 @@ class Editor extends Schema.TaggedClass("CodecEditor")("CodecEditor", { document: Schema.NonEmptyString }) {} class Editing extends Schema.TaggedClass("CodecEditing")("CodecEditing", { - contents: Schema.String + contents: Schema.String, + payload: Schema.Any }) {} class Preview extends Schema.TaggedClass("CodecPreview")("CodecPreview", { page: Schema.Number @@ -155,12 +156,58 @@ const historySnapshot = () => values: { Workspace: new Workspace({ revision: 3 }), "Workspace.Editor": new Editor({ document: "doc-1" }), - "Workspace.Editor.editing": new Editing({ contents: "hello" }) + "Workspace.Editor.editing": new Editing({ contents: "hello", payload: null }) } } } }) as Machine.Machine.Snapshot +class RichState extends Schema.TaggedClass("CodecRichState")("CodecRichState", { + createdAt: Schema.Date, + sequence: Schema.BigInt, + missing: Schema.Undefined +}) {} + +const RichStates = Machine.states({ RichState }) +const richMachine = Machine.make({ + id: "codec-rich", + states: RichStates.states, + events: Machine.events(), + initial: (to) => + to.RichState().resolve(({ target }) => + target(new RichState({ createdAt: new Date("2026-08-19T12:00:00.000Z"), sequence: 42n, missing: undefined })) + ) +}) + +interface OpaqueState { + readonly _tag: "CodecOpaqueState" + readonly resource: object +} + +const OpaqueState = Schema.declare((input): input is OpaqueState => + typeof input === "object" && input !== null && "_tag" in input && input._tag === "CodecOpaqueState" && + "resource" in input && typeof input.resource === "object" && input.resource !== null +) + +const OpaqueStates = Machine.states({ OpaqueState }) +const opaqueMachine = Machine.make({ + id: "codec-opaque", + states: OpaqueStates.states, + events: Machine.events(), + initial: (to) => to.OpaqueState().resolve(({ target }) => target({ _tag: "CodecOpaqueState", resource: {} })) +}) + +class OutputDone extends Schema.TaggedClass("CodecOutputDone")("CodecOutputDone", {}) {} +const OutputStates = Machine.states({ + OutputDone: { schema: OutputDone, type: "final", output: Schema.Any } +}) +const outputMachine = Machine.make({ + id: "codec-output", + states: OutputStates.states, + events: Machine.events(), + initial: (to) => to.OutputDone().resolve(({ target }) => target(new OutputDone({}))) +}) + const expectEncodeFailure = Effect.fnUntraced(function*(snapshot: unknown, boundary?: string) { const error = yield* Machine.encodeSnapshot( topologyMachine, @@ -218,6 +265,94 @@ describe("snapshot codec adversarial boundaries", () => { ]) })) + it.effect("uses canonical JSON codecs for supported rich state values", () => + Effect.gen(function*() { + const snapshot = { + path: "RichState" as const, + value: new RichState({ + createdAt: new Date("2026-08-19T12:00:00.000Z"), + sequence: 42n, + missing: undefined + }) + } + const encoded = yield* Machine.encodeSnapshot(richMachine, snapshot) + + assert.deepStrictEqual(encoded.active, [{ + path: "RichState", + value: { + _tag: "CodecRichState", + createdAt: "2026-08-19T12:00:00.000Z", + sequence: "42", + missing: null + } + }]) + assert.doesNotThrow(() => JSON.stringify(encoded)) + + const decoded = yield* Machine.decodeSnapshot(richMachine, JSON.parse(JSON.stringify(encoded))) + assert.instanceOf(decoded.value, RichState) + assert.instanceOf(decoded.value.createdAt, Date) + assert.strictEqual(decoded.value.sequence, 42n) + assert.strictEqual(decoded.value.missing, undefined) + })) + + it.effect("rejects cyclic state, completion, and history values with typed boundary failures", () => + Effect.gen(function*() { + const cyclic: Record = {} + cyclic.self = cyclic + + const assertFailure = (exit: Exit.Exit, boundary: "state" | "output" | "history") => { + assert(Exit.isFailure(exit)) + assert.strictEqual(Cause.hasDies(exit.cause), false) + const error = Cause.findErrorOption(exit.cause) + assert(Option.isSome(error)) + assert.instanceOf(error.value, Machine.MachineSchemaEncodeError) + assert.strictEqual(error.value.boundary, boundary) + } + + assertFailure( + yield* Effect.exit(Machine.encodeSnapshot(opaqueMachine, { + path: "OpaqueState", + value: { _tag: "CodecOpaqueState", resource: cyclic } + })), + "state" + ) + assertFailure( + yield* Effect.exit(Machine.encodeSnapshot(outputMachine, { + path: "OutputDone", + value: new OutputDone({}), + completed: [{ path: "OutputDone", output: cyclic }] + })), + "output" + ) + for (const output of [undefined, Symbol("local"), () => undefined]) { + assertFailure( + yield* Effect.exit(Machine.encodeSnapshot(outputMachine, { + path: "OutputDone", + value: new OutputDone({}), + completed: [{ path: "OutputDone", output }] + })), + "output" + ) + } + const snapshot = historySnapshot() + assertFailure( + yield* Effect.exit(Machine.encodeSnapshot(historyMachine, { + ...snapshot, + history: { + ...snapshot.history, + "Workspace.exact": { + ...snapshot.history!["Workspace.exact"]!, + values: { + ...snapshot.history!["Workspace.exact"]!.values, + "Workspace.Editor.editing": new Editing({ contents: "hello", payload: cyclic }) + } + } + } + })), + "history" + ) + })) + it.effect("round-trips shallow and deep history records through JSON", () => Effect.gen(function*() { const snapshot = historySnapshot() diff --git a/test/testing/Coverage.test.ts b/test/testing/Coverage.test.ts index 4663d9b..e3b3960 100644 --- a/test/testing/Coverage.test.ts +++ b/test/testing/Coverage.test.ts @@ -317,7 +317,7 @@ describe("MachineTest trace coverage", () => { }) describe("MachineTest observed graph", () => { - it.effect("deduplicates encoded snapshots while preserving concrete startup and event edges", () => + it.effect("deduplicates portable snapshots while preserving concrete startup and event edges", () => Effect.gen(function*() { const first = yield* MachineTest.run(counterMachine, { events: [new Add({ amount: 1 })] @@ -338,12 +338,8 @@ describe("MachineTest observed graph", () => { ) const nodes = Array.from(observed.graph) - const zero = nodes.find(([, node]) => - node.encoded.active.some(({ value }) => (value as { readonly value?: number }).value === 0) - )! - const one = nodes.find(([, node]) => - node.encoded.active.some(({ value }) => (value as { readonly value?: number }).value === 1) - )! + const zero = nodes.find(([, node]) => (node.snapshot.value as Count).value === 0)! + const one = nodes.find(([, node]) => (node.snapshot.value as Count).value === 1)! assert.notStrictEqual(zero[1].id, one[1].id) const shortest = Graph.dijkstra(observed.graph, { source: zero[0], @@ -402,8 +398,8 @@ describe("MachineTest observed graph", () => { assert.strictEqual(outside.length, 2) assert.notStrictEqual( - JSON.stringify(outside[0]!.encoded.history), - JSON.stringify(outside[1]!.encoded.history) + JSON.stringify(outside[0]!.snapshot.history), + JSON.stringify(outside[1]!.snapshot.history) ) assert.notStrictEqual(outside[0]!.id, outside[1]!.id) })) @@ -416,6 +412,7 @@ describe("MachineTest observed graph", () => { const observed = yield* MachineTest.observedGraph(opaqueMachine, [first, second]) assert.strictEqual(Graph.nodeCount(observed.graph), 2) + assert.ok(Array.from(observed.graph, ([, node]) => node).every((node) => node.encoded === undefined)) assert.strictEqual(MachineTest.coverage(opaqueMachine, [first, second]).logicalConfigurations.hit, 2) const firstBuffer = yield* MachineTest.run(opaqueMachine, { diff --git a/test/unstable/cluster/ClusterMachine.test.ts b/test/unstable/cluster/ClusterMachine.test.ts index bb797b1..12f9874 100644 --- a/test/unstable/cluster/ClusterMachine.test.ts +++ b/test/unstable/cluster/ClusterMachine.test.ts @@ -346,6 +346,46 @@ describe("ClusterMachine", () => { ))) })) + it.effect("rejects snapshot encoding failures before persistence", () => + Effect.gen(function*() { + interface OpaqueState { + readonly _tag: "OpaqueState" + readonly resource: object + } + + const OpaqueState = Schema.toCodecJson( + Schema.declare((value): value is OpaqueState => + typeof value === "object" && value !== null && "_tag" in value && value._tag === "OpaqueState" + ) + ) + const opaqueStates = Machine.states({ OpaqueState }) + const resource: { self?: unknown } = {} + resource.self = resource + const opaqueMachine = Machine.make({ + states: opaqueStates.states, + events: Machine.events(Fail), + initial: (to) => to.OpaqueState().resolve(({ target }) => target({ _tag: "OpaqueState", resource })) + }).handle({ + OpaqueState: { + on: { + Fail: (to) => to.full.OpaqueState().resolve(({ state: current, target }) => target(current)) + } + } + }) + const bridge = ClusterMachine.make("SnapshotEncodeFailureEntity", opaqueMachine, { version: "1" }) + const storage = makeTestStorage() + + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const makeClient = yield* bridge.entity.client + const result = yield* makeClient("opaque-1").send(new Fail({})) + + assertRejected(result, "SnapshotEncodeFailure") + assert.strictEqual(storage.entries.size, 0) + assert.strictEqual(storage.commits, 0) + }).pipe(Effect.provide(makeLayer(bridge, storage.service, () => Effect.void))) + })) + it.effect("does not apply a redelivered persisted request twice", () => Effect.gen(function*() { const gate = yield* Latch.make() diff --git a/typetest/testing/Coverage.tst.ts b/typetest/testing/Coverage.tst.ts index acbf675..9f3871b 100644 --- a/typetest/testing/Coverage.tst.ts +++ b/typetest/testing/Coverage.tst.ts @@ -52,12 +52,13 @@ describe("MachineTest coverage and observed graph", () => { MachineTest.ObservedGraphEdge > >() - expect>().type.toBe() + expect>().type.toBe() expect>().type.toBe() type Node = MachineTest.ObservedGraphNode type Edge = MachineTest.ObservedGraphEdge expect().type.toBe<"idle" | "done">() + expect().type.toBe() expect["event"]>().type.toBe() }) }) diff --git a/typetest/unstable/cluster/ClusterMachine.tst.ts b/typetest/unstable/cluster/ClusterMachine.tst.ts index d922502..e173014 100644 --- a/typetest/unstable/cluster/ClusterMachine.tst.ts +++ b/typetest/unstable/cluster/ClusterMachine.tst.ts @@ -24,6 +24,30 @@ describe("ClusterMachine", () => { value: Schema.Number }) {} + interface LocalResource { + readonly close: () => void + } + + const LocalResource = Schema.declare((value): value is LocalResource => + typeof value === "object" && value !== null && "close" in value && typeof value.close === "function" + ) + + class ResourceState extends Schema.TaggedClass("ResourceState")("ResourceState", { + resource: LocalResource + }) {} + + class ResourceEvent extends Schema.TaggedClass("ResourceEvent")("ResourceEvent", { + resource: LocalResource + }) {} + + class UnknownEvent extends Schema.TaggedClass("UnknownEvent")("UnknownEvent", { + payload: Schema.Unknown + }) {} + + class Scheduled extends Schema.TaggedClass("Scheduled")("Scheduled", { + at: Schema.Date + }) {} + class PlanningService extends Context.Service()("test/ClusterMachine/PlanningService") {} @@ -108,6 +132,103 @@ describe("ClusterMachine", () => { it("uses encoded Machine snapshots in checkpoints", () => { expect().type.toBe() + expect().type.toBe() + expect().type.toBe() + expect().type.toBe< + Readonly> + >() + }) + + it("requires JSON-encoded state, output, and public input schemas", () => { + const resourceStates = Machine.states({ ResourceState }) + const resourceMachine = Machine.make({ + states: resourceStates.states, + events: Machine.events(ResourceEvent), + initial: (to) => + to.ResourceState().resolve(({ target }) => target(new ResourceState({ resource: { close() {} } }))) + }) + + expect(ClusterMachine.make).type.not.toBeCallableWith( + "ResourceEntity", + resourceMachine, + { version: "1" } + ) + + const unknownEventMachine = Machine.make({ + states: states.states, + events: Machine.events(UnknownEvent), + initial: (to) => to.Count().resolve(({ target }) => target(new Count({ value: 0 }))) + }) + + expect(ClusterMachine.make).type.not.toBeCallableWith( + "UnknownEventEntity", + unknownEventMachine, + { version: "1" } + ) + + const anyOutputStates = Machine.states({ + Done: { schema: Done, type: "final", output: Schema.Any } + }) + const anyOutputMachine = Machine.make({ + states: anyOutputStates.states, + events: Machine.events(Reset), + initial: (to) => to.Done().resolve(({ target }) => target(new Done({ value: "done" }))) + }).handle({ + Done: { output: () => null } + }) + + expect(ClusterMachine.make).type.not.toBeCallableWith( + "AnyOutputEntity", + anyOutputMachine, + { version: "1" } + ) + + const neverOutputStates = Machine.states({ + Done: { schema: Done, type: "final", output: Schema.Never } + }) + const neverOutputMachine = Machine.make({ + states: neverOutputStates.states, + events: Machine.events(Reset), + initial: (to) => to.Done().resolve(({ target }) => target(new Done({ value: "done" }))) + }).handle({ + Done: { + output: () => { + throw new Error("unreachable") + } + } + }) + + expect(ClusterMachine.make).type.not.toBeCallableWith( + "NeverOutputEntity", + neverOutputMachine, + { version: "1" } + ) + + const scheduledStates = Machine.states({ Scheduled: Schema.toCodecJson(Scheduled) }) + const scheduledMachine = Machine.make({ + states: scheduledStates.states, + events: Machine.events(Reset), + initial: (to) => to.Scheduled().resolve(({ target }) => target(new Scheduled({ at: new Date("2026-08-19") }))) + }) + + expect(ClusterMachine.make).type.toBeCallableWith( + "ScheduledEntity", + scheduledMachine, + { version: "1" } + ) + + const internalResourceMachine = Machine.make({ + states: states.states, + events: Machine.events(Reset), + internalEvents: Machine.internalEvents(ResourceEvent), + initial: (to) => to.Count().resolve(({ target }) => target(new Count({ value: 0 }))) + }) + + expect(ClusterMachine.make).type.toBeCallableWith( + "InternalResourceEntity", + internalResourceMachine, + { version: "1" } + ) }) it("requires declared output implementations", () => {