diff --git a/.changeset/dynamic-child-machines.md b/.changeset/dynamic-child-machines.md new file mode 100644 index 0000000..5a9961b --- /dev/null +++ b/.changeset/dynamic-child-machines.md @@ -0,0 +1,14 @@ +--- +"@typeonce/effect-machine": minor +--- + +Add process-owned child machine spawning for runtime-sized child sets. + +Use `Machine.childFamily(machine)` to bind a child machine once, then call +`children.spawn(Family(id), { input })` inside an invoked Effect. Successfully +started children survive owner state changes and remain addressable through +machine references and `AtomMachine` until they stop or their parent stops. + +`Logic.Scope.spawn` accepts the same child descriptors for lower-level process +logic. Dynamic spawn calls retain child input, startup failure and service +inference, and check the child's declared parent protocol. diff --git a/README.md b/README.md index 215e148..294143d 100644 --- a/README.md +++ b/README.md @@ -582,6 +582,46 @@ entered. Use an Effect containing `Effect.sleep(...)` for generic work, while `from.timer(...)` keeps timer intent explicit and makes static durations visible through activity inspection. +### Spawn dynamic child machines + +Use `from.child(...)` when a state owns a fixed child lifecycle. Use the +`children` context inside an invoked Effect when the machine process owns an +open set of children that must survive state changes: + +```ts +const Plant = Machine.childFamily(plantMachine) + +const central = Machine.make({ + events: Machine.events(ResourcesOffered, PlantBroken) + // ... +}).handle({ + Commissioning: { + invoke: (from) => + from.effect("commission-wave", ({ children, state }) => + Effect.forEach( + state.plants, + (input) => children.spawn(Plant(input.id), { input }), + { discard: true } + )) + .onDone((to) => to.full.Operating()) + .onFailure((to) => to.full.CommissioningFailed()) + } +}) +``` + +`children.spawn` completes after initialization. The new child remains owned +by the machine process after the commissioning Effect completes or its state +exits. `children.sendTo` and `children.stop` address one active child from an +Effect; transition resolvers use `enqueue.sendTo` and `enqueue.stop` with the +same descriptor. Duplicate active ids fail with `ChildAlreadyExistsError` and +do not replace the existing child. Earlier successful spawns remain active if +a later spawn in the same wave fails. + +The child machine's declared `Machine.parent(...)` events must be accepted by +the owner. This is checked at each spawn call even though ids and cardinality +remain dynamic. `scope.spawn(child, { input })` provides the same descriptor +form for lower-level process logic, where the process event protocol is known. + ## Reactivity `AtomMachine` runs one lazy machine instance per `AtomRegistry`: @@ -603,6 +643,15 @@ The bridge exposes `ref`, `snapshot`, `state`, fail-aware `result`, writable equality-aware derivations. React applications using `@effect/atom-react` need a `RegistryProvider`. +Descriptors reconstructed from a `Machine.childFamily` resolve the same child +bridge by machine identity and id: + +```ts +const Plant = Machine.childFamily(plantMachine) +const plantAtom = centralAtom.child(Plant(selectedPlantId)) +const brokenAtom = AtomMachine.matchesChild(plantAtom, "Broken") +``` + Emissions stay streams rather than becoming retained atom state: ```ts diff --git a/docs/agent-guide.md b/docs/agent-guide.md index afeb483..eec45cf 100644 --- a/docs/agent-guide.md +++ b/docs/agent-guide.md @@ -315,6 +315,37 @@ machine. Do not start a promise inside a transition callback. A transition has no lifetime in which to own that work. A state does. +### Choose state-owned or process-owned children + +Use `from.child(...)` when the child belongs to one state and must stop when +that state exits. Use a child family and `children.spawn(...)` when runtime +events determine the ids or cardinality and the children must survive owner +state changes: + +```ts +const Worker = Machine.childFamily(workerMachine) + +Commissioning: { + invoke: (from) => + from.effect("start-workers", ({ children, state }) => + Effect.forEach( + state.workers, + (input) => children.spawn(Worker(input.id), { input }), + { discard: true } + ) + ) + .onDone((to) => to.full.Running()) + .onFailure((to) => to.full.Failed()) +} +``` + +The Effect owns the startup attempt. The machine process owns every child that +starts successfully. Leaving `Commissioning` does not stop those children. +Stop one with `children.stop(Worker(id))` inside an Effect or +`enqueue.stop(Worker(id))` inside a transition. A duplicate active id fails +instead of replacing the existing child, and a partially successful group is +not rolled back automatically. + ## Keep transition decisions synchronous A transition should choose the next state from the current snapshot and event. diff --git a/docs/effect-atom-react.md b/docs/effect-atom-react.md index e62fdab..555ecbd 100644 --- a/docs/effect-atom-react.md +++ b/docs/effect-atom-react.md @@ -200,3 +200,31 @@ the inferred dialog events. Use `dialogId` in atom labels for diagnostics. Do not pass it into `dialogMachine` as unused fake input. + +## 4. Selecting process-owned child machines + +Bind a machine definition once when a parent owns a runtime-sized set of child +machines: + +```ts +const Plant = Machine.childFamily(plantMachine) + +export const centralMachineAtom = machineAtoms.make(centralMachine) + +export const plantScopeFamily = Atom.family((plantId: string) => { + const plant = centralMachineAtom.child(Plant(plantId)) + + return { + stateAtom: plant.state, + isBrokenAtom: AtomMachine.matchesChild(plant, "Broken"), + sendAtom: plant.send, + stopAtom: plant.stop + } +}) +``` + +`Plant(plantId)` may be reconstructed wherever the id is available. Child +lookup and bridge reuse match by machine identity and id, not descriptor object +identity. Before the parent spawns that child, selectors contain `Option.none` +and `matchesChild` is `false`. They follow the child after startup and return to +the inactive values after it stops. diff --git a/src/Machine.ts b/src/Machine.ts index db5e858..6e755b6 100644 --- a/src/Machine.ts +++ b/src/Machine.ts @@ -2265,7 +2265,15 @@ export declare namespace Logic { * @category models * @since 0.4.0 */ - export interface Spawn { + export interface Spawn { + ( + child: Child & ChildMachine.Executable & ChildMachine.ParentCompatibility, + ...options: ChildMachine.SpawnArgs + ): Effect.Effect< + ChildMachine.Ref, + ChildAlreadyExistsError | ChildMachine.StartError, + ChildMachine.StartRequirements + > ( logic: Logic ): Effect.Effect< @@ -2305,7 +2313,7 @@ export declare namespace Logic { readonly parent: Address | undefined /** Starts a child process owned by this scope. */ - readonly spawn: Spawn + readonly spawn: Spawn /** Sends an event to a machine target or typed parent-local child address. */ readonly sendTo: { @@ -2345,6 +2353,7 @@ export declare namespace Logic { const ChildAddressTypeId = "~effect/Machine/ChildAddress" const ChildAddressCompatibilityErrorTypeId = "~effect/Machine/ChildAddressCompatibilityError" +const ChildParentCompatibilityErrorTypeId = "~effect/Machine/ChildParentCompatibilityError" const ChildMachineTypeId = "~effect/Machine/ChildMachine" type InvokeLifecycleId = string & { readonly [ChildAddressTypeId]?: never } @@ -2389,6 +2398,95 @@ export declare namespace ChildMachine { */ export type Any = ChildMachine + /** + * Bound constructor for an open family of child descriptors that share one + * machine definition. + * + * @category models + * @since 0.20.0 + */ + export interface Family { + (id: Id): ChildMachine + } + + /** + * Ensures a child machine's declared owner protocol is accepted by the + * process that will own it. + * + * @category utility types + * @since 0.20.0 + */ + export type ParentCompatibility = Child extends ChildMachine ? + Machine.Any extends M ? { + readonly [ChildParentCompatibilityErrorTypeId]: { + readonly child: unknown + readonly owner: OwnerEvent + } + } + : [Machine.EventOf>] extends [OwnerEvent] ? unknown : + { + readonly [ChildParentCompatibilityErrorTypeId]: { + readonly child: Machine.EventOf> + readonly owner: OwnerEvent + } + } + : never + + /** + * Ensures the selected child machine has complete handlers and outputs. + * + * @category utility types + * @since 0.20.0 + */ + export type Executable = Child["machine"] extends EnsureExecutable< + Machine.States, + Machine.UnhandledStates, + Machine.OutputStates + > ? unknown + : never + + /** + * Startup arguments accepted while spawning a child machine. + * + * @category utility types + * @since 0.20.0 + */ + export type SpawnArgs = Machine.InputSchema extends typeof Schema.Void ? + [options?: { readonly input?: never }] + : [options: { readonly input: Machine.Input }] + + /** + * Typed failures that may occur before a spawned child becomes active. + * + * @category utility types + * @since 0.20.0 + */ + export type StartError = Child extends ChildMachine ? + | Machine.InitialError + | Machine.Error + | ActionError | Machine.Services> + | InfiniteTransitionError + | MachineSchemaDecodeError + | StartupError + | StoppedError + : never + + /** + * Services needed to initialize a spawned child machine. + * + * @category utility types + * @since 0.20.0 + */ + export type StartRequirements = Child extends ChildMachine ? Exclude< + ExcludeCompatibleRuntime< + Exclude | Machine.Services>, MachineRuntimeRequirement>, + Machine.Event, + Machine.Emit + >, + Scope.Scope + > + : never + /** * Running machine reference selected by a child descriptor. * @@ -2418,6 +2516,34 @@ export declare namespace ChildMachine { : never } +/** + * Effectful operations for child machines owned directly by the current + * machine process. + * + * @category models + * @since 0.20.0 + */ +export interface ChildOwner { + /** Starts a process-owned child and returns once initialization succeeds. */ + readonly spawn: ( + child: Child & ChildMachine.Executable & ChildMachine.ParentCompatibility, + ...options: ChildMachine.SpawnArgs + ) => Effect.Effect< + ChildMachine.Ref, + ChildAlreadyExistsError | ChildMachine.StartError, + ChildMachine.StartRequirements + > + + /** Sends an event to one active child. Missing children are ignored. */ + readonly sendTo: ( + child: Child, + event: ChildMachine.Event + ) => Effect.Effect + + /** Stops one active child. Missing children are ignored. */ + readonly stop: (child: Child) => Effect.Effect +} + /** * Parent-local address for a child process that can receive events. * @@ -4799,6 +4925,8 @@ export declare namespace Machine { InputEvents extends ReadonlyArray = Events, ParentEvents extends ReadonlyArray = readonly [] > = MachineReferences & { + /** Process-owned child operations for dynamic child machine lifecycles. */ + readonly children: ChildOwner> /** Value owned by the state that owns this invocation. */ readonly state: StateByIdentifier /** Value owned by the nearest schema-backed ancestor, when one exists. */ @@ -8917,6 +9045,17 @@ export const logic: < export const child: (id: Id, machine: M) => ChildMachine = internal.child +/** + * Binds one machine definition to an open family of runtime child ids. + * + * Descriptors created by the returned function are interchangeable with + * {@link child} descriptors for the same id and machine definition. + * + * @category constructors + * @since 0.20.0 + */ +export const childFamily: (machine: M) => ChildMachine.Family = internal.childFamily + /** * Creates a typed parent-local address for lower-level child process logic. * diff --git a/src/internal/machine/atom.ts b/src/internal/machine/atom.ts index e8218f1..007128d 100644 --- a/src/internal/machine/atom.ts +++ b/src/internal/machine/atom.ts @@ -342,15 +342,7 @@ const makeChildFromRefAtom = - makeChildFromRefAtom( - makeChildRefAtom(ref as any, nested), - nested - ) - ) - const child = ( - nested: Nested - ): ChildMachineAtom => childFamily(nested) as ChildMachineAtom + const child = makeChildSelector(ref as any) return { ref, @@ -363,6 +355,30 @@ const makeChildFromRefAtom = ( + parentRef: Atom.Atom< + AsyncResult.AsyncResult>, StartError> + > +) => { + const byMachine = new WeakMap ChildMachineAtom>() + return (descriptor: Child): ChildMachineAtom => { + let family = byMachine.get(descriptor.machine) + if (family === undefined) { + const machine = descriptor.machine + const atoms = Atom.family((id: string) => { + const child = internalMachine.child(id, machine) + return makeChildFromRefAtom( + makeChildRefAtom(parentRef as any, child), + child + ) + }) + family = (id) => atoms(id) as ChildMachineAtom + byMachine.set(machine, family) + } + return family(descriptor.id) as ChildMachineAtom + } +} + const makeFromRefAtom = ( ref: Atom.Atom, StartError>> ): MachineAtom => { @@ -438,15 +454,7 @@ const makeFromRefAtom = ( ) const optionalRef = Atom.mapResult(ref, Option.some) - const childFamily = Atom.family((descriptor: Machine.ChildMachine.Any) => - makeChildFromRefAtom( - makeChildRefAtom(optionalRef as any, descriptor), - descriptor - ) - ) - const child = ( - descriptor: Child - ): ChildMachineAtom => childFamily(descriptor) as ChildMachineAtom + const child = makeChildSelector(optionalRef as any) return { ref, diff --git a/src/internal/machine/invocation.ts b/src/internal/machine/invocation.ts index 5e73b3b..f36ca0d 100644 --- a/src/internal/machine/invocation.ts +++ b/src/internal/machine/invocation.ts @@ -7,7 +7,7 @@ import * as Cause from "effect/Cause" import * as Effect from "effect/Effect" import * as Stream from "effect/Stream" -import type { ChildMachine, Inspection, Logic, Machine } from "../../Machine.js" +import type { ChildMachine, ChildOwner, Inspection, Logic, Machine } from "../../Machine.js" import * as Configuration from "./configuration.js" import { InfiniteTransitionError, MachineSchemaDecodeError, StoppedError } from "./errors.js" import * as InvocationEvent from "./invocationEvent.js" @@ -63,6 +63,16 @@ const streamLogic = ( const resolveValue = (value: unknown, context: Machine.InvokeContext): unknown => typeof value === "function" ? value(context) : value +const makeChildOwner = (scope: Runtime.ProcessScope): ChildOwner => ({ + spawn: + ((descriptor: ChildMachine.Any, options?: { readonly input?: unknown }) => + (scope.spawn as any)(descriptor, options)) as ChildOwner["spawn"], + sendTo: ((descriptor: ChildMachine.Any, event: unknown) => scope.sendTo(descriptor, event)) as ChildOwner< + any + >["sendTo"], + stop: ((descriptor: ChildMachine.Any) => scope.stopChild(descriptor)) as ChildOwner["stop"] +}) + const resolveOne = ( raw: Record, context: Machine.InvokeContext, @@ -291,11 +301,13 @@ export const startAll = ( paths: ReadonlyArray, event: Machine.LifecycleEvent ): Effect.Effect | undefined => { + const children = makeChildOwner(scope) const effects = Planner.sortEntryPaths(machine, paths) .filter((path) => configuration.active.has(path)) .flatMap((path) => { const context = { ...(Configuration.getMachineReferences(configuration) ?? { self: scope.self, parent: scope.parent }), + children, state: configuration.values.get(path), containingState: Configuration.getParentValue(machine, configuration, path), ancestors: Configuration.getParentValues(machine, configuration, path), diff --git a/src/internal/machine/machine.ts b/src/internal/machine/machine.ts index 5951d10..fd28b2d 100644 --- a/src/internal/machine/machine.ts +++ b/src/internal/machine/machine.ts @@ -2124,15 +2124,30 @@ export const transition = ( export const child = ( id: Id, machine: M +): ChildMachine => + makeChild(id, machine, (input) => + machine.input === undefined + ? (internalProcess.toProcessLogic as any)(machine) + : (internalProcess.toProcessLogic as any)(machine, input)) + +const makeChild = ( + id: Id, + machine: M, + makeLogic: (input?: unknown) => Logic ): ChildMachine => ({ [ChildMachineTypeId]: ChildMachineTypeId, id, machine, - [ChildMachineLogicTypeId]: (input) => + [ChildMachineLogicTypeId]: makeLogic +}) + +export const childFamily = (machine: M): ChildMachine.Family => { + const makeLogic = (input?: unknown): Logic => machine.input === undefined ? (internalProcess.toProcessLogic as any)(machine) : (internalProcess.toProcessLogic as any)(machine, input) -}) + return (id) => makeChild(id, machine, makeLogic) +} export const childAddress = (id: string): ChildAddress => id as ChildAddress @@ -2174,10 +2189,7 @@ export const spawn: { SpawnError, ChildInitialError > -} = (( - logic: Logic, - options?: SpawnOptions -) => +} = ((logic: Logic, options?: SpawnOptions) => Effect.flatMap( internalRuntime.MachineRuntime, (runtime) => options === undefined ? runtime.spawn(logic) : (runtime.spawn as any)(logic, options) diff --git a/src/internal/machine/runtime.ts b/src/internal/machine/runtime.ts index e372b0e..ed42e1d 100644 --- a/src/internal/machine/runtime.ts +++ b/src/internal/machine/runtime.ts @@ -18,9 +18,10 @@ import * as Scope from "effect/Scope" import * as Stream from "effect/Stream" import * as SynchronizedRef from "effect/SynchronizedRef" import type * as Take from "effect/Take" -import type { Inspection, Machine as MachineDefinition, MachineTarget } from "../../Machine.js" +import type { ChildMachine, Inspection, Machine as MachineDefinition, MachineTarget } from "../../Machine.js" import { ChildAlreadyExistsError, StoppedError } from "./errors.js" import * as InspectionRuntime from "./inspectionRuntime.js" +import { ChildMachineLogicTypeId } from "./symbols.js" type ChildDescriptor = { readonly id: string @@ -376,7 +377,7 @@ const sendMachineTarget = ( export interface ProcessScope { readonly self: ProcessAddress readonly parent: ProcessAddress | undefined - readonly spawn: ProcessSpawn + readonly spawn: ProcessSpawn readonly sendParent: (event: unknown) => Effect.Effect readonly emit: (event: unknown) => Effect.Effect readonly sendTo: { @@ -552,7 +553,15 @@ export interface ProcessLogic< run(context: ProcessContext): Effect.Effect } -export interface ProcessSpawn { +export interface ProcessSpawn { + ( + child: Child & ChildMachine.Executable & ChildMachine.ParentCompatibility, + ...options: ChildMachine.SpawnArgs + ): Effect.Effect< + ChildMachine.Ref, + ChildAlreadyExistsError | ChildMachine.StartError, + ChildMachine.StartRequirements + > ( logic: ProcessLogic ): Effect.Effect< @@ -1207,6 +1216,14 @@ const makeChildRuntimeSync = ( }) } + function spawn( + child: Child & ChildMachine.Executable & ChildMachine.ParentCompatibility, + ...options: ChildMachine.SpawnArgs + ): Effect.Effect< + ChildMachine.Ref, + ChildAlreadyExistsError | ChildMachine.StartError, + ChildMachine.StartRequirements + > function spawn( logic: ProcessLogic ): Effect.Effect< @@ -1232,32 +1249,44 @@ const makeChildRuntimeSync = ( ChildAlreadyExistsError | ChildInitialError, Exclude > - function spawn( - logic: ProcessLogic, - spawnOptions?: { + function spawn( + logicOrChild: ProcessLogic | ChildMachine.Any, + options?: { readonly id: string readonly descriptor?: ChildDescriptor readonly onOutcome?: ( - outcome: RuntimeOutcome + outcome: RuntimeOutcome ) => Effect.Effect readonly [activeSnapshotObserver]?: ( - snapshot: Extract, { readonly status: "active" }> + snapshot: Extract, { readonly status: "active" }> ) => Effect.Effect readonly [sendParentOverride]?: (event: unknown) => Effect.Effect - } - ): Effect.Effect< - MachineRef, - ChildAlreadyExistsError | ChildInitialError, - Exclude - > { + } | { readonly input?: unknown } + ): Effect.Effect, any, any> { + const descriptor = typeof logicOrChild === "object" && logicOrChild !== null && + ChildMachineLogicTypeId in logicOrChild + ? logicOrChild as ChildMachine.Any + : undefined + const logic = descriptor === undefined + ? logicOrChild as ProcessLogic + : descriptor[ChildMachineLogicTypeId]( + (options as { readonly input?: unknown } | undefined)?.input + ) as unknown as ProcessLogic + const spawnOptions = descriptor === undefined + ? options as { + readonly id: string + readonly descriptor?: ChildDescriptor + readonly onOutcome?: (outcome: RuntimeOutcome) => Effect.Effect + readonly [activeSnapshotObserver]?: ( + snapshot: Extract, { readonly status: "active" }> + ) => Effect.Effect + readonly [sendParentOverride]?: (event: unknown) => Effect.Effect + } | undefined + : { id: descriptor.id, descriptor } const token = Symbol() const key = spawnOptions?.id ?? token let startedChild: MachineRef | undefined - return Effect.suspend((): Effect.Effect< - MachineRef, - ChildAlreadyExistsError | ChildInitialError, - Exclude - > => { + return Effect.suspend(() => { if (registry.closed) { return Effect.interrupt } diff --git a/src/unstable/reactivity/AtomMachine.ts b/src/unstable/reactivity/AtomMachine.ts index f623adc..2ea96f9 100644 --- a/src/unstable/reactivity/AtomMachine.ts +++ b/src/unstable/reactivity/AtomMachine.ts @@ -148,9 +148,9 @@ export interface MachineAtom, void> /** - * Creates a reactive bridge for a directly invoked child machine. - * Reusing the same descriptor returns the same live bridge while it remains - * referenced. + * Creates a reactive bridge for a directly owned child machine. + * Descriptors with the same id and machine definition return the same live + * bridge while it remains referenced. * * @since 0.4.0 */ @@ -206,7 +206,8 @@ export const childEmissions: = internal.childEmissions /** - * Reactive access to one invoked child machine selected by its descriptor. + * Reactive access to one directly owned child machine selected by its + * descriptor. * * **Details** * @@ -295,8 +296,9 @@ export interface ChildMachineAtom /** - * Creates a reactive bridge for a directly owned nested child. Reusing the - * same descriptor returns the same live bridge while it remains referenced. + * Creates a reactive bridge for a directly owned nested child. Descriptors + * with the same id and machine definition return the same live bridge while + * it remains referenced. * * @since 0.4.0 */ @@ -418,7 +420,7 @@ export const selectSnapshot: < > = internal.selectSnapshot /** - * Selects the typed value for an active state path in an invoked child. + * Selects the typed value for an active state path in a directly owned child. * * Valid paths and their selected value types are inferred from the child * bridge. An inactive child produces `Option.none()`. Keep the returned atom @@ -513,7 +515,7 @@ export const matches: < ) => Atom.Atom> = internal.matches /** - * Returns whether a state path is active in an invoked child. + * Returns whether a state path is active in a directly owned child. * * Valid paths are inferred from the child bridge snapshot. * An inactive child produces `false`. Keep the returned atom stable when diff --git a/test/internal/machine/strategyDifferential.test.ts b/test/internal/machine/strategyDifferential.test.ts index 5fc963d..ee0644a 100644 --- a/test/internal/machine/strategyDifferential.test.ts +++ b/test/internal/machine/strategyDifferential.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest" -import { Deferred, Effect, Fiber, Schema, Stream } from "effect" +import { Deferred, Effect, Fiber, Option, Schema, Stream } from "effect" import { FastCheck } from "effect/testing" import { Machine } from "../../../src/index.js" import * as Configuration from "../../../src/internal/machine/configuration.js" @@ -881,6 +881,56 @@ describe("machine planner and runtime strategies", () => { } }) as Effect.Effect) + it.effect("keeps dynamically spawned child machines across generic and compiled state changes", () => + Effect.gen(function*() { + class ChildIdle extends Schema.TaggedClass("StrategyDynamicChildIdle")("ChildIdle", {}) {} + const childMachine = Machine.make({ + states: { ChildIdle }, + events: Machine.events(), + initial: (to) => to.ChildIdle().resolve(({ target }) => target(new ChildIdle({}))) + }).handle({ ChildIdle: {} }) + const Child = Machine.childFamily(childMachine) + class Commissioning extends Schema.TaggedClass("StrategyDynamicCommissioning")( + "Commissioning", + {} + ) {} + class Operating extends Schema.TaggedClass("StrategyDynamicOperating")("Operating", {}) {} + const machine = Machine.make({ + states: { Commissioning, Operating }, + events: Machine.events(), + initial: (to) => to.Commissioning().resolve(({ target }) => target(new Commissioning({}))) + }).handle({ + Commissioning: { + invoke: (from) => + from.effect("commission", ({ children }) => children.spawn(Child("runtime"))).onDone((to) => + to.full.Operating() + ).onFailure((to) => to.none) + }, + Operating: {} + }) + + assert.strictEqual(ExecutionPlan.selectExecutionPlanForTesting(machine, "auto").strategy, "indexed-flat") + + const results: Array = [] + for (const strategy of ["generic", "compiled"] as const) { + const ref = yield* openWithRuntimeStrategy(machine, strategy) + yield* ref.changes.pipe( + Stream.filter((snapshot) => snapshot.status === "active" && snapshot.state.path === "Operating"), + Stream.take(1), + Stream.runDrain + ) + const child = yield* ref.child(Child("runtime")) + assert(Option.isSome(child)) + results.push({ + parent: yield* ref.state, + child: yield* child.value.state + }) + yield* ref.stop + assert.strictEqual((yield* child.value.snapshot).status, "stopped") + } + assert.deepStrictEqual(results[1], results[0]) + }) as Effect.Effect) + it.effect("compares generated eligible models across canonical and indexed planners", () => Effect.gen(function*() { const generated = MachineTest.finiteModels({ diff --git a/test/machine/DynamicChildren.test.ts b/test/machine/DynamicChildren.test.ts new file mode 100644 index 0000000..99e011a --- /dev/null +++ b/test/machine/DynamicChildren.test.ts @@ -0,0 +1,343 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Fiber, Option, Schema, Stream } from "effect" +import { Machine } from "../../src/index.js" + +const waitForPath = , Event, Error, Output>( + ref: Machine.MachineRef, + path: State["path"] +) => + ref.changes.pipe( + Stream.filter((snapshot) => snapshot.status === "active" && snapshot.state.path === path), + Stream.take(1), + Stream.runDrain + ) + +describe("dynamic child machines", () => { + it.effect("spawns input-bearing children that survive owner state changes", () => + Effect.gen(function*() { + class PlantActive extends Schema.TaggedClass("DynamicPlantActive")("PlantActive", { + id: Schema.String, + produced: Schema.Number + }) {} + class Produce extends Schema.TaggedClass("DynamicPlantProduce")("Produce", { + amount: Schema.Number + }) {} + class Report extends Schema.TaggedClass("DynamicPlantReport")("Report", {}) {} + class PlantReported extends Schema.TaggedClass("DynamicPlantReported")("PlantReported", { + id: Schema.String, + produced: Schema.Number + }) {} + const PlantInput = Schema.Struct({ id: Schema.String, production: Schema.Number }) + const PlantOwnerEvents = Machine.events(PlantReported) + const plantStates = Machine.states({ PlantActive }) + const plantMachine = Machine.make({ + states: plantStates.states, + events: Machine.events(Produce, Report), + input: PlantInput, + parent: Machine.parent(PlantOwnerEvents), + initial: (to) => + to.PlantActive().resolve(({ input, target }) => + target(new PlantActive({ id: input.id, produced: input.production })) + ) + }).handle({ + PlantActive: { + on: { + Produce: (to) => + to.full.PlantActive().resolve(({ event, state, target }) => + target(new PlantActive({ ...state, produced: state.produced + event.amount })) + ), + Report: (to) => + to.none.resolve(({ parent, state }, enqueue) => { + enqueue.sendTo(parent, PlantOwnerEvents.PlantReported({ id: state.id, produced: state.produced })) + }) + } + } + }) + const Plant = Machine.childFamily(plantMachine) + + class Commissioning extends Schema.TaggedClass("DynamicParentCommissioning")( + "Commissioning", + { plants: Schema.Array(PlantInput) } + ) {} + class Operating extends Schema.TaggedClass("DynamicParentOperating")("Operating", { + reports: Schema.Number + }) {} + class Grow extends Schema.TaggedClass("DynamicGrow")("Grow", { + plants: Schema.Array(PlantInput) + }) {} + class Decommission extends Schema.TaggedClass("DynamicDecommission")("Decommission", { + id: Schema.String + }) {} + const parentStates = Machine.states({ Commissioning, Operating }) + const parentMachine = Machine.make({ + states: parentStates.states, + events: Machine.events(PlantOwnerEvents, Grow, Decommission), + input: Schema.Array(PlantInput), + initial: (to) => to.Commissioning().resolve(({ input, target }) => target(new Commissioning({ plants: input }))) + }).handle({ + Commissioning: { + invoke: (from) => + from.effect("commission-wave", ({ children, state }) => + Effect.forEach( + state.plants, + (input) => children.spawn(Plant(input.id), { input }), + { discard: true } + )).onDone((to) => to.full.Operating().resolve(({ target }) => target(new Operating({ reports: 0 })))) + .onFailure((to) => to.none) + }, + Operating: { + on: { + PlantReported: (to) => + to.full.Operating().resolve(({ state, target }) => target(new Operating({ reports: state.reports + 1 }))), + Grow: (to) => + to.full.Commissioning().resolve(({ event, target }) => + target(new Commissioning({ plants: event.plants })) + ), + Decommission: (to) => + to.none.resolve(({ event }, enqueue) => { + enqueue.stop(Plant(event.id)) + }) + } + } + }) + + const parent = yield* Machine.start(parentMachine, [ + { id: "p-1", production: 10 }, + { id: "p-2", production: 20 } + ]) + yield* waitForPath(parent, "Operating").pipe(Effect.timeout("1 second")) + + const first = yield* parent.child(Plant("p-1")) + const second = yield* parent.child(Plant("p-2")) + assert(Option.isSome(first)) + assert(Option.isSome(second)) + const reconstructedFirst = yield* parent.child(Machine.child("p-1", plantMachine)) + assert(Option.isSome(reconstructedFirst)) + assert.strictEqual(reconstructedFirst.value, first.value) + + const produced = yield* first.value.changes.pipe( + Stream.filter((snapshot) => snapshot.status === "active" && snapshot.state.value.produced === 15), + Stream.take(1), + Stream.runDrain, + Effect.forkChild + ) + yield* first.value.send(new Produce({ amount: 5 })) + yield* Fiber.join(produced) + + const thirdStarted = yield* parent.childChanges(Plant("p-3")).pipe( + Stream.filter(Option.isSome), + Stream.take(1), + Stream.runDrain, + Effect.forkChild + ) + yield* parent.send( + new Grow({ + plants: [ + { id: "p-3", production: 30 }, + { id: "p-4", production: 40 } + ] + }) + ) + yield* Fiber.join(thirdStarted) + yield* waitForPath(parent, "Operating").pipe(Effect.timeout("1 second")) + + const firstAfterTransitions = yield* parent.child(Plant("p-1")) + assert(Option.isSome(firstAfterTransitions)) + assert.strictEqual(firstAfterTransitions.value, first.value) + assert(Option.isSome(yield* parent.child(Plant("p-3")))) + assert.deepStrictEqual(yield* firstAfterTransitions.value.state, { + path: "PlantActive", + value: new PlantActive({ id: "p-1", produced: 15 }) + }) + + const reported = yield* parent.changes.pipe( + Stream.filter((snapshot) => + snapshot.status === "active" && snapshot.state.path === "Operating" && snapshot.state.value.reports === 1 + ), + Stream.take(1), + Stream.runDrain, + Effect.forkChild + ) + yield* first.value.send(new Report({})) + yield* Fiber.join(reported) + + const decommissioned = yield* parent.childChanges(Plant("p-1")).pipe( + Stream.filter(Option.isNone), + Stream.take(1), + Stream.runDrain, + Effect.forkChild + ) + yield* parent.send(new Decommission({ id: "p-1" })) + yield* Fiber.join(decommissioned) + assert(Option.isNone(yield* parent.child(Plant("p-1")))) + assert(Option.isSome(yield* parent.child(Plant("p-2")))) + + const recommissioned = yield* parent.childChanges(Plant("p-1")).pipe( + Stream.filter(Option.isSome), + Stream.take(1), + Stream.runCollect, + Effect.map((values) => Array.from(values)[0]!), + Effect.forkChild + ) + yield* parent.send(new Grow({ plants: [{ id: "p-1", production: 50 }] })) + const replacement = yield* Fiber.join(recommissioned) + assert(Option.isSome(replacement)) + assert.notStrictEqual(replacement.value, first.value) + assert.deepStrictEqual(yield* replacement.value.state, { + path: "PlantActive", + value: new PlantActive({ id: "p-1", produced: 50 }) + }) + + yield* parent.stop + assert.strictEqual((yield* second.value.snapshot).status, "stopped") + })) + + it.effect("rejects duplicate dynamic ids without replacing the active child", () => + Effect.gen(function*() { + class ChildIdle extends Schema.TaggedClass("DynamicDuplicateChildIdle")("ChildIdle", {}) {} + const childMachine = Machine.make({ + states: { ChildIdle }, + events: Machine.events(), + initial: (to) => to.ChildIdle().resolve(({ target }) => target(new ChildIdle({}))) + }).handle({ ChildIdle: {} }) + const Child = Machine.childFamily(childMachine) + class Starting extends Schema.TaggedClass("DynamicDuplicateStarting")("Starting", {}) {} + class DuplicateRejected extends Schema.TaggedClass("DynamicDuplicateRejected")( + "DuplicateRejected", + {} + ) {} + const parentMachine = Machine.make({ + states: { Starting, DuplicateRejected }, + events: Machine.events(), + initial: (to) => to.Starting().resolve(({ target }) => target(new Starting({}))) + }).handle({ + Starting: { + invoke: (from) => + from.effect("spawn-duplicate", ({ children }) => + children.spawn(Child("same")).pipe( + Effect.andThen(children.spawn(Child("same"))) + )).onDone((to) => to.none).onFailure((to) => to.full.DuplicateRejected()) + }, + DuplicateRejected: {} + }) + + const parent = yield* Machine.start(parentMachine) + yield* waitForPath(parent, "DuplicateRejected").pipe(Effect.timeout("1 second")) + + assert(Option.isSome(yield* parent.child(Child("same")))) + yield* parent.stop + })) + + it.effect("spawns child machine descriptors from process logic", () => + Effect.gen(function*() { + class WorkerIdle extends Schema.TaggedClass("DynamicLogicWorkerIdle")("WorkerIdle", { + id: Schema.String + }) {} + const Input = Schema.Struct({ id: Schema.String }) + const workerMachine = Machine.make({ + states: { WorkerIdle }, + events: Machine.events(), + input: Input, + initial: (to) => to.WorkerIdle().resolve(({ input, target }) => target(new WorkerIdle({ id: input.id }))) + }).handle({ WorkerIdle: {} }) + const Worker = Machine.childFamily(workerMachine) + let scoped: Machine.ChildMachine.Ref> | undefined + let second: Machine.ChildMachine.Ref> | undefined + const supervisorLogic = Machine.logic({ + initial: ({ spawn }) => + Effect.all([ + spawn(Worker("scoped"), { input: { id: "scoped" } }), + spawn(Worker("second"), { input: { id: "second" } }) + ]).pipe( + Effect.tap(([scopedRef, secondRef]) => + Effect.sync(() => { + scoped = scopedRef + second = secondRef + }) + ), + Effect.as(undefined) + ), + run: () => Effect.never + }) + const Supervisor = Machine.childAddress("supervisor") + class Running extends Schema.TaggedClass("DynamicLogicRunning")("Running", {}) {} + const parentMachine = Machine.make({ + states: { Running }, + events: Machine.events(), + initial: (to) => to.Running().resolve(({ target }) => target(new Running({}))) + }).handle({ + Running: { + invoke: (from) => from.logic("supervisor", { address: Supervisor, logic: supervisorLogic }) + } + }) + + const parent = yield* Machine.start(parentMachine) + assert(scoped !== undefined) + assert(second !== undefined) + assert.deepStrictEqual(yield* scoped.state, { + path: "WorkerIdle", + value: new WorkerIdle({ id: "scoped" }) + }) + assert.deepStrictEqual(yield* second.state, { + path: "WorkerIdle", + value: new WorkerIdle({ id: "second" }) + }) + + yield* parent.stop + assert.strictEqual((yield* scoped.snapshot).status, "stopped") + assert.strictEqual((yield* second.snapshot).status, "stopped") + })) + + it.effect("sends to and stops process-owned children from an invoked Effect", () => + Effect.gen(function*() { + class UnitActive extends Schema.TaggedClass("DynamicControlUnitActive")("UnitActive", { + count: Schema.Number + }) {} + class Increment extends Schema.TaggedClass("DynamicControlIncrement")("Increment", {}) {} + const unitMachine = Machine.make({ + states: { UnitActive }, + events: Machine.events(Increment), + initial: (to) => to.UnitActive().resolve(({ target }) => target(new UnitActive({ count: 0 }))) + }).handle({ + UnitActive: { + on: { + Increment: (to) => + to.full.UnitActive().resolve(({ state, target }) => target(new UnitActive({ count: state.count + 1 }))) + } + } + }) + const Unit = Machine.childFamily(unitMachine) + class Managing extends Schema.TaggedClass("DynamicControlManaging")("Managing", {}) {} + class Ready extends Schema.TaggedClass("DynamicControlReady")("Ready", {}) {} + const parentMachine = Machine.make({ + states: { Managing, Ready }, + events: Machine.events(), + initial: (to) => to.Managing().resolve(({ target }) => target(new Managing({}))) + }).handle({ + Managing: { + invoke: (from) => + from.effect("control-units", ({ children }) => + Effect.gen(function*() { + yield* children.spawn(Unit("kept")) + yield* children.spawn(Unit("stopped")) + yield* children.sendTo(Unit("kept"), new Increment({})) + yield* children.stop(Unit("stopped")) + })).onDone((to) => to.full.Ready()).onFailure((to) => to.none) + }, + Ready: {} + }) + + const parent = yield* Machine.start(parentMachine) + yield* waitForPath(parent, "Ready").pipe(Effect.timeout("1 second")) + const kept = yield* parent.child(Unit("kept")) + assert(Option.isSome(kept)) + yield* kept.value.changes.pipe( + Stream.filter((snapshot) => snapshot.status === "active" && snapshot.state.value.count === 1), + Stream.take(1), + Stream.runDrain, + Effect.timeout("1 second") + ) + assert(Option.isNone(yield* parent.child(Unit("stopped")))) + yield* parent.stop + })) +}) diff --git a/test/unstable/reactivity/AtomMachine.test.ts b/test/unstable/reactivity/AtomMachine.test.ts index e3cda1d..778d06b 100644 --- a/test/unstable/reactivity/AtomMachine.test.ts +++ b/test/unstable/reactivity/AtomMachine.test.ts @@ -225,6 +225,7 @@ describe("AtomMachine", () => { const childAtoms = parentAtoms.child(Child) assert.strictEqual(parentAtoms.child(Child), childAtoms) const Alias = Machine.child("counter", childMachine) + assert.strictEqual(parentAtoms.child(Alias), childAtoms) const Impostor = Machine.child("counter", makeCounterMachine()) const impostorAtoms = parentAtoms.child(Impostor) const selectedCount = AtomMachine.selectChild(childAtoms, "Count") @@ -301,6 +302,40 @@ describe("AtomMachine", () => { assert.strictEqual(yield* AtomRegistry.getResult(registry, countMatches), false) }))) + it.effect("reactively exposes a dynamically spawned child by family and runtime id", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const childMachine = makeCounterMachine() + const Child = Machine.childFamily(childMachine) + const parent = Machine.make({ + states: { Count }, + events: Machine.events(), + initial: (to) => to.Count().resolve(({ target }) => target(new Count({ value: 0 }))) + }).handle({ + Count: { + invoke: (from) => + from.effect("spawn-counter", ({ children }) => children.spawn(Child("dynamic"))).onDone((to) => to.none) + .onFailure((to) => to.none) + } + }) + const parentAtoms = AtomMachine.make(parent) + const childAtoms = parentAtoms.child(Child("dynamic")) + assert.strictEqual(parentAtoms.child(Child("dynamic")), childAtoms) + assert.strictEqual(parentAtoms.child(Machine.child("dynamic", childMachine)), childAtoms) + const selected = AtomMachine.selectChild(childAtoms, "Count") + const matches = AtomMachine.matchesChild(childAtoms, "Count") + + yield* mount(registry, childAtoms.state) + const active = yield* waitForResult(registry, selected, Option.isSome) + assert(Option.isSome(active)) + assert.strictEqual(active.value.value, 0) + assert.strictEqual(yield* AtomRegistry.getResult(registry, matches), true) + + yield* Effect.sync(() => registry.set(childAtoms.stop, undefined)) + assert(Option.isNone(yield* waitForResult(registry, childAtoms.ref, Option.isNone))) + assert.strictEqual(yield* AtomRegistry.getResult(registry, matches), false) + }))) + it.effect("exposes snapshots and sends events", () => Effect.scoped(Effect.gen(function*() { const registry = yield* makeRegistry diff --git a/typetest/machine/DynamicChildren.tst.ts b/typetest/machine/DynamicChildren.tst.ts new file mode 100644 index 0000000..b47a7ed --- /dev/null +++ b/typetest/machine/DynamicChildren.tst.ts @@ -0,0 +1,107 @@ +import { Effect, Schema } from "effect" +import { describe, expect, it } from "tstyche" +import { Machine } from "../../src/index.js" + +describe("dynamic child machines", () => { + class ChildIdle extends Schema.TaggedClass("DynamicTypesChildIdle")("ChildIdle", { + id: Schema.String + }) {} + class ChildEvent extends Schema.TaggedClass("DynamicTypesChildEvent")("ChildEvent", {}) {} + class ParentNotice extends Schema.TaggedClass("DynamicTypesParentNotice")("ParentNotice", {}) {} + class OtherEvent extends Schema.TaggedClass("DynamicTypesOtherEvent")("OtherEvent", {}) {} + class ParentIdle extends Schema.TaggedClass("DynamicTypesParentIdle")("ParentIdle", {}) {} + + const Input = Schema.Struct({ id: Schema.String }) + const ParentEvents = Machine.events(ParentNotice) + const childMachine = Machine.make({ + states: { ChildIdle }, + events: Machine.events(ChildEvent), + input: Input, + parent: Machine.parent(ParentEvents), + initial: (to) => to.ChildIdle().resolve(({ input, target }) => target(new ChildIdle({ id: input.id }))) + }).handle({ + ChildIdle: { + on: { ChildEvent: (to) => to.none } + } + }) + const Child = Machine.childFamily(childMachine) + const voidChildMachine = Machine.make({ + states: { ChildIdle }, + events: Machine.events(), + initial: (to) => to.ChildIdle().resolve(({ target }) => target(new ChildIdle({ id: "void" }))) + }).handle({ ChildIdle: {} }) + const VoidChild = Machine.childFamily(voidChildMachine) + + it("binds one machine type to runtime ids", () => { + expect(Child("p-1")).type.toBe>() + expect(Child("p-1")).type.toBeAssignableTo(Machine.child("p-1", childMachine)) + expect(Child("p-1").id).type.toBe<"p-1">() + expect(Child("p-1").machine).type.toBe() + }) + + it("types statechart-owned dynamic child operations", () => { + Machine.make({ + states: { ParentIdle }, + events: Machine.events(ParentNotice, OtherEvent), + initial: (to) => to.ParentIdle().resolve(({ target }) => target(new ParentIdle({}))) + }).handle({ + ParentIdle: { + invoke: (from) => + from.effect("spawn", ({ children }) => { + expect(children.spawn).type.toBeCallableWith(Child("p-1"), { input: { id: "p-1" } }) + expect(children.spawn).type.not.toBeCallableWith(Child("p-1")) + expect(children.spawn).type.toBeCallableWith(VoidChild("void")) + expect(children.spawn).type.not.toBeCallableWith(VoidChild("void"), { input: { id: "void" } }) + expect(children.spawn).type.not.toBeCallableWith( + Child("erased") as Machine.ChildMachine.Any, + { input: { id: "erased" } } + ) + expect(children.sendTo).type.toBeCallableWith(Child("p-1"), new ChildEvent({})) + expect(children.sendTo).type.not.toBeCallableWith(Child("p-1"), new ParentNotice({})) + expect(children.stop).type.toBeCallableWith(Child("p-1")) + + const spawned = children.spawn(Child("p-1"), { input: { id: "p-1" } }) + expect>().type.toBe>>() + expect>().type.toBe< + Machine.ChildAlreadyExistsError | Machine.ChildMachine.StartError> + >() + return spawned + }).onDone((to) => to.none).onFailure((to) => to.none) + } + }) + }) + + it("rejects a child whose parent protocol is not accepted", () => { + Machine.make({ + states: { ParentIdle }, + events: Machine.events(OtherEvent), + initial: (to) => to.ParentIdle().resolve(({ target }) => target(new ParentIdle({}))) + }).handle({ + ParentIdle: { + invoke: (from) => + from.effect("incompatible", ({ children }) => { + expect(children.spawn).type.not.toBeCallableWith(Child("p-1"), { input: { id: "p-1" } }) + return Effect.void + }).onDone((to) => to.none) + } + }) + }) + + it("adds child descriptors to process spawn interfaces", () => { + Machine.logic({ + initial: ({ spawn }) => { + expect(spawn).type.toBeCallableWith(Child("p-1"), { input: { id: "p-1" } }) + return Effect.succeed(undefined) + }, + run: () => Effect.never + }) + + Machine.logic({ + initial: ({ spawn }) => { + expect(spawn).type.not.toBeCallableWith(Child("p-1"), { input: { id: "p-1" } }) + return Effect.succeed(undefined) + }, + run: () => Effect.never + }) + }) +})