From edb9652a8f310de184af651ea87ead27397f0fb3 Mon Sep 17 00:00:00 2001 From: SandroMaglione Date: Sat, 15 Aug 2026 18:39:16 +0200 Subject: [PATCH] Add prepared machine lifecycle and bound invocation typing --- .changeset/calm-machines-prepare.md | 16 + README.md | 47 +- docs/agent-guide.md | 58 +- scripts/fixtures/consumer/deep-bound.ts | 9 +- src/Machine.ts | 714 ++++++++++++++++-- src/internal/machine/atom.ts | 102 ++- src/internal/machine/machine.ts | 6 + src/internal/machine/process.ts | 57 ++ src/internal/machine/runtime.ts | 215 ++++-- src/unstable/reactivity/AtomMachine.ts | 6 +- .../machine/strategyDifferential.test.ts | 39 +- .../machine/support/strategyDifferential.ts | 11 + test/machine/ActorEvents.test.ts | 137 ++++ test/unstable/reactivity/AtomMachine.test.ts | 32 + typetest/machine/ActorEvents.tst.ts | 67 ++ 15 files changed, 1349 insertions(+), 167 deletions(-) create mode 100644 .changeset/calm-machines-prepare.md diff --git a/.changeset/calm-machines-prepare.md b/.changeset/calm-machines-prepare.md new file mode 100644 index 0000000..1109f43 --- /dev/null +++ b/.changeset/calm-machines-prepare.md @@ -0,0 +1,16 @@ +--- +"@typeonce/effect-machine": minor +--- + +Add `Machine.prepare` for composing snapshot and emission streams before a machine initializes, while keeping `Machine.start` as the one-step convenience. + +```ts +const prepared = yield * Machine.prepare(machine) +yield * prepared.emissions.pipe( + Stream.runForEach(handleEmission), + Effect.forkScoped({ startImmediately: true }) +) +const ref = yield * prepared.start +``` + +AtomMachine emission streams use the same preparation boundary, and machine definitions now expose `definition.invoke(...)` so invocation `self` and `parent` references use the exact public input and `parentEvents` protocols. diff --git a/README.md b/README.md index 63ef387..dde42af 100644 --- a/README.md +++ b/README.md @@ -188,14 +188,24 @@ object to `send` or `Machine.plan` for those events. `ref.emissions` is a hot `Stream`: it publishes only notifications produced after subscription, replays nothing, and completes when the actor terminates. Snapshots remain separate and stateful: `ref.changes` begins with the current -lifecycle snapshot and then follows later changes. Because `Machine.start` -returns only after initialization, startup emissions are not observable from -the returned ref; represent startup facts in state when they must be retained. +lifecycle snapshot and then follows later changes. Use `Machine.prepare` when +an observer must be installed before initial-entry actions run: ```ts -const next = ref.emissions.pipe(Stream.take(1), Stream.runHead) +const prepared = yield * Machine.prepare(machine) + +yield * prepared.emissions.pipe( + Stream.runForEach(handleEmission), + Effect.forkScoped({ startImmediately: true }) +) + +const ref = yield * prepared.start ``` +`Machine.start(machine)` remains the one-step convenience for callers that do +not observe startup emissions. Preparation does not retain or replay an +emission: the observer is simply subscribed before initialization begins. + Invalid event and emission constructions fail the machine with a typed `MachineSchemaDecodeError`; they do not throw from the constructor call. @@ -322,6 +332,35 @@ invoke: Machine.invoke({ }) ``` +The standalone `Machine.invoke(...)` constructor does not know the owning +definition, so its `self` and `parent` references are non-sendable. When an +invocation callback sends through either reference, construct it through the +owning definition so those references use its exact public input and +`parentEvents` protocols: + +```ts +const definition = Machine.make({ + events: Commands, + internalEvents: InternalEvents, + parentEvents: ParentEvents + // ... +}) + +const machine = definition.handle({ + Saving: { + invoke: definition.invoke({ + id: "notify-parent", + effect: ({ parent }) => + parent === undefined + ? Effect.void + : parent.send(ParentEvents.SaveStarted()), + onDone: ({ target }) => target.none(), + onFailure: ({ target }) => target.none() + }) + } +}) +``` + A direct `invoke: { ... }` object is also supported when its lifecycle handlers do not need source-derived context. Reuse one exported `Machine.child(id, machine)` descriptor for invocation, `sendTo`, and child diff --git a/docs/agent-guide.md b/docs/agent-guide.md index f8f5624..96ea0bd 100644 --- a/docs/agent-guide.md +++ b/docs/agent-guide.md @@ -105,9 +105,10 @@ its extra control is required: `AtomMachine.resume(machine, snapshot)` for service-free machines. - Use one invocation object: `effect` for one-shot work, `after` for a timer, `logic` for reusable process logic, and `child` for a complete child - statechart. `Machine.invoke({...})` preserves owner context and source - channels across sibling lifecycle handlers. Use a direct object only when its - lifecycle handlers do not need source-derived context. + statechart. `Machine.invoke({...})` preserves owner state and source channels + across sibling lifecycle handlers. Use `definition.invoke({...})` when a + callback uses `self` or `parent`; the bound constructor preserves the + definition's exact public input and `parentEvents` protocols. - Use `Machine.child(id, machine)` for a complete statechart descriptor and `Machine.childAddress(id)` for a low-level process address. A logic invocation is addressable only when `Machine.invoke` receives that @@ -517,8 +518,20 @@ targets an actor mailbox and is processed later. `enqueue.emit(...)` is neither: it publishes a one-off outward notification. Observe it with `ref.emissions`, a hot non-replayed `Stream` that completes with the actor. `ref.changes` is stateful and begins with the current lifecycle snapshot. -Startup emissions occur before `Machine.start` returns and therefore are not -visible through the returned ref; use state for facts that must be retained. +Use `Machine.prepare(machine)` to obtain `changes` and `emissions` before +initialization. Subscribe to the desired stream and then evaluate +`prepared.start`. `Machine.start(machine)` remains the one-step convenience +when startup observation is unnecessary. Emissions are still never retained or +replayed; state remains the representation for facts that must be retained. + +```ts +const prepared = yield* Machine.prepare(machine) +yield* prepared.emissions.pipe( + Stream.runForEach(handleEmission), + Effect.forkScoped({ startImmediately: true }) +) +const ref = yield* prepared.start +``` For child-to-parent input, export a public builder protocol and reuse it at both composition boundaries: @@ -682,6 +695,33 @@ invoke: Machine.invoke({ }) ``` +The standalone constructor cannot know the owning machine's input protocols, +so its `self` and `parent` references are non-sendable. When a source sends +through either reference, use the owning definition's bound constructor: + +```ts +const definition = Machine.make({ + events: Commands, + internalEvents: InternalEvents, + parentEvents: ParentEvents, + // ... +}) + +const machine = definition.handle({ + Saving: { + invoke: definition.invoke({ + id: "notify-parent", + effect: ({ parent }) => + parent === undefined + ? Effect.void + : parent.send(ParentEvents.SaveStarted()), + onDone: ({ target }) => target.none(), + onFailure: ({ target }) => target.none() + }) + } +}) +``` + A direct `invoke: { ... }` object remains available when lifecycle handlers do not need source-derived context. @@ -729,9 +769,11 @@ parentRef.child(Editor) parentAtom.child(Editor) ``` -Child emissions are delivered through the parent's internal protocol. -`onSnapshot`, `onDone`, and `onFailure` are direct parent transitions. Invoked -child IDs must be unique while simultaneously active. +Child emissions remain on the child's hot `emissions` stream; they are never +delivered implicitly to the parent. A child sends an input explicitly with +`enqueue.sendTo(parent, ParentEvents.Example())`. `onSnapshot`, `onDone`, and +`onFailure` are direct parent transitions. Invoked child IDs must be unique +while simultaneously active. Descriptors with the same id and machine identity address the same child, even when independently constructed. The descriptor objects themselves are not diff --git a/scripts/fixtures/consumer/deep-bound.ts b/scripts/fixtures/consumer/deep-bound.ts index b8a17b5..97d29cd 100644 --- a/scripts/fixtures/consumer/deep-bound.ts +++ b/scripts/fixtures/consumer/deep-bound.ts @@ -86,16 +86,17 @@ const States = Machine.defineStates({ }) const Emissions = Machine.emittedEvents(Emitted.cases.Notice) -const machine = Machine.make({ +const definition = Machine.make({ states: States.states, events: Machine.events(Event.cases.Begin, Event.cases.Save, ChildParentEvents), internalEvents: Machine.internalEvents(Internal.cases.Loaded, Internal.cases.ChildCompleted), emittedEvents: Emissions, input: Schema.Struct({ seed: Schema.String }), initial: ({ seed: _seed }) => States.initial.Idle(State.cases.Idle.make({})) -}).handle({ +}) +const machine = definition.handle({ Idle: { - invoke: Machine.invoke({ + invoke: definition.invoke({ id: "deep-inline-invoke", effect: Effect.asVoid(ExternalService), onDone: ({ target }) => target.none() @@ -121,7 +122,7 @@ const machine = Machine.make({ } }, Saving: { - invoke: Machine.invoke({ + invoke: definition.invoke({ child: Child, input: ({ state }) => ({ value: state.value }), onDone: ({ target }) => target.none() diff --git a/src/Machine.ts b/src/Machine.ts index 18ebe6f..ca2e0c7 100644 --- a/src/Machine.ts +++ b/src/Machine.ts @@ -246,6 +246,15 @@ export interface Machine< ParentEvents > + /** + * Preserves invocation inference with this machine's public input and parent + * protocols. Prefer this bound constructor when an invocation source or + * lifecycle handler uses `self` or `parent`. + * + * @since 0.11.0 + */ + readonly invoke: Invoker + /** @internal */ readonly initial: (...args: [...Machine.InputArgs]) => Machine.InitialResult } @@ -1694,6 +1703,33 @@ export type RuntimeOutcome = readonly snapshot: Extract, { readonly status: "stopped" }> } +/** + * A fresh machine whose observable streams exist before initialization. + * + * Subscribe to `emissions` before evaluating `start` when initial-entry + * emissions must be observed. `start` is one-shot: concurrent and repeated + * evaluations share the same initialization and running reference. + * + * @category models + * @since 0.11.0 + */ +export interface Prepared { + /** Stable machine definition id, or a generated fallback when none was declared. */ + readonly id: string + + /** Unique identity reserved for this prepared machine instance. */ + readonly sessionId: string + + /** Waits for startup, then streams the initial lifecycle snapshot and later changes. */ + readonly changes: Stream.Stream, StartError> + + /** Streams ephemeral notifications published after subscription. */ + readonly emissions: Stream.Stream + + /** Initializes this machine once and returns its running reference. */ + readonly start: Effect.Effect, StartError, StartRequirements> +} + /** * Provides access to a running machine's state, lifecycle, event input, and * termination operations. @@ -2128,6 +2164,8 @@ export declare namespace Machine { /** @internal */ readonly handle: any /** @internal */ + readonly invoke: any + /** @internal */ readonly initial: any } @@ -4616,13 +4654,18 @@ export declare namespace Machine { States extends StateSchemas, Events extends ReadonlyArray, Emits extends ReadonlyArray, - StateId extends StateIdentifier + StateId extends StateIdentifier, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = & InvokeOwned & ( | { readonly id: string - readonly effect: InvokeSource, InvokeContext> + readonly effect: InvokeSource< + Effect.Effect, + InvokeContext + > readonly after?: never readonly logic?: never readonly child?: never @@ -4633,7 +4676,10 @@ export declare namespace Machine { } | { readonly id: string - readonly after: InvokeSource> + readonly after: InvokeSource< + Duration.Input, + InvokeContext + > readonly effect?: never readonly logic?: never readonly child?: never @@ -4645,7 +4691,10 @@ export declare namespace Machine { | { readonly id: string readonly address: string - readonly logic: InvokeSource> + readonly logic: InvokeSource< + AnyLogicSource, + InvokeContext + > readonly effect?: never readonly after?: never readonly child?: never @@ -4655,7 +4704,10 @@ export declare namespace Machine { } | { readonly child: ChildMachine.Any - readonly input?: {} | null | ((context: InvokeContext) => unknown) + readonly input?: + | {} + | null + | ((context: InvokeContext) => unknown) readonly id?: never readonly address?: never readonly effect?: never @@ -4672,10 +4724,16 @@ export declare namespace Machine { States extends StateSchemas, Events extends ReadonlyArray, Emits extends ReadonlyArray, - StateId extends StateIdentifier + StateId extends StateIdentifier, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = - | InvokeConfig - | ReadonlyArray> + | InvokeConfig + | ReadonlyArray> + + type TypedInvokeDefinition = + | InvokeTyped + | ReadonlyArray> type InvokeHandlerRequirement = IsAny extends true ? { readonly handler: Handler } : [Value] extends [never] ? { readonly handler?: never } @@ -4697,7 +4755,9 @@ export declare namespace Machine { Emits extends ReadonlyArray, StateId extends StateIdentifier, Fx extends Effect.Effect, - Source = Fx + Source = Fx, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = & { readonly id: InvokeLifecycleId @@ -4714,7 +4774,15 @@ export declare namespace Machine { States, Events, Emits, - InvokeDoneContext>> + InvokeDoneContext< + States, + Events, + Emits, + StateId, + Effect.Success>, + InputEvents, + ParentEvents + > > > & InvokeFailureRequirement< @@ -4723,7 +4791,15 @@ export declare namespace Machine { States, Events, Emits, - InvokeFailureContext>> + InvokeFailureContext< + States, + Events, + Emits, + StateId, + Effect.Error>, + InputEvents, + ParentEvents + > > > @@ -4731,10 +4807,15 @@ export declare namespace Machine { States extends StateSchemas, Events extends ReadonlyArray, Emits extends ReadonlyArray, - StateId extends StateIdentifier + StateId extends StateIdentifier, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = { readonly id: InvokeLifecycleId - readonly after: InvokeSource> + readonly after: InvokeSource< + Duration.Input, + InvokeContext + > readonly effect?: never readonly logic?: never readonly child?: never @@ -4745,7 +4826,7 @@ export declare namespace Machine { States, Events, Emits, - InvokeDoneContext + InvokeDoneContext > } @@ -4761,7 +4842,9 @@ export declare namespace Machine { ChildOutput, ChildInitialError, Address extends ChildAddress, - Source = Logic + Source = Logic, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = & { readonly id: InvokeLifecycleId @@ -4774,16 +4857,36 @@ export declare namespace Machine { States, Events, Emits, - InvokeSnapshotContext + InvokeSnapshotContext< + States, + Events, + Emits, + StateId, + ChildState, + ChildError, + ChildOutput, + InputEvents, + ParentEvents + > > } & InvokeDoneRequirement< ChildOutput, - InvokeTransition> + InvokeTransition< + States, + Events, + Emits, + InvokeDoneContext + > > & InvokeFailureRequirement< ChildError, - InvokeTransition> + InvokeTransition< + States, + Events, + Emits, + InvokeFailureContext + > > export type ChildInvokeArgs< @@ -4792,7 +4895,9 @@ export declare namespace Machine { Emits extends ReadonlyArray, StateId extends StateIdentifier, ChildDefinition extends Machine.Any, - Child extends ChildMachine + Child extends ChildMachine, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = & { readonly child: @@ -4819,12 +4924,17 @@ export declare namespace Machine { StateId, Snapshot>, Error, - Output + Output, + InputEvents, + ParentEvents > > } & (Input extends typeof Schema.Void ? { readonly input?: never } : { - readonly input: InvokeSource["Type"], InvokeContext> + readonly input: InvokeSource< + Input["Type"], + InvokeContext + > }) & InvokeDoneRequirement< Output, @@ -4832,7 +4942,15 @@ export declare namespace Machine { States, Events, Emits, - InvokeDoneContext> + InvokeDoneContext< + States, + Events, + Emits, + StateId, + Output, + InputEvents, + ParentEvents + > > > & InvokeFailureRequirement< @@ -4846,7 +4964,9 @@ export declare namespace Machine { Events, Emits, StateId, - Error | ActionError> + Error | ActionError>, + InputEvents, + ParentEvents > > > @@ -4863,6 +4983,8 @@ export declare namespace Machine { Events extends ReadonlyArray, Emits extends ReadonlyArray, StateId extends StateIdentifier, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, Raw > = Raw extends InvokeTyped ? unknown : Raw extends { readonly effect: infer Source } ? @@ -4873,7 +4995,7 @@ export declare namespace Machine { States, Events, Emits, - InvokeDoneContext> + InvokeDoneContext, InputEvents, ParentEvents> > > & InvokeFailureRequirement< @@ -4882,7 +5004,7 @@ export declare namespace Machine { States, Events, Emits, - InvokeFailureContext> + InvokeFailureContext, InputEvents, ParentEvents> > > & { readonly onSnapshot?: never } @@ -4892,7 +5014,7 @@ export declare namespace Machine { States, Events, Emits, - InvokeDoneContext + InvokeDoneContext > readonly onFailure?: never readonly onSnapshot?: never @@ -4905,7 +5027,7 @@ export declare namespace Machine { States, Events, Emits, - InvokeDoneContext> + InvokeDoneContext, InputEvents, ParentEvents> > > & InvokeFailureRequirement< @@ -4914,7 +5036,15 @@ export declare namespace Machine { States, Events, Emits, - InvokeFailureContext> + InvokeFailureContext< + States, + Events, + Emits, + StateId, + InvokeRuntimeError, + InputEvents, + ParentEvents + > > > & { @@ -4935,7 +5065,9 @@ export declare namespace Machine { StateId, ChildLogic extends Logic ? ChildState : never, InvokeRuntimeError, - InvokeOutput + InvokeOutput, + InputEvents, + ParentEvents > > } @@ -4948,7 +5080,15 @@ export declare namespace Machine { States, Events, Emits, - InvokeDoneContext> + InvokeDoneContext< + States, + Events, + Emits, + StateId, + Output, + InputEvents, + ParentEvents + > > > & InvokeFailureRequirement< @@ -4962,12 +5102,17 @@ export declare namespace Machine { Events, Emits, StateId, - Error | ActionError> + Error | ActionError>, + InputEvents, + ParentEvents > > > & (Input extends typeof Schema.Void ? { readonly input?: never } : { - readonly input: InvokeSource["Type"], InvokeContext> + readonly input: InvokeSource< + Input["Type"], + InvokeContext + > }) & { readonly onSnapshot?: InvokeTransition< @@ -4981,7 +5126,9 @@ export declare namespace Machine { StateId, Snapshot>, Error, - Output + Output, + InputEvents, + ParentEvents > > } @@ -4993,11 +5140,21 @@ export declare namespace Machine { Events extends ReadonlyArray, Emits extends ReadonlyArray, StateId extends StateIdentifier, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, Raw > = Raw extends ReadonlyArray ? { - readonly [Index in keyof Raw]: ContextualInvokeConfig + readonly [Index in keyof Raw]: ContextualInvokeConfig< + States, + Events, + Emits, + StateId, + InputEvents, + ParentEvents, + Raw[Index] + > } - : ContextualInvokeConfig + : ContextualInvokeConfig type OutputHandlerConfig< States extends StateSchemas, @@ -5049,7 +5206,9 @@ export declare namespace Machine { context: StateActionContext, enqueue: Enqueue, EmitOf> ) => StateActionResult - readonly invoke?: InvokeDefinition + readonly invoke?: + | InvokeDefinition + | TypedInvokeDefinition readonly always?: | (( context: AlwaysContext, @@ -5300,14 +5459,26 @@ export declare namespace Machine { type HandlerInvokeContextAtPath< AllStates extends StateSchemas, Events extends ReadonlyArray, + InputEvents extends ReadonlyArray, Emits extends ReadonlyArray, + ParentEvents extends ReadonlyArray, Config, StateId extends StateNodeIdentifier, NodeConfig = HandlerConfigAtPath > = StateId extends StateIdentifier ? NodeConfig extends { readonly invoke: infer Invoke } ? HandlerValidationAtPath< StateId, - { readonly invoke: ContextualInvokeDefinition } + { + readonly invoke: ContextualInvokeDefinition< + AllStates, + Events, + Emits, + StateId, + InputEvents, + ParentEvents, + Invoke + > + } > : unknown : unknown @@ -5315,14 +5486,18 @@ export declare namespace Machine { type HandlerInvokeContexts< AllStates extends StateSchemas, Events extends ReadonlyArray, + InputEvents extends ReadonlyArray, Emits extends ReadonlyArray, + ParentEvents extends ReadonlyArray, Config > = Types.UnionToIntersection< StateNodeIdentifier extends infer StateId extends StateNodeIdentifier ? StateId extends StateNodeIdentifier ? HandlerInvokeContextAtPath< AllStates, Events, + InputEvents, Emits, + ParentEvents, Config, StateId > @@ -5919,7 +6094,7 @@ export declare namespace Machine { >( config: & Config - & HandlerInvokeContexts> + & HandlerInvokeContexts> & HandlerTreeValidation< States, Events, @@ -6559,7 +6734,7 @@ type DynamicEffectInvokeSource< Emits extends ReadonlyArray, StateId extends Machine.StateIdentifier, Source extends ( - context: Machine.InvokeContext + context: Machine.InvokeContext ) => Effect.Effect > = { readonly id: InvokeLifecycleId @@ -6581,7 +6756,15 @@ type DynamicEffectDoneHandler< States, Events, Emits, - Machine.InvokeDoneContext>>> + Machine.InvokeDoneContext< + States, + Events, + Emits, + StateId, + Effect.Success>>, + readonly [], + readonly [] + > > type DynamicEffectFailureHandler< @@ -6594,7 +6777,15 @@ type DynamicEffectFailureHandler< States, Events, Emits, - Machine.InvokeFailureContext>>> + Machine.InvokeFailureContext< + States, + Events, + Emits, + StateId, + Effect.Error>>, + readonly [], + readonly [] + > > type DynamicEffectInvokeResult< @@ -6604,7 +6795,7 @@ type DynamicEffectInvokeResult< StateId extends Machine.StateIdentifier, Source extends (...args: ReadonlyArray) => Effect.Effect > = - & Machine.InvokeConfig + & Machine.InvokeConfig & Machine.InvokeTyped< Effect.Success>, Effect.Error>, @@ -6614,6 +6805,325 @@ type DynamicEffectInvokeResult< type InvokeChannelIsNever = IsAny extends true ? false : [Value] extends [never] ? true : false +type BoundDynamicEffectInvokeSource< + States extends Machine.StateSchemas, + Events extends ReadonlyArray, + Emits extends ReadonlyArray, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, + StateId extends Machine.StateIdentifier, + Source extends ( + context: Machine.InvokeContext + ) => Effect.Effect +> = { + readonly id: InvokeLifecycleId + readonly effect: Source + readonly after?: never + readonly logic?: never + readonly child?: never + readonly address?: never + readonly onSnapshot?: never +} + +type BoundDynamicEffectDoneHandler< + States extends Machine.StateSchemas, + Events extends ReadonlyArray, + Emits extends ReadonlyArray, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, + StateId extends Machine.StateIdentifier, + Source extends (...args: ReadonlyArray) => Effect.Effect +> = Machine.InvokeTransition< + States, + Events, + Emits, + Machine.InvokeDoneContext< + States, + Events, + Emits, + StateId, + Effect.Success>>, + InputEvents, + ParentEvents + > +> + +type BoundDynamicEffectFailureHandler< + States extends Machine.StateSchemas, + Events extends ReadonlyArray, + Emits extends ReadonlyArray, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, + StateId extends Machine.StateIdentifier, + Source extends (...args: ReadonlyArray) => Effect.Effect +> = Machine.InvokeTransition< + States, + Events, + Emits, + Machine.InvokeFailureContext< + States, + Events, + Emits, + StateId, + Effect.Error>>, + InputEvents, + ParentEvents + > +> + +type BoundDynamicEffectInvokeResult< + States extends Machine.StateSchemas, + Events extends ReadonlyArray, + Emits extends ReadonlyArray, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, + StateId extends Machine.StateIdentifier, + Source extends (...args: ReadonlyArray) => Effect.Effect +> = + & Machine.InvokeConfig + & Machine.InvokeTyped< + Effect.Success>, + Effect.Error>, + Effect.Services>, + never + > + +/** + * Machine-bound invocation constructor that preserves the owning machine's + * public input and parent protocols in every invocation callback. + * + * @category constructors + * @since 0.11.0 + */ +export interface Invoker< + States extends Machine.StateSchemas, + Events extends ReadonlyArray, + Emits extends ReadonlyArray, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray +> { + < + StateId extends Machine.StateIdentifier, + const Source extends ( + context: Machine.InvokeContext + ) => Effect.Effect + >( + config: + & BoundDynamicEffectInvokeSource + & { + readonly onDone: BoundDynamicEffectDoneHandler< + States, + Events, + Emits, + InputEvents, + ParentEvents, + StateId, + Source + > + readonly onFailure: BoundDynamicEffectFailureHandler< + States, + Events, + Emits, + InputEvents, + ParentEvents, + StateId, + Source + > + }, + ..._validation: InvokeChannelIsNever>> extends true ? [ + "onDone must be omitted when the Effect output is never" + ] + : InvokeChannelIsNever>> extends true ? [ + "onFailure must be omitted when the Effect error is never" + ] + : [] + ): BoundDynamicEffectInvokeResult + < + StateId extends Machine.StateIdentifier, + const Source extends ( + context: Machine.InvokeContext + ) => Effect.Effect + >( + config: + & BoundDynamicEffectInvokeSource + & { + readonly onDone: BoundDynamicEffectDoneHandler< + States, + Events, + Emits, + InputEvents, + ParentEvents, + StateId, + Source + > + readonly onFailure?: never + }, + ..._validation: InvokeChannelIsNever>> extends true ? [ + "onDone must be omitted when the Effect output is never" + ] + : [] + ): BoundDynamicEffectInvokeResult + < + StateId extends Machine.StateIdentifier, + const Source extends ( + context: Machine.InvokeContext + ) => Effect.Effect + >( + config: + & BoundDynamicEffectInvokeSource + & { + readonly onDone?: never + readonly onFailure: BoundDynamicEffectFailureHandler< + States, + Events, + Emits, + InputEvents, + ParentEvents, + StateId, + Source + > + }, + ..._validation: InvokeChannelIsNever>> extends true ? [ + "onFailure must be omitted when the Effect error is never" + ] + : [] + ): BoundDynamicEffectInvokeResult + < + StateId extends Machine.StateIdentifier, + const Source extends ( + context: Machine.InvokeContext + ) => Effect.Effect + >( + config: + & BoundDynamicEffectInvokeSource + & { + readonly onDone?: never + readonly onFailure?: never + } + ): BoundDynamicEffectInvokeResult + < + StateId extends Machine.StateIdentifier, + const Fx extends Effect.Effect, + const Config extends object + >( + config: + & Config + & Machine.EffectInvokeArgs< + States, + Events, + Emits, + StateId, + Fx, + Fx, + InputEvents, + ParentEvents + > + ): + & Config + & Machine.InvokeOwned + & Machine.InvokeTyped, Effect.Error, Effect.Services, never> + , const Config extends object>( + config: Config & Machine.TimerInvokeArgs + ): Config & Machine.InvokeOwned & Machine.InvokeTyped + < + StateId extends Machine.StateIdentifier, + ChildState, + ChildEvent, + ChildError, + ChildRequirements, + ChildOutput, + ChildInitialError, + Address extends ChildAddress + >( + config: Machine.LogicInvokeArgs< + States, + Events, + Emits, + StateId, + ChildState, + ChildEvent, + ChildError, + ChildRequirements, + ChildOutput, + ChildInitialError, + Address, + (context: Machine.InvokeContext) => Logic< + ChildState, + ChildEvent, + ChildError, + ChildRequirements, + ChildOutput, + ChildInitialError + >, + InputEvents, + ParentEvents + > + ): + & Machine.InvokeConfig + & Machine.InvokeTyped + < + StateId extends Machine.StateIdentifier, + ChildState, + ChildEvent, + ChildError, + ChildRequirements, + ChildOutput, + ChildInitialError, + Address extends ChildAddress, + const Config extends object + >( + config: + & Config + & Machine.LogicInvokeArgs< + States, + Events, + Emits, + StateId, + ChildState, + ChildEvent, + ChildError, + ChildRequirements, + ChildOutput, + ChildInitialError, + Address, + Logic, + InputEvents, + ParentEvents + > + ): + & Config + & Machine.InvokeOwned + & Machine.InvokeTyped + < + StateId extends Machine.StateIdentifier, + const Child extends ChildMachine.Any, + const Config extends object + >( + config: + & Config + & Machine.ChildInvokeArgs< + States, + Events, + Emits, + StateId, + Child["machine"], + Child, + InputEvents, + ParentEvents + > + ): + & Config + & Machine.InvokeOwned + & Machine.InvokeTyped< + Machine.Output, + Machine.Error | ActionError>, + Machine.Services, + Machine.InitialError, + Machine.Emit, + Machine.EventOf> + > +} + /** * Preserves inference for a state-owned invocation configuration. * @@ -6630,6 +7140,12 @@ type InvokeChannelIsNever = IsAny extends true ? false : [Value] e * lifecycle `id` and a typed communication `address`. Child descriptors already * own their identity, so `id` and `address` must not be repeated. * + * Owner references are intentionally non-sendable here because this standalone + * constructor does not know the owning definition. When a callback sends to + * `self` or `parent`, use the bound constructor on the owning definition + * (`definition.invoke(...)`) so the exact public input and `parentEvents` + * protocols are available. + * * ```ts * invoke: Machine.invoke({ * id: "load", @@ -6652,7 +7168,7 @@ export const invoke: { const Emits extends ReadonlyArray, StateId extends Machine.StateIdentifier, const Source extends ( - context: Machine.InvokeContext + context: Machine.InvokeContext ) => Effect.Effect >( config: @@ -6675,7 +7191,7 @@ export const invoke: { const Emits extends ReadonlyArray, StateId extends Machine.StateIdentifier, const Source extends ( - context: Machine.InvokeContext + context: Machine.InvokeContext ) => Effect.Effect >( config: @@ -6695,7 +7211,7 @@ export const invoke: { const Emits extends ReadonlyArray, StateId extends Machine.StateIdentifier, const Source extends ( - context: Machine.InvokeContext + context: Machine.InvokeContext ) => Effect.Effect >( config: @@ -6715,7 +7231,7 @@ export const invoke: { const Emits extends ReadonlyArray, StateId extends Machine.StateIdentifier, const Source extends ( - context: Machine.InvokeContext + context: Machine.InvokeContext ) => Effect.Effect >( config: @@ -6733,7 +7249,9 @@ export const invoke: { const Fx extends Effect.Effect, const Config extends object >( - config: Config & Machine.EffectInvokeArgs + config: + & Config + & Machine.EffectInvokeArgs ): & Config & Machine.InvokeOwned @@ -6750,7 +7268,7 @@ export const invoke: { StateId extends Machine.StateIdentifier, const Config extends object >( - config: Config & Machine.TimerInvokeArgs + config: Config & Machine.TimerInvokeArgs ): Config & Machine.InvokeOwned & Machine.InvokeTyped < const States extends Machine.StateSchemas, @@ -6777,17 +7295,19 @@ export const invoke: { ChildOutput, ChildInitialError, Address, - (context: Machine.InvokeContext) => Logic< + (context: Machine.InvokeContext) => Logic< ChildState, ChildEvent, ChildError, ChildRequirements, ChildOutput, ChildInitialError - > + >, + readonly [], + readonly [] > ): - & Machine.InvokeConfig + & Machine.InvokeConfig & Machine.InvokeTyped< ChildOutput, ChildError, @@ -6821,7 +7341,10 @@ export const invoke: { ChildRequirements, ChildOutput, ChildInitialError, - Address + Address, + Logic, + readonly [], + readonly [] > ): & Config @@ -6840,7 +7363,9 @@ export const invoke: { const Child extends ChildMachine.Any, const Config extends object >( - config: Config & Machine.ChildInvokeArgs + config: + & Config + & Machine.ChildInvokeArgs ): & Config & Machine.InvokeOwned @@ -7410,6 +7935,89 @@ export const watch: ( ref: MachineRef ) => Stream.Stream> = internal.watch +/** + * Prepares a fresh machine without initializing it. + * + * Use this constructor when observation must be composed before initial-entry + * actions run. Subscribe to `prepared.emissions`, then evaluate + * `prepared.start`. Ordinary callers can continue to use {@link start}, which + * starts directly and does not allocate the prepared lifecycle boundary. + * + * ```ts + * const prepared = yield* Machine.prepare(machine) + * + * yield* prepared.emissions.pipe( + * Stream.runForEach(handleEmission), + * Effect.forkScoped({ startImmediately: true }) + * ) + * + * const ref = yield* prepared.start + * ``` + * + * @category constructors + * @since 0.11.0 + */ +export const prepare: < + const States extends Machine.StateSchemas, + const Events extends ReadonlyArray, + const Emits extends ReadonlyArray = readonly [], + const Input extends Schema.Top = typeof Schema.Void, + UnhandledStates extends Machine.StateIdentifier = Machine.StateIdentifier, + E = never, + R = never, + InitialE = never, + InitialR = never, + FinalStates extends Machine.StateIdentifier = never, + Output = never, + OutputStates extends Machine.StateIdentifier = never, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] +>( + machine: + & Machine< + States, + Events, + Input, + UnhandledStates, + E, + R, + InitialE, + InitialR, + FinalStates, + Output, + Emits, + OutputStates, + InputEvents, + ParentEvents + > + & EnsureExecutable, + ...args: [...Machine.InputArgs] +) => Effect.Effect< + Prepared< + Machine.Snapshot, + Machine.EventInputOf, + | E + | ActionError + | InfiniteTransitionError + | MachineSchemaDecodeError + | StoppedError, + Output, + Machine.EmittedEventOf, + | InitialE + | E + | ActionError + | InfiniteTransitionError + | MachineSchemaDecodeError + | StartupError + | StoppedError, + ExcludeCompatibleRuntime< + ExecutionServices, + Machine.EventOf, + Machine.EmitOf + > + > +> = internal.prepare as any + /** * Starts a machine. * diff --git a/src/internal/machine/atom.ts b/src/internal/machine/atom.ts index 682e65b..8549989 100644 --- a/src/internal/machine/atom.ts +++ b/src/internal/machine/atom.ts @@ -7,11 +7,12 @@ import * as Data from "effect/Data" import * as Effect from "effect/Effect" import * as Equal from "effect/Equal" +import * as Fiber from "effect/Fiber" import * as Option from "effect/Option" import type * as Schema from "effect/Schema" import type * as Scope from "effect/Scope" import * as Stream from "effect/Stream" -import { AsyncResult, Atom, type AtomRegistry } from "effect/unstable/reactivity" +import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity" import type * as Machine from "../../Machine.js" import type { Bound, ChildMachineAtom, MachineAtom } from "../../unstable/reactivity/AtomMachine.js" import * as internalMachine from "./machine.js" @@ -72,6 +73,11 @@ type MachineStartError = | Machine.StoppedError | RuntimeError +const preparedByMachineAtom = new WeakMap< + object, + Atom.Atom, any>> +>() + const runMachineAtomEffect = ( get: Atom.AtomContext, start: Effect.Effect, StartError, Requirements> @@ -83,53 +89,16 @@ const runMachineAtomEffect = , - const Emits extends ReadonlyArray = any, - const Input extends Schema.Top = typeof Schema.Void, - UnhandledStates extends Machine.Machine.StateIdentifier = Machine.Machine.StateIdentifier, - E = never, - R = never, - InitialE = never, - InitialR = never, - FinalStates extends Machine.Machine.StateIdentifier = never, - Output = never, - OutputStates extends Machine.Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events, - ParentEvents extends ReadonlyArray = readonly [] ->( +const startPreparedMachineAtomEffect = ( get: Atom.AtomContext, - machine: - & Machine.Machine< - States, - Events, - Input, - UnhandledStates, - E, - R, - InitialE, - InitialR, - FinalStates, - Output, - Emits, - OutputStates, - InputEvents, - ParentEvents + prepared: Atom.Atom< + AsyncResult.AsyncResult< + Machine.Prepared, + never > - & EnsureExecutable, - args: [...Machine.Machine.InputArgs] -): Effect.Effect< - Machine.MachineRef< - Machine.Machine.Snapshot, - Machine.Machine.EventInputOf, - MachineRuntimeError, - Output, - Machine.Machine.EmittedEventOf - >, - MachineStartError, - MachineRequirements, Machine.Machine.EmitOf> -> => runMachineAtomEffect(get, internalMachine.start(machine as any, ...args) as any) + > +): Effect.Effect => + runMachineAtomEffect(get, get.result(prepared).pipe(Effect.flatMap((prepared) => prepared.start))) const resumeMachineAtomEffect = ( get: Atom.AtomContext, @@ -144,8 +113,31 @@ type RefEmitted = Ref extends Machine.MachineRef( self: MachineAtom -): Stream.Stream => - Atom.toStreamResult(self.ref).pipe(Stream.flatMap((ref) => ref.emissions)) +): Stream.Stream => { + const prepared = preparedByMachineAtom.get(self as object) + if (prepared === undefined) { + return Atom.toStreamResult(self.ref).pipe(Stream.flatMap((ref) => ref.emissions)) + } + return Stream.unwrap( + Effect.gen(function*() { + const registry = yield* AtomRegistry.AtomRegistry + const releasePrepared = yield* Effect.sync(() => registry.mount(prepared)) + yield* Effect.addFinalizer(() => Effect.sync(releasePrepared)) + const machine = yield* Atom.getResult(prepared) + const pull = yield* Stream.toPull(machine.emissions as Stream.Stream) + const firstPull = yield* pull.pipe(Effect.forkScoped({ startImmediately: true })) + const releaseRef = yield* Effect.sync(() => registry.mount(self.ref)) + yield* Effect.addFinalizer(() => Effect.sync(releaseRef)) + yield* Atom.getResult(self.ref) + let first = true + return Stream.fromPull(Effect.succeed(Effect.suspend(() => { + if (!first) return pull + first = false + return Fiber.join(firstPull) + }))) + }) + ) +} export const childEmissions = ( self: ChildMachineAtom @@ -666,8 +658,11 @@ export const make: { Machine.Machine.EmittedEventOf > } = ((machine: Machine.Machine.Any, ...args: ReadonlyArray) => { - const ref = Atom.make((get) => startMachineAtomEffect(get, machine as any, args as [])) - return makeFromRefAtom(ref as any) + const prepared = Atom.make(() => internalMachine.prepare(machine as any, ...(args as []))) + const ref = Atom.make((get) => startPreparedMachineAtomEffect(get, prepared as any)) + const result = makeFromRefAtom(ref as any) + preparedByMachineAtom.set(result, prepared as any) + return result }) as any export const resume: { @@ -688,8 +683,11 @@ const makeWithRuntime = ( machine: Machine.Machine.Any, args: ReadonlyArray ): MachineAtom => { - const ref = runtime.atom((get) => startMachineAtomEffect(get, machine as any, args as [])) - return makeFromRefAtom(ref as any) + const prepared = runtime.atom(() => internalMachine.prepare(machine as any, ...(args as []))) + const ref = runtime.atom((get) => startPreparedMachineAtomEffect(get, prepared as any)) + const result = makeFromRefAtom(ref as any) + preparedByMachineAtom.set(result, prepared as any) + return result } const resumeWithRuntime = ( diff --git a/src/internal/machine/machine.ts b/src/internal/machine/machine.ts index 2cb2d49..87386ee 100644 --- a/src/internal/machine/machine.ts +++ b/src/internal/machine/machine.ts @@ -104,6 +104,8 @@ const Proto = { } } +const makeBoundInvoke = (config: unknown): unknown => config + const cloneWithHandlers = ( self: Machine.Any, handlers: Machine.StateConfigs @@ -121,6 +123,7 @@ const cloneWithHandlers = ( machine.makeTargetBuilder = self.makeTargetBuilder machine.handlers = handlers machine.handle = makeHandle(machine) + machine.invoke = makeBoundInvoke Protocol.copyProtocol(self, machine) return machine } @@ -879,6 +882,7 @@ export const make: Make = (< self.makeTargetBuilder = makeTargetBuilder(config.states, self.stateNodes) self.handlers = Object.create(null) self.handle = makeHandle(self) + self.invoke = makeBoundInvoke Protocol.setProtocol(self) return self }) as Make @@ -1358,6 +1362,8 @@ export const watch = ( ref: MachineRef ): Stream.Stream> => internalRuntime.watch(ref) +export const prepare = internalProcess.prepare + export const start: < const States extends Machine.StateSchemas, const Events extends ReadonlyArray, diff --git a/src/internal/machine/process.ts b/src/internal/machine/process.ts index 47072d1..fc4e46f 100644 --- a/src/internal/machine/process.ts +++ b/src/internal/machine/process.ts @@ -825,6 +825,18 @@ export const startWithRuntimeStrategyForTesting = ( machine.id === undefined ? undefined : { id: machine.id } ) +/** @internal Test-only prepared startup strategy selection for a fresh machine. */ +export const prepareWithRuntimeStrategyForTesting = ( + machine: Machine.Any, + strategy: internalRuntime.ProcessRuntimeStrategy, + ...args: ReadonlyArray +): Effect.Effect, any, any> => + internalRuntime.prepareProcessWithStrategyForTesting( + (toProcessLogic as any)(machine, ...args), + strategy, + machine.id === undefined ? undefined : { id: machine.id } + ) + /** @internal Test-only runtime strategy selection for a resumed machine. */ export const resumeWithRuntimeStrategyForTesting = ( machine: Machine.Any, @@ -881,6 +893,51 @@ export const start: < machine.id === undefined ? undefined : { id: machine.id } ) as any +export const prepare: < + const States extends Machine.StateSchemas, + const Events extends ReadonlyArray, + const Emits extends ReadonlyArray = readonly [], + const Input extends Schema.Top = typeof Schema.Void, + UnhandledStates extends Machine.StateIdentifier = Machine.StateIdentifier, + E = never, + R = never, + InitialE = never, + InitialR = never, + FinalStates extends Machine.StateIdentifier = never, + Output = never +>( + machine: Machine, + ...args: [...Machine.InputArgs] +) => Effect.Effect< + internalRuntime.PreparedProcess< + Machine.Snapshot, + Machine.EventOf, + | E + | ActionError + | InfiniteTransitionError + | MachineSchemaDecodeError + | StoppedError, + Output, + Machine.EmittedEventOf, + | InitialE + | E + | ActionError + | InfiniteTransitionError + | MachineSchemaDecodeError + | StartupError + | StoppedError, + ExcludeCompatibleRuntime< + Exclude, internalRuntime.MachineRuntime>, + Machine.EventOf, + Machine.EmitOf + > + > +> = (machine, ...args) => + internalRuntime.prepareProcess( + toProcessLogic(machine, ...args), + machine.id === undefined ? undefined : { id: machine.id } + ) as any + export const resume: < const States extends Machine.StateSchemas, const Events extends ReadonlyArray, diff --git a/src/internal/machine/runtime.ts b/src/internal/machine/runtime.ts index a7c3e03..f6bde59 100644 --- a/src/internal/machine/runtime.ts +++ b/src/internal/machine/runtime.ts @@ -302,6 +302,22 @@ export interface MachineRef Stream.Stream> } +export interface PreparedProcess< + out State, + in Event, + out Error, + out Output, + out Emitted, + out StartError, + StartRequirements +> { + readonly id: string + readonly sessionId: string + readonly changes: Stream.Stream, StartError> + readonly emissions: Stream.Stream + readonly start: Effect.Effect, StartError, StartRequirements> +} + interface ProcessAddress { readonly id: string readonly sessionId: string @@ -609,6 +625,8 @@ const makeProcessRuntime: Effect.Effect = Effect.sync(() => { interface StartInternalOptions { readonly detached?: boolean readonly id?: string + readonly sessionId?: string + readonly emissions?: EmissionRuntime readonly onOutcome?: (outcome: RuntimeOutcome) => Effect.Effect readonly onSnapshot?: ( snapshot: Extract, { readonly status: "active" }> @@ -802,6 +820,54 @@ const EmissionsClosed: unique symbol = Symbol("effect/Machine/EmissionsClosed") type LazyEmissions = PubSub.PubSub | typeof EmissionsClosed | undefined +interface EmissionRuntime { + readonly emit: (event: unknown) => Effect.Effect + readonly close: () => Effect.Effect + readonly stream: Stream.Stream +} + +const makeEmissionRuntime = (): EmissionRuntime => { + let emissions: LazyEmissions + const getOrCreate: Effect.Effect | undefined> = Effect.suspend(() => { + const observed = emissions + if (observed === EmissionsClosed) return Effect.succeed(undefined) + if (observed !== undefined) return Effect.succeed(observed) + return PubSub.unbounded().pipe( + Effect.flatMap((candidate) => + Effect.sync(() => { + const latest = emissions + if (latest === EmissionsClosed) return [undefined, true] as const + if (latest !== undefined) return [latest, true] as const + emissions = candidate + return [candidate, false] as const + }).pipe( + Effect.flatMap(([selected, discard]) => + discard ? PubSub.shutdown(candidate).pipe(Effect.as(selected)) : Effect.succeed(selected) + ) + ) + ) + ) + }) + return { + emit: (event) => + Effect.suspend(() => + emissions === undefined || emissions === EmissionsClosed + ? Effect.void + : PubSub.publish(emissions, event).pipe(Effect.asVoid) + ), + close: () => { + const observed = emissions + emissions = EmissionsClosed + return observed === undefined || observed === EmissionsClosed ? Effect.void : PubSub.shutdown(observed) + }, + stream: Stream.unwrap( + getOrCreate.pipe( + Effect.map((emissions) => emissions === undefined ? Stream.empty : Stream.fromPubSub(emissions)) + ) + ) + } +} + const childlessRuntime: ChildRuntime = { close: () => Effect.void, spawn: (() => Effect.die(new Error("Childless machine logic cannot spawn a process"))) as ProcessSpawn, @@ -1104,41 +1170,10 @@ const startGenericInternal: < | { readonly _tag: "Done"; readonly output: Output } | { readonly _tag: "Failure"; readonly cause: Cause.Cause } - const sessionId = yield* runtime.nextSessionId + const sessionId = options.sessionId ?? (yield* runtime.nextSessionId) const id = requestedId ?? sessionId const queue = yield* Queue.unbounded>() - let emissions: LazyEmissions - const getOrCreateEmissions: Effect.Effect | undefined> = Effect.suspend(() => { - const observed = emissions - if (observed === EmissionsClosed) return Effect.succeed(undefined) - if (observed !== undefined) return Effect.succeed(observed) - return PubSub.unbounded().pipe( - Effect.flatMap((candidate) => - Effect.sync(() => { - const latest = emissions - if (latest === EmissionsClosed) return [undefined, true] as const - if (latest !== undefined) return [latest, true] as const - emissions = candidate - return [candidate, false] as const - }).pipe( - Effect.flatMap(([selected, discard]) => - discard ? PubSub.shutdown(candidate).pipe(Effect.as(selected)) : Effect.succeed(selected) - ) - ) - ) - ) - }) - const closeEmissions = (): Effect.Effect => { - const observed = emissions - emissions = EmissionsClosed - return observed === undefined || observed === EmissionsClosed ? Effect.void : PubSub.shutdown(observed) - } - const emit = (event: unknown): Effect.Effect => - Effect.suspend(() => - emissions === undefined || emissions === EmissionsClosed - ? Effect.void - : PubSub.publish(emissions, event).pipe(Effect.asVoid) - ) + const emissions = options.emissions ?? makeEmissionRuntime() const termination = yield* Deferred.make() const done = yield* Deferred.make() const awaitCompletion = Deferred.await(done).pipe(Effect.exit, Effect.asVoid) @@ -1209,7 +1244,7 @@ const startGenericInternal: < } const cleanupStartupFailure = (exit: Exit.Exit): Effect.Effect => Exit.isFailure(exit) - ? closeChildren(exit).pipe(Effect.andThen(closeEmissions())) + ? closeChildren(exit).pipe(Effect.andThen(emissions.close())) : Effect.void const cleanup = onStopSync === undefined ? onStop ?? Effect.void : Effect.sync(onStopSync) const sendParent = overrideSendParent ?? (parent === undefined ? noParentSend : parent.send) @@ -1219,7 +1254,7 @@ const startGenericInternal: < parent, spawn, sendParent, - emit, + emit: emissions.emit, sendTo: ((target: unknown, event: unknown) => isProcessAddress(target) ? target.send(event) : sendTo(target as ChildSelector, event)) as ProcessScope< Event @@ -1402,7 +1437,7 @@ const startGenericInternal: < Effect.andThen(Queue.shutdown(queue)), Effect.andThen(closeChildren(exit)), Effect.andThen(setAndPublishSnapshot(snapshot)), - Effect.andThen(closeEmissions()), + Effect.andThen(emissions.close()), Effect.andThen(Effect.sync(() => { if (Exit.isFailure(exit)) { failAcknowledgedMessage(inFlightMessage, exit.cause) @@ -1559,11 +1594,7 @@ const startGenericInternal: < state: SynchronizedRef.get(current).pipe(Effect.map((current) => current.snapshot.state)), snapshot: SynchronizedRef.get(current).pipe(Effect.map((current) => current.snapshot)), changes: changesStream, - emissions: Stream.unwrap( - getOrCreateEmissions.pipe( - Effect.map((emissions) => emissions === undefined ? Stream.empty : Stream.fromPubSub(emissions)) - ) - ) as Stream.Stream, + emissions: emissions.stream as Stream.Stream, join: Deferred.await(done), stop, send: self.send, @@ -1717,6 +1748,7 @@ class CompiledProcess implements MachineRef { private interruptRequested = false private offerRevision = 0 private inFlightMessage: ProcessMessage | undefined + private readonly externalEmissions: EmissionRuntime | undefined private emissionsPubSub: LazyEmissions constructor( @@ -1727,6 +1759,7 @@ class CompiledProcess implements MachineRef { ) { this.sessionId = sessionId this.id = options.id ?? sessionId + this.externalEmissions = options.emissions this.send = (event) => this.offerMessage(event) this.address = { id: this.id, @@ -1908,7 +1941,7 @@ class CompiledProcess implements MachineRef { } get emissions(): Stream.Stream { - return this.emissionsStream() as Stream.Stream + return (this.externalEmissions?.stream ?? this.emissionsStream()) as Stream.Stream } get join(): Effect.Effect { @@ -2268,6 +2301,7 @@ class CompiledProcess implements MachineRef { } private emitEvent(event: unknown): Effect.Effect { + if (this.externalEmissions !== undefined) return this.externalEmissions.emit(event) return Effect.suspend(() => this.emissionsPubSub === undefined || this.emissionsPubSub === EmissionsClosed ? Effect.void @@ -2280,6 +2314,7 @@ class CompiledProcess implements MachineRef { } private closeEmissions(): Effect.Effect { + if (this.externalEmissions !== undefined) return this.externalEmissions.close() const observed = this.emissionsPubSub this.emissionsPubSub = EmissionsClosed return observed === undefined || observed === EmissionsClosed ? Effect.void : PubSub.shutdown(observed) @@ -2523,7 +2558,7 @@ const startCompactCompiledInternal: typeof startGenericInternal = Effect.fnUntra logic: ProcessLogic, options: StartInternalOptions ) { - const sessionId = yield* options.runtime.nextSessionId + const sessionId = options.sessionId ?? (yield* options.runtime.nextSessionId) const services = yield* Effect.context() const execution = logic.execution as CompiledProcessExecution const process = new CompiledProcess(logic, options, services, sessionId) @@ -2636,3 +2671,97 @@ export const startProcess: < } ) }) + +const prepareProcessWithStrategy = Effect.fnUntraced(function*< + State, + Event, + Error, + Requirements, + Output, + InitialError, + Emitted +>( + logic: ProcessLogic, + strategy: ProcessRuntimeStrategy, + options?: { + readonly id?: string + } +) { + const runtime = yield* makeProcessRuntime + const sessionId = yield* runtime.nextSessionId + const emissions = makeEmissionRuntime() + const started = yield* Deferred.make, InitialError>() + const internalOptions: StartInternalOptions = options === undefined + ? { + detached: true, + emissions, + runtime, + sessionId + } + : { + ...options, + detached: true, + emissions, + runtime, + sessionId + } + const initialize = strategy === "generic" + ? startGenericInternal(logic, internalOptions) + : strategy === "compiled" + ? logic.execution?._tag === "Compiled" + ? startCompactCompiledInternal(logic, internalOptions) + : Effect.die(new Error("Machine cannot force the compiled runtime for generic process logic")) + : startLogicInternal(logic, internalOptions) + const start = yield* Effect.cached( + initialize.pipe( + Effect.onExit((exit) => Deferred.done(started, exit)) + ) as Effect.Effect, InitialError, Requirements> + ) + return { + id: options?.id ?? sessionId, + sessionId, + changes: Stream.unwrap( + Deferred.await(started).pipe(Effect.map((ref) => ref.changes)) + ), + emissions: emissions.stream as Stream.Stream, + start + } +}) + +export const prepareProcess: < + State, + Event, + Error = never, + Requirements = never, + Output = never, + InitialError = never, + Emitted = never +>( + logic: ProcessLogic, + options?: { + readonly id?: string + } +) => Effect.Effect< + PreparedProcess +> = + ((logic: ProcessLogic, options?: { readonly id?: string }) => + prepareProcessWithStrategy(logic, "auto", options)) as any + +/** @internal Test-only prepared startup strategy selection. */ +export const prepareProcessWithStrategyForTesting = < + State, + Event, + Error = never, + Requirements = never, + Output = never, + InitialError = never, + Emitted = never +>( + logic: ProcessLogic, + strategy: ProcessRuntimeStrategy, + options?: { + readonly id?: string + } +): Effect.Effect< + PreparedProcess +> => prepareProcessWithStrategy(logic, strategy, options) as any diff --git a/src/unstable/reactivity/AtomMachine.ts b/src/unstable/reactivity/AtomMachine.ts index b96007b..99026db 100644 --- a/src/unstable/reactivity/AtomMachine.ts +++ b/src/unstable/reactivity/AtomMachine.ts @@ -167,8 +167,10 @@ type RefOutput = Ref extends Machine.MachineRef = Ref extends Machine.MachineRef ? Emitted : never /** - * Observes ephemeral notifications from the actor owned by a machine atom. - * The stream requires an `AtomRegistry` and does not replay earlier emissions. + * Observes ephemeral notifications from the running machine owned by a machine + * atom. When this stream activates a fresh bridge, it subscribes before machine + * initialization and observes initial-entry emissions. It never replays + * emissions from a machine that was already running. * * @category getters * @since 0.10.0 diff --git a/test/internal/machine/strategyDifferential.test.ts b/test/internal/machine/strategyDifferential.test.ts index e353f90..7255fd6 100644 --- a/test/internal/machine/strategyDifferential.test.ts +++ b/test/internal/machine/strategyDifferential.test.ts @@ -7,7 +7,11 @@ import * as ExecutionPlan from "../../../src/internal/machine/executionPlan.js" import { MachineTest } from "../../../src/testing/index.js" import type { DifferentialStep } from "../../machine/support/runtimeDifferential.js" import { verifyManagedExecution } from "../../machine/support/runtimeDifferential.js" -import { openWithRuntimeStrategy, verifyPlannerStrategies } from "./support/strategyDifferential.js" +import { + openWithRuntimeStrategy, + prepareWithRuntimeStrategy, + verifyPlannerStrategies +} from "./support/strategyDifferential.js" class Count extends Schema.TaggedClass("StrategyCount")("Count", { value: Schema.Number @@ -385,6 +389,39 @@ describe("machine planner and runtime strategies", () => { } }) as Effect.Effect) + it.effect("observes initial emissions from prepared generic and compiled runtimes", () => + Effect.gen(function*() { + class Idle extends Schema.TaggedClass("StrategyPreparedIdle")("Idle", {}) {} + class Ready extends Schema.TaggedClass("StrategyPreparedReady")("Ready", {}) {} + const states = Machine.defineStates({ Idle }) + const Emissions = Machine.emittedEvents(Ready) + const machine = Machine.make({ + states: states.states, + events: Machine.events(), + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + entry: (_, enqueue) => { + enqueue.emit(Emissions.Ready()) + return undefined + } + } + }) + + for (const strategy of ["generic", "compiled"] as const) { + const prepared = yield* prepareWithRuntimeStrategy(machine, strategy) + const observed = yield* prepared.emissions.pipe( + Stream.take(1), + Stream.runCollect, + Effect.forkChild({ startImmediately: true }) + ) + const ref = yield* prepared.start + assert.deepStrictEqual(Array.from(yield* Fiber.join(observed)), [new Ready({})]) + yield* ref.stop + } + }) as Effect.Effect) + it.effect("matches acknowledged probe delivery in generic and compiled managed runtimes", () => Effect.gen(function*() { const machine = makeFlatMachine() diff --git a/test/internal/machine/support/strategyDifferential.ts b/test/internal/machine/support/strategyDifferential.ts index af161a2..591b442 100644 --- a/test/internal/machine/support/strategyDifferential.ts +++ b/test/internal/machine/support/strategyDifferential.ts @@ -117,6 +117,17 @@ export const openWithRuntimeStrategy = ( ): Effect.Effect, unknown> => Process.startWithRuntimeStrategyForTesting(machine, strategy) as any +export const prepareWithRuntimeStrategy = ( + machine: Machine.Machine.Any, + strategy: "generic" | "compiled" +): Effect.Effect< + { + readonly emissions: import("effect/Stream").Stream + readonly start: Effect.Effect, unknown> + }, + unknown +> => Process.prepareWithRuntimeStrategyForTesting(machine, strategy) as any + export const resumeWithRuntimeStrategy = ( machine: Machine.Machine.Any, snapshot: Machine.Machine.Snapshot, diff --git a/test/machine/ActorEvents.test.ts b/test/machine/ActorEvents.test.ts index 39fd0c2..80c138b 100644 --- a/test/machine/ActorEvents.test.ts +++ b/test/machine/ActorEvents.test.ts @@ -6,6 +6,92 @@ const collectNext = (stream: Stream.Stream) => stream.pipe(Stream.take(1), Stream.runCollect, Effect.map(Array.from), Effect.forkChild({ startImmediately: true })) describe("actor event channels", () => { + it.effect("observes initial emissions through a prepared machine", () => + Effect.gen(function*() { + class Idle extends Schema.TaggedClass("PreparedEmissionIdle")("Idle", {}) {} + class Ready extends Schema.TaggedClass("PreparedEmissionReady")("Ready", {}) {} + + const states = Machine.defineStates({ Idle }) + const Emissions = Machine.emittedEvents(Ready) + let initializations = 0 + const machine = Machine.make({ + id: "prepared-emission", + states: states.states, + events: Machine.events(), + emittedEvents: Emissions, + initial: () => { + initializations += 1 + return states.initial.Idle(new Idle({})) + } + }).handle({ + Idle: { + entry: (_, enqueue) => { + enqueue.emit(Emissions.Ready()) + return undefined + } + } + }) + + const prepared = yield* Machine.prepare(machine) + assert.strictEqual(initializations, 0) + assert.strictEqual(prepared.id, "prepared-emission") + + const emitted = yield* collectNext(prepared.emissions) + const changed = yield* prepared.changes.pipe( + Stream.take(1), + Stream.runCollect, + Effect.forkChild({ startImmediately: true }) + ) + const [first, second] = yield* Effect.all( + [prepared.start, prepared.start], + { concurrency: 2 } + ) + + assert.strictEqual(first, second) + assert.strictEqual(first.id, prepared.id) + assert.strictEqual(first.sessionId, prepared.sessionId) + assert.strictEqual(initializations, 1) + assert.deepStrictEqual(Array.from(yield* Fiber.join(emitted)), [new Ready({})]) + assert.deepStrictEqual(Array.from(yield* Fiber.join(changed)).map(({ state }) => state.path), ["Idle"]) + + yield* first.stop + assert.deepStrictEqual(Array.from(yield* Stream.runCollect(prepared.emissions)), []) + })) + + it.effect("fails prepared startup when an initial emission is invalid", () => + Effect.gen(function*() { + class Idle extends Schema.TaggedClass("PreparedInvalidIdle")("Idle", {}) {} + class Published extends Schema.TaggedClass("PreparedInvalidPublished")("Published", { + value: Schema.Number + }) {} + const states = Machine.defineStates({ Idle }) + const Emissions = Machine.emittedEvents(Published) + const machine = Machine.make({ + states: states.states, + events: Machine.events(), + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + entry: (_, enqueue) => { + enqueue.emit(Emissions.Published({ value: "invalid" } as never)) + return undefined + } + } + }) + + const prepared = yield* Machine.prepare(machine) + const observed = yield* prepared.emissions.pipe( + Stream.runCollect, + Effect.forkChild({ startImmediately: true }) + ) + const error = yield* Effect.flip(prepared.start) + + assert.instanceOf(error, Machine.MachineSchemaDecodeError) + assert.strictEqual(error.boundary, "emission") + assert.deepStrictEqual(Array.from(yield* Fiber.join(observed)), []) + })) + it.effect("publishes root emissions as a hot, non-replayed Effect Stream", () => Effect.gen(function*() { class Idle extends Schema.TaggedClass("EmissionRootIdle")("Idle", {}) {} @@ -184,4 +270,55 @@ describe("actor event channels", () => { assert.deepStrictEqual((notices as Array).map(({ value }) => value), [1]) assert.strictEqual(yield* parent.join, "parent event") })) + + it.effect("types and delivers parent input from a machine-bound invocation source", () => + Effect.gen(function*() { + class ChildIdle extends Schema.TaggedClass("BoundInvokeChildIdle")("ChildIdle", {}) {} + class ParentWaiting extends Schema.TaggedClass("BoundInvokeParentWaiting")( + "ParentWaiting", + {} + ) {} + class ParentDone extends Schema.TaggedClass("BoundInvokeParentDone")("ParentDone", {}) {} + class ChildReady extends Schema.TaggedClass("BoundInvokeChildReady")("ChildReady", {}) {} + + const ParentEvents = Machine.events(ChildReady) + const childStates = Machine.defineStates({ ChildIdle }) + const childDefinition = Machine.make({ + states: childStates.states, + events: Machine.events(), + parentEvents: ParentEvents, + initial: () => childStates.initial.ChildIdle(new ChildIdle({})) + }) + const childMachine = childDefinition.handle({ + ChildIdle: { + invoke: childDefinition.invoke({ + id: "notify-ready", + effect: ({ parent }) => parent === undefined ? Effect.void : parent.send(ParentEvents.ChildReady()), + onDone: ({ target }) => target.none(), + onFailure: ({ target }) => target.none() + }) + } + }) + const Child = Machine.child("bound-invoke-child", childMachine) + const parentStates = Machine.defineStates({ + ParentWaiting, + ParentDone: { schema: ParentDone, type: "final", output: Schema.Void } + }) + const parentMachine = Machine.make({ + states: parentStates.states, + events: ParentEvents, + initial: () => parentStates.initial.ParentWaiting(new ParentWaiting({})) + }).handle({ + ParentWaiting: { + invoke: Machine.invoke({ child: Child, onFailure: ({ target }) => target.none() }), + on: { + ChildReady: ({ target }) => target.full.ParentDone(new ParentDone({})) + } + }, + ParentDone: { output: () => undefined } + }) + + const parent = yield* Machine.start(parentMachine) + assert.strictEqual(yield* parent.join, undefined) + }) as Effect.Effect) }) diff --git a/test/unstable/reactivity/AtomMachine.test.ts b/test/unstable/reactivity/AtomMachine.test.ts index 6a083c0..a6fac47 100644 --- a/test/unstable/reactivity/AtomMachine.test.ts +++ b/test/unstable/reactivity/AtomMachine.test.ts @@ -84,6 +84,38 @@ const makeCounterMachine = () => }) describe("AtomMachine", () => { + it.effect("observes initial emissions when the emission stream starts the machine", () => + Effect.scoped(Effect.gen(function*() { + class Idle extends Schema.TaggedClass("AtomPreparedIdle")("Idle", {}) {} + class ReadyEmission extends Schema.TaggedClass("AtomPreparedReady")("ReadyEmission", {}) {} + const states = Machine.defineStates({ Idle }) + const Emissions = Machine.emittedEvents(ReadyEmission) + const machine = Machine.make({ + states: states.states, + events: Machine.events(), + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + entry: (_, enqueue) => { + enqueue.emit(Emissions.ReadyEmission()) + return undefined + } + } + }) + const registry = yield* makeRegistry + const bridge = AtomMachine.make(machine) + const observed = yield* AtomMachine.emissions(bridge).pipe( + Stream.take(1), + Stream.runCollect, + Effect.provideService(AtomRegistry.AtomRegistry, registry), + Effect.forkScoped({ startImmediately: true }) + ) + + assert.deepStrictEqual(Array.from(yield* Fiber.join(observed)), [new ReadyEmission({})]) + assert.strictEqual((yield* AtomRegistry.getResult(registry, bridge.snapshot)).status, "active") + }))) + it.effect("resumes lazily once per registry and disposes the resumed runtime", () => Effect.gen(function*() { let initialCalls = 0 diff --git a/typetest/machine/ActorEvents.tst.ts b/typetest/machine/ActorEvents.tst.ts index e8b77fb..a2cad3f 100644 --- a/typetest/machine/ActorEvents.tst.ts +++ b/typetest/machine/ActorEvents.tst.ts @@ -110,6 +110,22 @@ describe("machine actor event channels", () => { }) it("infers emitted streams through MachineRef and AtomMachine", () => { + const preparedEffect = Machine.prepare(childMachine) + type Prepared = Effect.Success + expect().type.toBe>() + expect().type.toBe< + Stream.Stream< + Machine.RuntimeSnapshot< + Machine.Machine.Snapshot, + Machine.InfiniteTransitionError | Machine.MachineSchemaDecodeError | Machine.StoppedError + >, + | Machine.InfiniteTransitionError + | Machine.MachineSchemaDecodeError + | Machine.StartupError + | Machine.StoppedError + > + >() + const started = Machine.start(childMachine) type Ref = Effect.Success expect().type.toBe>() @@ -119,4 +135,55 @@ describe("machine actor event channels", () => { expect>().type.toBe() expect>().type.toBe() }) + + it("binds invocation self and parent references to the owning machine protocols", () => { + const definition = Machine.make({ + states: states.states, + events: Events, + internalEvents: InternalEvents, + parentEvents: ParentEvents, + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }) + + definition.handle({ + Idle: { + invoke: definition.invoke({ + id: "notify-parent", + effect: ({ parent, self }) => { + expect(self.send).type.toBeCallableWith(Events.Ping()) + expect(self.send).type.not.toBeCallableWith(InternalEvents.Local()) + if (parent !== undefined) { + expect(parent.send).type.toBeCallableWith(ParentEvents.ParentNotice({ value: 1 })) + expect(parent.send).type.not.toBeCallableWith(Events.Ping()) + } + return Effect.void + }, + onDone: ({ parent, self, target }) => { + expect(self.send).type.toBeCallableWith(Events.Ping()) + if (parent !== undefined) { + expect(parent.send).type.toBeCallableWith(ParentEvents.ParentNotice({ value: 1 })) + } + return target.none() + } + }) + } + }) + + definition.handle({ + Idle: { + invoke: Machine.invoke({ + id: "owner-independent", + effect: ({ parent, self }) => { + expect(self.send).type.not.toBeCallableWith(Events.Ping()) + if (parent !== undefined) { + expect(parent.send).type.not.toBeCallableWith(ParentEvents.ParentNotice({ value: 1 })) + } + return Effect.void + }, + onDone: ({ target }) => target.none() + }) + } + }) + }) })