From 040def33dd4866b1ab6247531bf8ad002e2b6a2e Mon Sep 17 00:00:00 2001 From: SandroMaglione Date: Sat, 15 Aug 2026 17:05:05 +0200 Subject: [PATCH] Redesign actor event semantics --- .changeset/fresh-actors-emit.md | 27 + README.md | 89 ++- docs/agent-guide.md | 125 +++- examples/pokemon/src/machine.ts | 4 +- examples/pokemon/src/machines/replace.ts | 23 +- examples/pokemon/src/machines/selection.ts | 12 +- examples/pokemon/src/pokemon.ts | 3 + perf/types/exact-channels-control.ts | 2 +- scripts/fixtures/consumer/deep-bound.ts | 24 +- scripts/type-performance.mjs | 8 +- src/Machine.ts | 600 +++++++++++++----- src/internal/machine/atom.ts | 45 +- src/internal/machine/command.ts | 4 +- src/internal/machine/commandRuntime.ts | 13 +- src/internal/machine/configuration.ts | 28 +- src/internal/machine/errors.ts | 2 +- src/internal/machine/executionPlan.ts | 97 ++- src/internal/machine/invocation.ts | 5 +- src/internal/machine/machine.ts | 84 ++- src/internal/machine/planner.ts | 87 ++- src/internal/machine/process.ts | 26 +- src/internal/machine/protocol.ts | 111 +++- src/internal/machine/runtime.ts | 144 ++++- src/unstable/reactivity/AtomMachine.ts | 46 +- .../machine/strategyDifferential.test.ts | 66 +- test/machine/ActorEvents.test.ts | 181 ++++++ test/machine/Choice.test.ts | 4 +- test/machine/Machine.test.ts | 42 +- test/machine/RuntimeDifferential.test.ts | 25 +- test/machine/StructuralStates.test.ts | 4 +- .../machine/support/activityLifecycleModel.ts | 6 +- test/machine/support/runtimeDifferential.ts | 8 +- test/testing/Probe.test.ts | 1 + test/testing/Runtime.test.ts | 1 + test/unstable/cluster/ClusterMachine.test.ts | 2 +- typetest/machine/ActorEvents.tst.ts | 121 ++++ typetest/machine/Choice.tst.ts | 4 +- typetest/machine/History.tst.ts | 14 +- typetest/machine/Machine.tst.ts | 34 +- typetest/machine/StructuralStates.tst.ts | 14 +- 40 files changed, 1676 insertions(+), 460 deletions(-) create mode 100644 .changeset/fresh-actors-emit.md create mode 100644 test/machine/ActorEvents.test.ts create mode 100644 typetest/machine/ActorEvents.tst.ts diff --git a/.changeset/fresh-actors-emit.md b/.changeset/fresh-actors-emit.md new file mode 100644 index 0000000..1147043 --- /dev/null +++ b/.changeset/fresh-actors-emit.md @@ -0,0 +1,27 @@ +--- +"@typeonce/effect-machine": minor +--- + +Separate actor inputs from outward notifications. Declare emissions with `Machine.emittedEvents`, publish them with `emit`, and observe the hot, non-replaying `MachineRef.emissions` stream. Children declare the public inputs they expect from their owner through `parentEvents`, then communicate explicitly with the typed, optional `parent` actor reference: + +```ts +const Emissions = Machine.emittedEvents(Progress) +const ParentEvents = Machine.events(Completed) + +const worker = Machine.make({ + // ... + emittedEvents: Emissions, + parentEvents: ParentEvents +}).handle({ + Working: { + entry: ({ parent }, enqueue) => { + enqueue.emit(Emissions.Progress({ value: 0.5 })) + if (parent !== undefined) { + enqueue.sendTo(parent, ParentEvents.Completed({ value: 42 })) + } + } + } +}) +``` + +Handler contexts also expose typed `self`; invoked-child composition checks that every `parentEvents` case is accepted by the parent. This release renames structural handler ancestry to `containingState` and `ancestors`, supports zero-payload event and emission constructors with `()`, and exposes root and child emission streams through AtomMachine. diff --git a/README.md b/README.md index 8bfa458..4e8ff17 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ Define schemas first, derive the state topology, then add behavior: ```ts import { Machine } from "@typeonce/effect-machine" -import { Effect, Schema } from "effect" +import { Effect, Schema, Stream } from "effect" const State = Schema.TaggedUnion({ Idle: {}, @@ -70,8 +70,8 @@ const program = Effect.gen(function*() { ``` `Machine.start` returns a `MachineRef` with `send`, `state`, `snapshot`, -`changes`, `join`, and `stop`. Sending enqueues an event; observe `changes` or -use the testing probe when work must be causally acknowledged. +`changes`, `emissions`, `join`, and `stop`. Sending enqueues an event; observe +`changes` or use the testing probe when work must be causally acknowledged. ## Modeling workflow @@ -123,10 +123,11 @@ schema-backed paths. Add a schema later if the state starts owning data. Put data on the narrowest state where it is valid. If sibling phases share data, put it on their compound parent. -### Separate public and internal events +### Separate inputs, raised events, and emissions -`events` is the public command protocol. Invoke results, timer deliveries, -raised events, and child emissions belong in `internalEvents`: +`events` is the public actor-input protocol. Events raised to the same machine +belong in `internalEvents`. Ephemeral outward notifications have their own +`emittedEvents` protocol: ```ts const Command = Schema.TaggedUnion({ Save: {} }) @@ -134,15 +135,20 @@ const Internal = Schema.TaggedUnion({ Saved: { id: Schema.String }, SaveFailed: { message: Schema.String } }) +const Emitted = Schema.TaggedUnion({ + SaveObserved: { id: Schema.String } +}) export const CommandEvent = Machine.events(Command) export type PublicCommandEvent = Machine.EventOf const InternalEvent = Machine.internalEvents(Internal) +const Emissions = Machine.emittedEvents(Emitted) const definition = Machine.make({ states: States.states, events: CommandEvent, internalEvents: InternalEvent, + emittedEvents: Emissions, initial: () => States.initial.Idle.from() }) ``` @@ -157,6 +163,7 @@ events without exposing schema `.make` methods: ```ts ref.send(CommandEvent.Save()) enqueue.raise(InternalEvent.Saved({ id: "entry-1" })) +enqueue.emit(Emissions.SaveObserved({ id: "entry-1" })) ``` The returned constructors preserve each schema's make input, including required @@ -167,6 +174,65 @@ Schemas with an open discriminator such as `_tag: Schema.String` remain valid protocols but cannot expose a finite constructor set; pass a complete event 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. + +```ts +const next = ref.emissions.pipe(Stream.take(1), Stream.runHead) +``` + +Invalid event and emission constructions fail the machine with a typed +`MachineSchemaDecodeError`; they do not throw from the constructor call. + +### Send explicitly between actors + +`raise` targets the current machine in the same macrostep. `sendTo` targets an +actor mailbox and is processed later. A child declares the subset of parent +inputs it may send with `parentEvents`: + +```ts +const ParentEvents = Machine.events(ChildFinished) + +const child = Machine.make({ + states: ChildStates.states, + events: ChildEvents, + parentEvents: ParentEvents, + initial: () => ChildStates.initial.Working.from() +}).handle({ + Working: { + on: { + Finish: ({ parent, target }, enqueue) => { + if (parent !== undefined) { + enqueue.sendTo(parent, ParentEvents.ChildFinished({ id: "job-1" })) + } + return target.full.Done.from() + } + } + }, + Done: {} +}) + +const Child = Machine.child("worker", child) +const ParentInputs = Machine.events(Start, ParentEvents) +``` + +The same child remains isolated and may be started as a root, where `parent` is +`undefined`. When `Child` is invoked, the parent definition must accept every +event in `parentEvents`; otherwise `.handle(...)` is a compile-time error. +Inside the child, the parent reference accepts only those declared events. +`emit` never sends to the parent: it only publishes on the emitting actor's +`emissions` stream. + +Every handler also receives `self`, which can be targeted with `sendTo` when a +later mailbox turn is required. Use `raise` instead for same-macrostep work. +Structural state values use distinct names: `containingState` is the immediate +valued state in the same statechart, while `ancestors` maps valued ancestor +paths. `parent` always means the owning actor reference. + ### Choose the target by scope | Builder | Use when | Preserves | @@ -271,6 +337,17 @@ The bridge exposes `ref`, `snapshot`, `state`, fail-aware `result`, writable equality-aware derivations. React applications using `@effect/atom-react` need a `RegistryProvider`. +Emissions stay streams rather than becoming retained atom state: + +```ts +const rootEmissions = AtomMachine.emissions(counterAtom) +const childEmissions = AtomMachine.childEmissions(counterAtom.child(Worker)) +``` + +These streams require the same `AtomRegistry`, follow the currently mounted +actor instance, and do not replay notifications from an earlier subscription +or child instance. + ## Persistence Logical snapshots can be validated for storage or transport: diff --git a/docs/agent-guide.md b/docs/agent-guide.md index 87c83bd..a12aa1e 100644 --- a/docs/agent-guide.md +++ b/docs/agent-guide.md @@ -22,10 +22,10 @@ declared: 1. Domain schemas used by state and event fields. 2. Tagged schemas for states that own data. -3. Tagged public-event, internal-event, and emitted-event schemas. +3. Tagged public-event, internal-event, parent-event, and emitted-event schemas. 4. `Machine.defineStates`. -5. `Machine.make`, including input, events, internal events, emits, and the - initial function. +5. `Machine.make`, including input, `events`, `internalEvents`, `parentEvents`, + `emittedEvents`, and the initial function. 6. One or more `.handle(...)` calls. 7. Child descriptors. 8. Runtime, Atom, or Cluster adapters. @@ -80,7 +80,7 @@ the deferred constructors preserve that identity after decoding. a handler. - Every declared output schema needs a matching handler implementation before planning or execution. -- `parents` keys are full dotted paths. +- Handler `ancestors` keys are full dotted paths. - Invoke lifetimes follow state entry and exit, not the spelling of the target builder. - Handle every typed invoked Effect failure with `onFailure`. Defects and @@ -88,9 +88,10 @@ the deferred constructors preserve that identity after decoding. - Reuse an exported child descriptor for inline invocation, `sendTo`, and child lookup. Independently constructed descriptors are equivalent only when both their id and machine identity match. -- `events` is the public input protocol. `internalEvents` contains machine-local - deliveries such as raised events and invoked-child emissions. Handlers see - both; typed public `send` and `Machine.plan` accept only `events`. +- `events` is the public actor-input protocol. `internalEvents` contains + machine-local raised events. `parentEvents` describes the public events a + child may send to its owner. `emittedEvents` describes outward ephemeral + notifications and is never delivered implicitly to a parent. - Event tags in `events` and `internalEvents` must be disjoint. - Event tags must also be unique within each protocol list. @@ -156,8 +157,8 @@ States.get(snapshot, "Form") // type error: no value schema For a schema-less path, builders expose only `.from(...)`; the direct callable form is reserved for already-decoded schema values. Structural ancestors are -also omitted from `parents`; an immediate structural parent is typed as -`undefined`. Add `schema` when a state begins to own data or needs runtime +also omitted from `ancestors`; an immediate structural containing state is +typed as `undefined`. Add `schema` when a state begins to own data or needs runtime validation and persistence for that data. Use an atomic state when no child phase can be active beneath it. @@ -380,7 +381,7 @@ parallel builders still require a callback selecting their active child or every active region. Omitted input is normalized to `{}` and still passes through `schema.makeEffect`, including refinements. -## Reading state and parents +## Reading state and structural ancestors `Machine.defineStates` returns typed helpers: @@ -402,16 +403,18 @@ States.matches(ready, "Route.Ready.Saving") All paths are checked against the definition. `get` and `getWithParents` accept only schema-backed paths; use `matches` or `getSnapshot` for any active path. -`context.parent` is the immediate typed parent value (`undefined` at a root or -when that parent is schema-less). `parents` contains only valued ancestors. Use -its full paths when another ancestor value is needed: +`context.containingState` is the immediate typed state value (`undefined` at a +root or when that state is schema-less). `context.ancestors` contains only +valued structural ancestors. This is separate from `context.parent`, which is +the owning actor reference or `undefined` for a root actor. Use full state paths +when another ancestor value is needed: ```ts -parents["Route.Ready"] -parents["Route.Ready.Editing"] +ancestors["Route.Ready"] +ancestors["Route.Ready.Editing"] ``` -Do not guess short properties such as `parents.Ready`. +Do not guess short properties such as `ancestors.Ready`. ### Inspecting the full transition configuration @@ -473,11 +476,72 @@ Closed statechart and actor operations use `enqueue`: ```ts Submit: ({ target }, enqueue) => { - enqueue.emit(new SaveRequested({})) + enqueue.emit(Emissions.SaveRequested()) return target.local.Saving.from() } ``` +Declare emission constructors separately from actor inputs: + +```ts +const Emissions = Machine.emittedEvents(SaveRequested, AuditRecorded) + +const definition = Machine.make({ + events: Commands, + internalEvents: InternalEvents, + emittedEvents: Emissions, + // ... +}) +``` + +`enqueue.raise(...)` is a same-macrostep input to self. `enqueue.sendTo(...)` +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. + +For child-to-parent input, export a public builder protocol and reuse it at both +composition boundaries: + +```ts +export const ParentEvents = Machine.events(ChildFinished) + +const child = Machine.make({ + events: ChildEvents, + parentEvents: ParentEvents, + // ... +}).handle({ + Working: { + on: { + Finish: ({ parent }, enqueue) => { + if (parent !== undefined) { + enqueue.sendTo(parent, ParentEvents.ChildFinished()) + } + } + } + } +}) + +const parent = Machine.make({ + events: Machine.events(ParentCommands, ParentEvents), + // ... +}) +``` + +Invoking the child under a parent that lacks any required `parentEvents` case +is a type error. Within child handlers, `parent` accepts only that protocol. +The same child may run as a root, where `parent` is `undefined`. `self` accepts +the machine's public inputs. Neither actor reference is a structural state +value; use `containingState` and `ancestors` for statechart ancestry. + +Atom-backed actors retain the same transient semantics. Use +`AtomMachine.emissions(machineAtom)` for a root and +`AtomMachine.childEmissions(childAtom)` for the currently active child. Both +return streams requiring the corresponding `AtomRegistry`; emissions are not +stored as atom state. + For asynchronous validation or persistence, invoke an Effect or child machine from the state and handle its typed success or failure event in a later transition. This keeps `(state, event) => [nextState, commands]` synchronous. @@ -551,8 +615,9 @@ type AnyHandledEvent = Machine.Machine.Event `MachineRef.send`, `machineAtom.send`, and `Machine.plan` accept decoded public events or constructions returned by `Machine.events`. Transition handlers -receive only decoded events. Raised events and child emissions additionally -accept constructions from `Machine.internalEvents`. The +receive only decoded events. Raised events additionally accept constructions +from `Machine.internalEvents`; outward notifications accept constructions from +`Machine.emittedEvents`. The local planner and runtime intentionally share the complete decoder to support those internal deliveries, so JavaScript or `any` can bypass the local public distinction. @@ -582,11 +647,11 @@ self-interrupts fails the parent. `onDone` is required when the output is not are forbidden when their channel is `never`. The source may also be a function of the owning state's entry context when it -needs `state`, `parent`, `parents`, or the entry `event`. Source construction +needs `state`, `containingState`, `ancestors`, or the entry `event`. Source construction errors, defects, and interruption are machine failures rather than a second phase in `onFailure`. -When a source function reads `state`, `parent`, `parents`, or the entry `event`, +When a source function reads `state`, `containingState`, `ancestors`, or the entry `event`, `Machine.invoke` infers that owner context and the returned Effect's output, error, and service channels together. No return annotation is needed: @@ -955,13 +1020,19 @@ Wrap the initial builder result: initial: () => States.initial.Idle.from() ``` -### Invoked child emits events not accepted by the parent +### Invoked child expects events not accepted by the parent -Create an internal descriptor from the child's emitted schemas: +Export one parent-event protocol from the child boundary and compose it into +the parent's public events: ```ts -events: Machine.events(Submit), -internalEvents: Machine.internalEvents(...ChildMachine.emits) +export const ChildParentEvents = Machine.events(ChildFinished) + +// child +parentEvents: ChildParentEvents + +// parent +events: Machine.events(Submit, ChildParentEvents) ``` ### An internal event is rejected by `send` @@ -996,10 +1067,10 @@ handlers own behavior. ### Parent property does not exist -Use its full path: +Use the structural ancestor's full path: ```ts -parents["Route.Ready"] +ancestors["Route.Ready"] ``` ### Child descriptor types are unrelated diff --git a/examples/pokemon/src/machine.ts b/examples/pokemon/src/machine.ts index cb4e2bf..ed0c9f9 100644 --- a/examples/pokemon/src/machine.ts +++ b/examples/pokemon/src/machine.ts @@ -4,7 +4,7 @@ import { Effect, Schema } from "effect" import { Atom } from "effect/unstable/reactivity" import { ReplaceMachine } from "./machines/replace.ts" import { SelectionMachine } from "./machines/selection.ts" -import { Pokemon, PokemonService, ReplaceInTeam } from "./pokemon.ts" +import { Pokemon, PokemonService, TeamEvents } from "./pokemon.ts" class ActiveTeam extends Schema.TaggedClass("ActiveTeam")("ActiveTeam", { team: Schema.Array(Pokemon) @@ -17,7 +17,7 @@ export const ReplaceChild = Machine.child("replace", ReplaceMachine) const machine = Machine.make({ states: States.states, - events: Machine.events(ReplaceInTeam), + events: TeamEvents, initial: () => States.initial.Loading.from() }).handle({ Loading: { diff --git a/examples/pokemon/src/machines/replace.ts b/examples/pokemon/src/machines/replace.ts index 61066e5..1a2bcc1 100644 --- a/examples/pokemon/src/machines/replace.ts +++ b/examples/pokemon/src/machines/replace.ts @@ -1,6 +1,6 @@ import { Machine } from "@typeonce/effect-machine" import { Effect, Schema } from "effect" -import { Pokemon, PokemonService, ReplaceInTeam } from "../pokemon.ts" +import { Pokemon, PokemonService, TeamEvents } from "../pokemon.ts" class Replacing extends Schema.TaggedClass("Replacing")("Replacing", { id: Pokemon.fields.id @@ -29,11 +29,13 @@ const replaceWithRandom = Effect.sleep("500 millis").pipe( export const ReplaceStates = Machine.defineStates({ Idle: {}, Replacing }) -export const ReplaceEvents = Machine.events(ReplacePokemon, Replaced) +export const ReplaceEvents = Machine.events(ReplacePokemon) +const ReplaceInternalEvents = Machine.internalEvents(Replaced) export const ReplaceMachine = Machine.make({ states: ReplaceStates.states, events: ReplaceEvents, - emits: [ReplaceInTeam], + internalEvents: ReplaceInternalEvents, + parentEvents: TeamEvents, initial: () => ReplaceStates.initial.Idle.from() }).handle({ Idle: { @@ -45,11 +47,16 @@ export const ReplaceMachine = Machine.make({ invoke: Machine.invoke({ id: "replaceWithRandom", effect: replaceWithRandom, - onDone: ({ output, target, state }, enqueue) => { - enqueue.emit(new ReplaceInTeam({ id: state.id, pokemon: output.pokemon })) - return target.full.Idle.from() - }, + onDone: ({ output }, enqueue) => enqueue.raise(ReplaceInternalEvents.Replaced({ pokemon: output.pokemon })), onFailure: ({ target }) => target.full.Idle.from() - }) + }), + on: { + Replaced: ({ event, parent, state, target }, enqueue) => { + if (parent !== undefined) { + enqueue.sendTo(parent, TeamEvents.ReplaceInTeam({ id: state.id, pokemon: event.pokemon })) + } + return target.full.Idle.from() + } + } } }) diff --git a/examples/pokemon/src/machines/selection.ts b/examples/pokemon/src/machines/selection.ts index 000e798..b1e2447 100644 --- a/examples/pokemon/src/machines/selection.ts +++ b/examples/pokemon/src/machines/selection.ts @@ -1,6 +1,6 @@ import { Machine } from "@typeonce/effect-machine" import { Effect, Option, Schema } from "effect" -import { Pokemon, PokemonService, ReplaceInTeam } from "../pokemon.ts" +import { Pokemon, PokemonService, TeamEvents } from "../pokemon.ts" class Search extends Schema.TaggedClass("Search")("Search", { searchText: Schema.String @@ -72,7 +72,7 @@ export const SelectionEvents = Machine.events(SelectPokemon, UpdateSearchText, S export const SelectionMachine = Machine.make({ states: SelectionStates.states, events: SelectionEvents, - emits: [ReplaceInTeam], + parentEvents: TeamEvents, initial: () => SelectionStates.initial.form.from((form) => form @@ -93,8 +93,10 @@ export const SelectionMachine = Machine.make({ states: { WithPokemon: { on: { - ReplacePokemon: ({ event, state, target }, enqueue) => { - enqueue.emit(new ReplaceInTeam({ id: event.id, pokemon: state.pokemon })) + ReplacePokemon: ({ event, parent, state, target }, enqueue) => { + if (parent !== undefined) { + enqueue.sendTo(parent, TeamEvents.ReplaceInTeam({ id: event.id, pokemon: state.pokemon })) + } return target.full.form.from((form) => form .search.from({ searchText: "" }, (search) => search.NoPokemon.from()) @@ -106,7 +108,7 @@ export const SelectionMachine = Machine.make({ Searching: { invoke: Machine.invoke({ id: "search", - effect: ({ parents }) => searchPokemon(parents["form.search"].searchText), + effect: ({ ancestors }) => searchPokemon(ancestors["form.search"].searchText), onDone: ({ output, target }) => output.result.pipe( Option.match({ diff --git a/examples/pokemon/src/pokemon.ts b/examples/pokemon/src/pokemon.ts index 5fde6e7..cf91c02 100644 --- a/examples/pokemon/src/pokemon.ts +++ b/examples/pokemon/src/pokemon.ts @@ -1,3 +1,4 @@ +import { Machine } from "@typeonce/effect-machine" import { Array, Context, Effect, flow, Layer, Option, Random, Schema } from "effect" import { FetchHttpClient, @@ -22,6 +23,8 @@ export class ReplaceInTeam extends Schema.TaggedClass("ReplaceInT pokemon: Pokemon }) {} +export const TeamEvents = Machine.events(ReplaceInTeam) + export class PokemonService extends Context.Service()("app/PokemonService", { make: Effect.gen(function*() { const baseClient = yield* HttpClient.HttpClient diff --git a/perf/types/exact-channels-control.ts b/perf/types/exact-channels-control.ts index 84906d6..d07471b 100644 --- a/perf/types/exact-channels-control.ts +++ b/perf/types/exact-channels-control.ts @@ -21,7 +21,7 @@ export const machine = Machine.make({ states: States.states, events: Machine.events(Start), internalEvents: Machine.internalEvents(Loaded), - emits: [Notice], + emittedEvents: Machine.emittedEvents(Notice), input: Input, initial: (input) => States.initial.Idle(Idle.make({ value: input.seed })) }) diff --git a/scripts/fixtures/consumer/deep-bound.ts b/scripts/fixtures/consumer/deep-bound.ts index 1c62f19..6838ca9 100644 --- a/scripts/fixtures/consumer/deep-bound.ts +++ b/scripts/fixtures/consumer/deep-bound.ts @@ -43,16 +43,19 @@ const ChildStates = Machine.defineStates({ output: Schema.String } }) +const ChildParentEvents = Machine.events(Internal.cases.ChildNotice) const childMachine = Machine.make({ states: ChildStates.states, events: Machine.events(), - emits: [Internal.cases.ChildNotice], + parentEvents: ChildParentEvents, input: Schema.Struct({ value: Schema.String }), initial: ({ value }) => ChildStates.initial.Done(ChildState.cases.Done.make({ value })) }).handle({ Done: { - entry: ({ state }, enqueue) => { - enqueue.emit(Internal.cases.ChildNotice.make({ value: state.value })) + entry: ({ parent, state }, enqueue) => { + if (parent !== undefined) { + enqueue.sendTo(parent, ChildParentEvents.ChildNotice({ value: state.value })) + } }, output: ({ state }) => state.value } @@ -82,11 +85,12 @@ const States = Machine.defineStates({ } }) +const Emissions = Machine.emittedEvents(Emitted.cases.Notice) const machine = Machine.make({ states: States.states, - events: Machine.events(Event.cases.Begin, Event.cases.Save), - internalEvents: Machine.internalEvents(Internal.cases.Loaded, Internal.cases.ChildCompleted, ...childMachine.emits), - emits: [Emitted.cases.Notice], + 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({ @@ -124,7 +128,7 @@ const machine = Machine.make({ }, on: { ChildNotice: ({ event, target }, enqueue) => { - enqueue.emit(Emitted.cases.Notice.make({ value: event.value })) + enqueue.emit(Emissions.Notice({ value: event.value })) return target.local.Saving(State.cases.Saving.make({ value: event.value })) }, ChildCompleted: ({ event, target }) => target.full.Done(State.cases.Done.make({ value: event.value })) @@ -298,11 +302,13 @@ const machineAtom = Bound.make(machine, { seed: "initial" }) type Snapshot = Machine.Machine.Snapshot type StateSuccess = Atom.Success type SendEvent = typeof machineAtom.send extends Atom.Writable ? InputEvent : never -type Output = typeof machineAtom extends AtomMachine.MachineAtom ? Value : never +type Output = typeof machineAtom extends AtomMachine.MachineAtom ? Value : never type Failure = Atom.Failure type StateIsExact = Expect> -type EventsArePublicOnly = Expect>> +type EventsArePublicOnly = Expect< + Equal>> +> type OutputIsExact = Expect> type RuntimeErrorIsPreserved = Expect, RuntimeFailure>> type FailureIsNotUnknown = Expect> diff --git a/scripts/type-performance.mjs b/scripts/type-performance.mjs index 95df47d..6846440 100644 --- a/scripts/type-performance.mjs +++ b/scripts/type-performance.mjs @@ -63,16 +63,16 @@ const scenarios = [ label: "Machine.make (3 states, 2 events)", file: "make.ts", control: "make-control", - maxInstantiations: 11_000, - maxMarginalInstantiations: 7_500 + maxInstantiations: 17_500, + maxMarginalInstantiations: 13_500 }, { id: "handle", label: "machine.handle (3 states, 2 transitions)", file: "handle.ts", control: "make", - maxInstantiations: 30_000, - maxMarginalInstantiations: 19_000 + maxInstantiations: 40_000, + maxMarginalInstantiations: 23_000 }, { id: "dynamic-invoke-control", diff --git a/src/Machine.ts b/src/Machine.ts index e9e8025..5dd8ff9 100644 --- a/src/Machine.ts +++ b/src/Machine.ts @@ -50,6 +50,7 @@ export const TypeId: TypeId = "~effect/Machine" declare const MachineOutputStatesTypeId: unique symbol declare const MachineTypeId: unique symbol declare const EventConstructionTypeId: unique symbol +declare const EmittedEventConstructionTypeId: unique symbol declare const EventProtocolTypeId: unique symbol const ChildMachineLogicTypeId: typeof internal.ChildMachineLogicTypeId = internal.ChildMachineLogicTypeId @@ -125,7 +126,8 @@ export interface Machine< Output = never, Emits extends ReadonlyArray = readonly [], OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > extends Pipeable { readonly [TypeId]: TypeId /** @internal Stable type-level carrier for machine protocol extraction. */ @@ -142,7 +144,8 @@ export interface Machine< Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > /** @internal Prevents output implementation evidence from being widened. */ readonly [MachineOutputStatesTypeId]: Readonly> @@ -162,8 +165,7 @@ export interface Machine< readonly events: Machine.EventProtocol<"public", InputEvents> /** - * Events reserved for raised events, child emissions, and other machine-local - * work. + * Events reserved for raised events and other machine-local work. * * @since 0.4.0 */ @@ -173,11 +175,24 @@ export interface Machine< > /** - * Events that the machine may emit to its parent or external adapter. + * Ephemeral outward notifications published to this actor's observers. + * Emissions are never delivered implicitly to an owning actor. * * @since 0.4.0 */ - readonly emits: Emits + readonly emittedEvents: Machine.EventProtocol<"emitted", Emits> + + /** + * Public input events accepted by an owning actor when this machine is + * running as a child. + * + * The protocol types the optional `parent` actor reference exposed to + * handlers and is checked against a concrete parent's public events at + * composition boundaries. + * + * @since 0.10.0 + */ + readonly parentEvents: Machine.EventProtocol<"public", ParentEvents> /** * Optional schema used to decode the machine input before initialization. @@ -227,7 +242,8 @@ export interface Machine< FinalStates, Output, OutputStates, - InputEvents + InputEvents, + ParentEvents > /** @internal */ @@ -347,7 +363,8 @@ export type ExecutionServices = | Exclude, MachineRuntimeRequirement> /** - * Managed runtime capability used to deliver raised and emitted events. + * Managed runtime capability used to deliver raised events and publish + * emitted notifications. * * @category models * @since 0.4.0 @@ -361,11 +378,42 @@ export interface Runtime { readonly raise: (event: Machine.EventInput) => Effect.Effect /** - * Emits an event through the running machine's parent boundary. + * Publishes an ephemeral notification to the running actor's observers. * * @since 0.4.0 */ - readonly sendParent: (event: Emits) => Effect.Effect + readonly emit: ( + event: Machine.EmittedEventInput + ) => Effect.Effect +} + +/** + * Minimal typed reference accepted by targeted actor commands. + * + * @category models + * @since 0.10.0 + */ +export interface ActorRef { + readonly id: string + readonly sessionId: string + readonly send: (event: Event) => Effect.Effect +} + +/** + * Actor references available while evaluating machine behavior. + * + * @category models + * @since 0.10.0 + */ +export interface ActorContext< + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray +> { + /** Reference to the current actor. Sending queues a later mailbox event. */ + readonly self: ActorRef> + + /** Reference to the owning actor, or `undefined` when running as a root. */ + readonly parent: ActorRef> | undefined } /** @@ -380,11 +428,12 @@ export interface Enqueue { /** Raises an event inside the current macrostep. */ readonly raise: (event: Machine.EventInput) => void - /** Emits an event through the machine's parent boundary. */ - readonly emit: (event: Emits) => void + /** Publishes an ephemeral notification to this actor's observers. */ + readonly emit: (event: Machine.EmittedEventInput) => void /** Sends an event to an invoked child after the transition is selected. */ readonly sendTo: { + (target: ActorRef, event: Event): void (child: Child, event: ChildMachine.Event): void
>(child: Address, event: ChildAddress.Event
): void } @@ -405,7 +454,7 @@ export interface Enqueue { export type Command = | { readonly _tag: "SendTo" - readonly child: ChildMachine.Any | ChildAddress + readonly target: ActorRef | ChildMachine.Any | ChildAddress readonly event: unknown } | { @@ -692,6 +741,12 @@ type ValidateInputEventProtocol< > = [DuplicateInput] extends [never] ? unknown : EventProtocolError<"Public event tags must be unique", DuplicateInput> +type ValidateEmittedEventProtocol< + EmittedEvents extends ReadonlyArray, + DuplicateEmitted extends PropertyKey = DuplicateEventTag +> = [DuplicateEmitted] extends [never] ? unknown + : EventProtocolError<"Emitted event tags must be unique", DuplicateEmitted> + type ValidateInternalEventProtocol< InputEvents extends ReadonlyArray, InternalEvents extends ReadonlyArray, @@ -704,6 +759,14 @@ type ValidateInternalEventProtocol< : EventProtocolError<"Public and internal event tags must be disjoint", Overlap> : EventProtocolError<"Internal event tags must be unique", DuplicateInternal> +type ValidateEventProtocolBuilder< + Kind extends Machine.EventProtocolKind, + Inputs extends ReadonlyArray>, + Schemas extends ReadonlyArray = Machine.EventProtocolInputSchemasOf, + Duplicate extends PropertyKey = DuplicateEventTag +> = [Duplicate] extends [never] ? unknown + : EventProtocolError<"Event protocol tags must be unique", Duplicate> + const SnapshotBuilderStateTypeId: typeof internal.SnapshotBuilderStateTypeId = internal.SnapshotBuilderStateTypeId const SnapshotBuilderConstructionTypeId: unique symbol = Symbol("effect/Machine/SnapshotBuilderConstruction") @@ -1645,7 +1708,9 @@ export type RuntimeOutcome = * @category models * @since 0.4.0 */ -export interface MachineRef { +export interface MachineRef + extends ActorRef +{ /** Stable machine definition id, or a generated fallback when none was declared. */ readonly id: string @@ -1658,9 +1723,15 @@ export interface MachineRef> - /** Streams lifecycle snapshots published after subscription. */ + /** Streams the current lifecycle snapshot followed by later changes. */ readonly changes: Stream.Stream> + /** + * Streams ephemeral notifications published after subscription. Emissions + * are not retained or replayed; the stream completes when the actor stops. + */ + readonly emissions: Stream.Stream + /** Waits for machine output or fails when execution fails or is stopped. */ readonly join: Effect.Effect @@ -1772,13 +1843,6 @@ export declare namespace Logic { /** * Machine-local capabilities available while process logic initializes. * - * **Gotchas** - * - * `sendParent` accepts `unknown` because process logic is independent from - * the parent that eventually owns it. Prefer typed process output or an - * invocation lifecycle transition when either can represent the - * communication. - * * @category models * @since 0.4.0 */ @@ -1792,14 +1856,14 @@ export declare namespace Logic { /** Starts a child process owned by this scope. */ readonly spawn: Spawn - /** Sends an untyped event to the owning process. */ - readonly sendParent: (event: unknown) => Effect.Effect - - /** Sends an event through a typed parent-local child address. */ - readonly sendTo:
>( - id: Address, - event: ChildAddress.Event
- ) => Effect.Effect + /** Sends an event to an actor reference or typed parent-local child address. */ + readonly sendTo: { + (target: ActorRef, event: TargetEvent): Effect.Effect +
>( + id: Address, + event: ChildAddress.Event
+ ): Effect.Effect + } /** Stops a child process selected by its parent-local address. */ readonly stopChild: (id: ChildAddress) => Effect.Effect @@ -1888,7 +1952,8 @@ export declare namespace ChildMachine { | InfiniteTransitionError | MachineSchemaDecodeError | StoppedError, - Machine.Output + Machine.Output, + Machine.EmittedEvent > : never @@ -1898,7 +1963,8 @@ export declare namespace ChildMachine { * @category utility types * @since 0.4.0 */ - export type Event = Ref extends MachineRef ? Event : never + export type Event = Child extends ChildMachine ? Machine.EventInput> + : never } /** @@ -2011,7 +2077,8 @@ export declare namespace Machine { Output, Emits extends ReadonlyArray, OutputStates extends StateIdentifier, - InputEvents extends ReadonlyArray + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray > { readonly states: States readonly events: Events @@ -2023,9 +2090,10 @@ export declare namespace Machine { readonly initialServices: InitialR readonly finalStates: FinalStates readonly output: Output - readonly emits: Emits + readonly emittedEvents: Emits readonly outputStates: OutputStates readonly inputEvents: InputEvents + readonly parentEvents: ParentEvents } /** @@ -2043,11 +2111,12 @@ export declare namespace Machine { export interface Any extends Pipeable { readonly [TypeId]: TypeId /** @internal */ - readonly [MachineTypeId]: TypeCarrier + readonly [MachineTypeId]: TypeCarrier readonly states: StateSchemas readonly events: EventProtocol.Any<"public"> readonly internalEvents: EventProtocol.Any<"internal"> - readonly emits: ReadonlyArray + readonly emittedEvents: EventProtocol.Any<"emitted"> + readonly parentEvents: EventProtocol.Any<"public"> readonly input: Schema.Top | undefined readonly id: string | undefined /** @internal */ @@ -2148,7 +2217,10 @@ export declare namespace Machine { * @category utility types * @since 0.4.0 */ - export type Emits = M[typeof MachineTypeId]["emits"] + export type EmittedEvents = M[typeof MachineTypeId]["emittedEvents"] + + /** @deprecated Use {@link EmittedEvents}. */ + export type Emits = EmittedEvents /** * Extracts state paths with implemented output handlers. @@ -2166,6 +2238,9 @@ export declare namespace Machine { */ export type InputEvents = M[typeof MachineTypeId]["inputEvents"] + /** Extracts the public input protocol required from an owning actor. */ + export type ParentEvents = M[typeof MachineTypeId]["parentEvents"] + /** Extracts an internal event schema tuple from the complete and public protocols. */ export type InternalEventSchemas< Events extends ReadonlyArray, @@ -2207,47 +2282,70 @@ export declare namespace Machine { readonly _tag: Event["_tag"] } + /** Opaque construction returned by {@link emittedEvents}. */ + export interface EmittedEventConstruction { + readonly [EmittedEventConstructionTypeId]: Event + readonly _tag: Event["_tag"] + } + /** A decoded event or a deferred machine-bound construction of that event. */ export type EventInput = | Event | (Event extends { readonly _tag: PropertyKey } ? EventConstruction : never) + /** A decoded emitted event or a deferred emitted-event construction. */ + export type EmittedEventInput = + | Event + | (Event extends { readonly _tag: PropertyKey } ? EmittedEventConstruction : never) + /** Event inputs accepted for a schema tuple at machine delivery boundaries. */ export type EventInputOf> = EventInput> /** Identifies whether an event protocol is accepted publicly or only inside a machine. */ - export type EventProtocolKind = "public" | "internal" + export type EventProtocolKind = "public" | "internal" | "emitted" type EventConstructorInput = Omit type EventConstructor< EventSchema extends TaggedSchema, - Tag extends EventSchema["Type"]["_tag"] - > = {} extends EventConstructorInput ? - (input?: EventConstructorInput) => EventConstruction> - : (input: EventConstructorInput) => EventConstruction> + Tag extends EventSchema["Type"]["_tag"], + Kind extends EventProtocolKind + > = {} extends EventConstructorInput ? ( + input?: EventConstructorInput + ) => Kind extends "emitted" ? EmittedEventConstruction> + : EventConstruction> + : ( + input: EventConstructorInput + ) => Kind extends "emitted" ? EmittedEventConstruction> + : EventConstruction> type FiniteEventTag = string extends Tag ? never : number extends Tag ? never : symbol extends Tag ? never : Tag - type EventConstructorsForSchema = EventSchema extends { + type EventConstructorsForSchema< + EventSchema extends TaggedSchema, + Kind extends EventProtocolKind + > = EventSchema extends { readonly cases: infer Cases extends Readonly> } ? { - readonly [Tag in keyof Cases]: Tag extends Cases[Tag]["Type"]["_tag"] ? EventConstructor + readonly [Tag in keyof Cases]: Tag extends Cases[Tag]["Type"]["_tag"] ? EventConstructor : never } : EventSchema extends { readonly members: infer Members extends ReadonlyArray } ? - Types.UnionToIntersection> + Types.UnionToIntersection> : { - readonly [Tag in FiniteEventTag]: EventConstructor + readonly [Tag in FiniteEventTag]: EventConstructor } /** Protocol-bound constructors keyed by each configured event tag. */ - export type EventConstructors> = { - readonly [Tag in keyof Types.UnionToIntersection>]: - Types.UnionToIntersection>[Tag] + export type EventConstructors< + Events extends ReadonlyArray, + Kind extends EventProtocolKind = "public" + > = { + readonly [Tag in keyof Types.UnionToIntersection>]: + Types.UnionToIntersection>[Tag] } /** @@ -2261,7 +2359,7 @@ export declare namespace Machine { export type EventProtocol< Kind extends EventProtocolKind, Schemas extends ReadonlyArray - > = EventConstructors & { + > = EventConstructors & { readonly [EventProtocolTypeId]: { readonly kind: Kind readonly schemas: Schemas @@ -2278,6 +2376,26 @@ export declare namespace Machine { } } + /** A schema or an existing protocol accepted by an event builder. */ + export type EventProtocolInput = TaggedSchema | EventProtocol.Any + + type EventProtocolInputSchemas< + Kind extends EventProtocolKind, + Input extends EventProtocolInput + > = Input extends EventProtocol ? Schemas + : Input extends TaggedSchema ? readonly [Input] + : readonly [] + + /** Flattens schema and protocol inputs into one owned schema tuple. */ + export type EventProtocolInputSchemasOf< + Kind extends EventProtocolKind, + Inputs extends ReadonlyArray> + > = Inputs extends readonly [ + infer Head extends EventProtocolInput, + ...infer Tail extends ReadonlyArray> + ] ? readonly [...EventProtocolInputSchemas, ...EventProtocolInputSchemasOf] + : readonly [] + /** @internal Extracts the schema tuple carried opaquely by an event protocol. */ export type EventProtocolSchemas = Protocol[typeof EventProtocolTypeId]["schemas"] @@ -2287,7 +2405,10 @@ export declare namespace Machine { * @category utility types * @since 0.4.0 */ - export type Emit = EmitOf> + export type EmittedEvent = EmittedEventOf> + + /** @deprecated Use {@link EmittedEvent}. */ + export type Emit = EmittedEvent /** * A schema whose decoded value contains a `_tag` discriminator. @@ -3075,7 +3196,10 @@ export declare namespace Machine { * @category utility types * @since 0.4.0 */ - export type EmitOf> = Emits[number]["Type"] + export type EmittedEventOf> = Emits[number]["Type"] + + /** @deprecated Use {@link EmittedEventOf}. */ + export type EmitOf> = EmittedEventOf /** * Event values received by lifecycle callbacks. @@ -3819,11 +3943,13 @@ export declare namespace Machine { StateId extends StateIdentifier, EventTag extends TagOf, E, - R - > { + R, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] + > extends ActorContext { readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues /** Complete logical configuration captured at the start of this microstep. */ readonly snapshot: Snapshot readonly event: EventByTag @@ -3847,11 +3973,13 @@ 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 [] + > extends ActorContext { readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues readonly event: LifecycleEvent } @@ -3865,11 +3993,13 @@ 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 [] + > extends ActorContext { readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues readonly event: LifecycleEvent } @@ -3886,12 +4016,14 @@ export declare namespace Machine { StateId extends StateIdentifier, State, Error, - Output - > { + Output, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] + > extends ActorContext { readonly id: string readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues readonly target: TargetBuilder readonly snapshot: Extract, { readonly status: "active" }> } @@ -3907,12 +4039,14 @@ export declare namespace Machine { Events extends ReadonlyArray, Emits extends ReadonlyArray, StateId extends StateIdentifier, - Output - > { + Output, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] + > extends ActorContext { readonly id: string readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues readonly snapshot: Snapshot readonly target: TargetBuilder readonly output: Output @@ -3924,12 +4058,14 @@ export declare namespace Machine { Events extends ReadonlyArray, Emits extends ReadonlyArray, StateId extends StateIdentifier, - Error - > { + Error, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] + > extends ActorContext { readonly id: string readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues readonly snapshot: Snapshot readonly target: TargetBuilder readonly error: Error @@ -3945,11 +4081,13 @@ 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 [] + > extends ActorContext { readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues /** Complete logical configuration captured at the start of this microstep. */ readonly snapshot: Snapshot readonly event: LifecycleEvent @@ -3974,11 +4112,13 @@ 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 [] + > extends ActorContext { readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues /** Complete logical configuration captured at the start of this microstep. */ readonly snapshot: Snapshot readonly event: LifecycleEvent @@ -3999,13 +4139,15 @@ export declare namespace Machine { States extends StateSchemas, Events extends ReadonlyArray, Emits extends ReadonlyArray, - ChoiceId extends ChoiceIdentifier - > { - readonly parent: StateByIdentifier< + ChoiceId extends ChoiceIdentifier, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] + > extends ActorContext { + readonly containingState: StateByIdentifier< States, Extract, StateIdentifier> > - readonly parents: { + readonly ancestors: { readonly [Parent in Extract, ValuedStateIdentifier>]: StateByIdentifier< States, Parent @@ -4027,8 +4169,8 @@ export declare namespace Machine { StateId extends StateIdentifier > { readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues readonly event: LifecycleEvent } @@ -4062,8 +4204,8 @@ export declare namespace Machine { StateId extends StateIdentifier > { readonly state: StateByIdentifier - readonly parent: ParentStateValue - readonly parents: ParentStateValues + readonly containingState: ParentStateValue + readonly ancestors: ParentStateValues readonly event: LifecycleEvent readonly outputs: ParallelOutputRegions } @@ -4314,6 +4456,14 @@ export declare namespace Machine { } ? Emitted : Invoke extends { readonly child: ChildMachine } ? Emit : never + + /** Public parent inputs required by an invoked child machine. */ + export type InvokeParentEvents = Invoke extends { + readonly [InvokeTypeId]: { readonly parentEvents: Types.Covariant } + } ? ParentEvent + : IsAny extends true ? never + : Invoke extends { readonly child: ChildMachine } ? EventOf> + : never /** Extracts transition results returned by invocation lifecycle handlers. */ export type InvokeOutcomeReturn = Invoke extends unknown ? | (Invoke extends { readonly onDone?: infer Handler } ? EventTransitionReturn> : never) @@ -4423,13 +4573,14 @@ export declare namespace Machine { } /** Type evidence retained by {@link invoke} without affecting runtime data. */ - export interface InvokeTyped { + export interface InvokeTyped { readonly [InvokeTypeId]: { readonly output: Types.Covariant readonly error: Types.Covariant readonly requirements: Types.Covariant readonly initialError: Types.Covariant readonly emits: Types.Covariant + readonly parentEvents: Types.Covariant } } @@ -4858,47 +5009,49 @@ export declare namespace Machine { Emits extends ReadonlyArray, StateId extends StateIdentifier, E, - R + R, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = { readonly entry?: ( - context: StateActionContext, + context: StateActionContext, enqueue: Enqueue, EmitOf> ) => StateActionResult readonly exit?: ( - context: StateActionContext, + context: StateActionContext, enqueue: Enqueue, EmitOf> ) => StateActionResult readonly invoke?: InvokeDefinition readonly always?: | (( - context: AlwaysContext, + context: AlwaysContext, enqueue: Enqueue, EmitOf> ) => HandlerResult) | { /** Statically declared upper bound of possible target paths. */ readonly targets?: ReadonlyArray> readonly transition: ( - context: AlwaysContext, + context: AlwaysContext, enqueue: Enqueue, EmitOf> ) => HandlerResult } readonly onDone?: | (( - context: DoneContext, + context: DoneContext, enqueue: Enqueue, EmitOf> ) => HandlerResult) | { /** Statically declared upper bound of possible target paths. */ readonly targets?: ReadonlyArray> readonly transition: ( - context: DoneContext, + context: DoneContext, enqueue: Enqueue, EmitOf> ) => HandlerResult } readonly on?: { readonly [EventTag in TagOf]?: | (( - context: HandlerContext, + context: HandlerContext, enqueue: Enqueue, EmitOf> ) => HandlerResult) | { @@ -4910,12 +5063,12 @@ export declare namespace Machine { */ readonly targets?: ReadonlyArray> readonly transition: ( - context: HandlerContext, + context: HandlerContext, enqueue: Enqueue, EmitOf> ) => HandlerResult } } - readonly initial?: StateInitialHandler + readonly initial?: StateInitialHandler } & ActiveOutputHandlerConfig /** Values supplied when a statechart implicitly enters a state's initial children. */ @@ -4940,17 +5093,21 @@ export declare namespace Machine { States extends StateSchemas, Events extends ReadonlyArray, Emits extends ReadonlyArray, - StateId extends StateIdentifier - > = StateActionContext + StateId extends StateIdentifier, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] + > = StateActionContext /** Initial child value implementation for a compound or parallel state. */ export type StateInitialHandler< States extends StateSchemas, Events extends ReadonlyArray, Emits extends ReadonlyArray, - StateId extends StateIdentifier + StateId extends StateIdentifier, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = ( - context: StateInitialContext, + context: StateInitialContext, enqueue: Enqueue, EmitOf> ) => StateInitialValue @@ -4963,7 +5120,7 @@ export declare namespace Machine { > { readonly event: LifecycleEvent readonly target: HistoryDefaultTargetBuilder - readonly parent: ParentId + readonly owner: ParentId } /** @@ -5003,13 +5160,15 @@ export declare namespace Machine { States extends StateSchemas, Events extends ReadonlyArray, Emits extends ReadonlyArray, - ChoiceId extends ChoiceIdentifier + ChoiceId extends ChoiceIdentifier, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > { readonly choice: { /** Statically inspectable upper bound of every possible target. */ readonly targets: readonly [StateNodeIdentifier, ...ReadonlyArray>] readonly transition: ( - context: ChoiceContext, + context: ChoiceContext, enqueue: Enqueue, EmitOf> ) => ChoiceResult } @@ -5033,10 +5192,12 @@ 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 entry?: ( - context: StateActionContext, + context: StateActionContext, enqueue: Enqueue, EmitOf> ) => StateActionResult readonly exit?: never @@ -5057,10 +5218,12 @@ export declare namespace Machine { Emits extends ReadonlyArray, StateId extends StateIdentifier, E, - R + R, + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] > = NodeByIdentifier extends { readonly type: "final" } ? - FinalStateConfig - : ActiveStateConfig + FinalStateConfig + : ActiveStateConfig type HandlerNodeConfig< States extends StateSchemas, @@ -5068,9 +5231,11 @@ export declare namespace Machine { Emits extends ReadonlyArray, Path extends StateNodeIdentifier, E, - R - > = Path extends ChoiceIdentifier ? ChoiceStateConfig - : Path extends StateIdentifier ? HandlerConfig + R, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray + > = Path extends ChoiceIdentifier ? ChoiceStateConfig + : Path extends StateIdentifier ? HandlerConfig : never type HandlerChildren = Node extends { readonly states: infer Children extends StateSchemas } ? Children : never @@ -5153,9 +5318,11 @@ export declare namespace Machine { Emits extends ReadonlyArray, E, R, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, StateId extends StateNodeIdentifier > = - & HandlerNodeConfig + & HandlerNodeConfig & (StateId extends ChoiceIdentifier ? { readonly states?: never readonly history?: never @@ -5172,6 +5339,8 @@ export declare namespace Machine { Emits, E, R, + InputEvents, + ParentEvents, Extract> > readonly history?: HistoryDefaultConfig< @@ -5194,6 +5363,8 @@ export declare namespace Machine { Emits extends ReadonlyArray, E, R, + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, Prefix extends string > = { readonly [Key in ActiveStateKey | ChoiceStateKey]?: HandlerNode< @@ -5203,6 +5374,8 @@ export declare namespace Machine { Emits, E, R, + InputEvents, + ParentEvents, HandlerNodeId> > } @@ -5437,6 +5610,7 @@ export declare namespace Machine { AllStates extends StateSchemas, Node, Events extends ReadonlyArray, + InputEvents extends ReadonlyArray, Emits extends ReadonlyArray, StateId extends StateNodeIdentifier, Config, @@ -5451,7 +5625,7 @@ export declare namespace Machine { & HandlerOnTargetValidation & HandlerDirectTargetValidation & HandlerDirectTargetValidation - & HandlerInvokeEmitsValidation + & HandlerInvokeParentEventsValidation & HandlerChildrenValidation & HandlerOutputRequirementValidation & HandlerRuntimeValidation @@ -5480,17 +5654,17 @@ export declare namespace Machine { } : unknown - type HandlerInvokeEmitsValidation< + type HandlerInvokeParentEventsValidation< Events extends ReadonlyArray, StateId extends string, Config > = [InvokeReturn] extends [never] ? unknown - : [Exclude>, EventOf>] extends [never] ? unknown + : [Exclude>, EventOf>] extends [never] ? unknown : { readonly invoke: HandlerValidationError< - "Invoked child emits events not accepted by the parent machine", + "Invoked child expects parent events not accepted by this machine", StateId, - Exclude>, EventOf> + Exclude>, EventOf> > } @@ -5510,6 +5684,7 @@ export declare namespace Machine { type HandlerNodeValidationAtPath< AllStates extends StateSchemas, Events extends ReadonlyArray, + InputEvents extends ReadonlyArray, Emits extends ReadonlyArray, Config, AvailableOutputStates extends StateIdentifier, @@ -5520,6 +5695,7 @@ export declare namespace Machine { AllStates, HandlerNodeByPath, Events, + InputEvents, Emits, StateId, NodeConfig, @@ -5531,6 +5707,7 @@ export declare namespace Machine { type HandlerTreeNodeValidations< AllStates extends StateSchemas, Events extends ReadonlyArray, + InputEvents extends ReadonlyArray, Emits extends ReadonlyArray, Config, AvailableOutputStates extends StateIdentifier @@ -5539,6 +5716,7 @@ export declare namespace Machine { StateId extends StateNodeIdentifier ? HandlerNodeValidationAtPath< AllStates, Events, + InputEvents, Emits, Config, AvailableOutputStates, @@ -5551,12 +5729,13 @@ export declare namespace Machine { type HandlerTreeValidation< AllStates extends StateSchemas, Events extends ReadonlyArray, + InputEvents extends ReadonlyArray, Emits extends ReadonlyArray, Config, AvailableOutputStates extends StateIdentifier > = & HandlerUnknownStateKeyValidation - & HandlerTreeNodeValidations + & HandlerTreeNodeValidations type HandlerHasRequiredInitial< AllStates extends StateSchemas, @@ -5652,6 +5831,7 @@ export declare namespace Machine { Output, OutputStates extends StateIdentifier, InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray, Config > = Machine< AllStates, @@ -5670,7 +5850,8 @@ export declare namespace Machine { Output, Emits, OutputStates | Extract["outputState"], StateIdentifier>, - InputEvents + InputEvents, + ParentEvents > /** @@ -5692,15 +5873,29 @@ export declare namespace Machine { FinalStates extends StateIdentifier, Output, OutputStates extends StateIdentifier, - InputEvents extends ReadonlyArray + InputEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray > { - >( + < + const Config extends HandlerTree< + States, + States, + Events, + Emits, + E, + R, + InputEvents, + ParentEvents, + "" + > + >( config: & Config & HandlerInvokeContexts> & HandlerTreeValidation< States, Events, + InputEvents, Emits, NoInfer, | OutputStates @@ -5723,6 +5918,7 @@ export declare namespace Machine { Output, OutputStates, InputEvents, + ParentEvents, Config > } @@ -5874,7 +6070,8 @@ export const isFinal: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: Machine< States, @@ -5889,10 +6086,11 @@ export const isFinal: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents >, state: Machine.Snapshot -) => state is Machine.SnapshotContainingFinal = internal.isFinal +) => state is Machine.SnapshotContainingFinal = internal.isFinal as any /** * Defines a state tree while preserving literal state paths. @@ -5939,7 +6137,8 @@ type MakeConfig< Input extends Schema.Top, InitialE, InitialR, - InternalEvents extends ReadonlyArray + InternalEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray > = { readonly id?: string readonly states: States & DefineStateTreeInput> @@ -5952,7 +6151,8 @@ type MakeConfig< NoInfer, NoInfer > - readonly emits?: Emits + readonly emittedEvents?: Machine.EventProtocol<"emitted", Emits> + readonly parentEvents?: Machine.EventProtocol<"public", ParentEvents> readonly input?: Input readonly initial: (...args: [...Machine.InputArgs]) => Machine.InitialResult } @@ -5964,7 +6164,8 @@ type MakeResult< Input extends Schema.Top, InitialE, InitialR, - InternalEvents extends ReadonlyArray + InternalEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray > = Machine< States, readonly [...InputEvents, ...InternalEvents], @@ -5978,7 +6179,8 @@ type MakeResult< Machine.TerminalOutput, Emits, never, - InputEvents + InputEvents, + ParentEvents > interface Make { @@ -5989,11 +6191,12 @@ interface Make { const Input extends Schema.Top = typeof Schema.Void, InitialE = never, InitialR = never, - const InternalEvents extends ReadonlyArray = readonly [] + const InternalEvents extends ReadonlyArray = readonly [], + const ParentEvents extends ReadonlyArray = readonly [] >( - config: MakeConfig, + config: MakeConfig, ..._validation: ValidateDefinedStates> - ): MakeResult + ): MakeResult < const States extends Machine.StateSchemas, const InputEvents extends ReadonlyArray, @@ -6001,10 +6204,11 @@ interface Make { const Input extends Schema.Top = typeof Schema.Void, InitialE = never, InitialR = never, - const InternalEvents extends ReadonlyArray = readonly [] + const InternalEvents extends ReadonlyArray = readonly [], + const ParentEvents extends ReadonlyArray = readonly [] >( config: - & Omit, "states"> + & Omit, "states"> & { readonly states: InvalidDefinedStateTreeInput } ): never } @@ -6021,9 +6225,12 @@ interface Make { * to implement state behavior with ordinary TypeScript control flow. * * `Machine.events` defines the public input protocol. `Machine.internalEvents` - * adds events used for raised events, child emissions, and other machine-local - * deliveries. Both descriptors expose deferred constructors while retaining - * their schemas opaquely for runtime validation. Their tags must be disjoint. + * adds raised events and other machine-local deliveries. + * `Machine.emittedEvents` defines outward ephemeral notifications, while + * `parentEvents` declares the inputs a child may explicitly send to its owning + * actor. All descriptors expose deferred constructors while retaining their + * schemas opaquely for runtime validation. Public and internal tags must be + * disjoint. * * **Example** (Typed counter machine) * @@ -6092,9 +6299,14 @@ export type EventOf = Machine.EventO * @category constructors * @since 0.10.0 */ -export const events = >( - ...schemas: Schemas & ValidateInputEventProtocol> -): Machine.EventProtocol<"public", readonly [...Schemas]> => internal.events(...(schemas as Schemas)) +export const events: { + >( + ...schemas: Schemas & ValidateInputEventProtocol> + ): Machine.EventProtocol<"public", readonly [...Schemas]> + >>( + ...inputs: Inputs & ValidateEventProtocolBuilder<"public", Inputs> + ): Machine.EventProtocol<"public", Machine.EventProtocolInputSchemasOf<"public", Inputs>> +} = internal.events as any /** * Defines an internal event protocol and returns deferred constructors for @@ -6111,15 +6323,38 @@ export const events = * const machine = Machine.make({ internalEvents: InternalEvents, ... }) * * // Inside a transition callback: - * return [nextState, [raise(InternalEvents.Loaded({ value }))]] + * enqueue.raise(InternalEvents.Loaded({ value })) * ``` * * @category constructors * @since 0.10.0 */ -export const internalEvents = >( - ...schemas: Schemas & ValidateInternalEventProtocol> -): Machine.EventProtocol<"internal", readonly [...Schemas]> => internal.internalEvents(...(schemas as Schemas)) +export const internalEvents: { + >( + ...schemas: Schemas & ValidateInternalEventProtocol> + ): Machine.EventProtocol<"internal", readonly [...Schemas]> + >>( + ...inputs: Inputs & ValidateEventProtocolBuilder<"internal", Inputs> + ): Machine.EventProtocol<"internal", Machine.EventProtocolInputSchemasOf<"internal", Inputs>> +} = internal.internalEvents as any + +/** + * Defines the ephemeral notifications a machine may publish to external + * observers. Emitted events are separate from actor input and are never sent + * implicitly to a parent actor. Observe them through `MachineRef.emissions` or + * the AtomMachine emission stream adapters. + * + * @category constructors + * @since 0.10.0 + */ +export const emittedEvents: { + >( + ...schemas: Schemas & ValidateEmittedEventProtocol> + ): Machine.EventProtocol<"emitted", readonly [...Schemas]> + >>( + ...inputs: Inputs & ValidateEventProtocolBuilder<"emitted", Inputs> + ): Machine.EventProtocol<"emitted", Machine.EventProtocolInputSchemasOf<"emitted", Inputs>> +} = internal.emittedEvents as any /** * Encodes a decoded machine snapshot into a normalized data representation. @@ -6181,7 +6416,8 @@ export const encodeSnapshot: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: Machine< States, @@ -6196,14 +6432,15 @@ export const encodeSnapshot: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents >, snapshot: Machine.Snapshot ) => Effect.Effect< Machine.EncodedSnapshot, MachineSchemaEncodeError, Machine.SnapshotEncodingServices -> = internal.encodeSnapshot +> = internal.encodeSnapshot as any /** * Decodes a normalized data representation into a validated machine snapshot. @@ -6262,7 +6499,8 @@ export const decodeSnapshot: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: Machine< States, @@ -6277,14 +6515,15 @@ export const decodeSnapshot: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents >, encoded: unknown ) => Effect.Effect< Machine.Snapshot, MachineSchemaDecodeError, Machine.SnapshotDecodingServices -> = internal.decodeSnapshot +> = internal.decodeSnapshot as any type DynamicEffectInvokeSource< States extends Machine.StateSchemas, @@ -6582,7 +6821,8 @@ export const invoke: { Machine.Error | ActionError>, Machine.Services, Machine.InitialError, - Machine.Emit + Machine.Emit, + Machine.EventOf> > } = ((config: unknown) => config) as any type RetagFields = Omit @@ -6689,7 +6929,8 @@ export const planInitial: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: & Machine< @@ -6705,7 +6946,8 @@ export const planInitial: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > & EnsureExecutable, ...args: [...Machine.InputArgs] @@ -6746,7 +6988,7 @@ export const planInitial: < ), InitialE | E | InfiniteTransitionError | MachineSchemaDecodeError | StartupError, never -> = internal.planInitial +> = internal.planInitial as any /** * Returns every compiled state node in definition order. @@ -6850,7 +7092,8 @@ export const enabled: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: Machine< States, @@ -6865,10 +7108,11 @@ export const enabled: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents >, state: Machine.Snapshot -) => ReadonlyArray> = internal.enabled +) => ReadonlyArray> = internal.enabled as any /** * Plans the next state snapshot synchronously. @@ -6927,7 +7171,8 @@ export const plan: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: & Machine< @@ -6943,7 +7188,8 @@ export const plan: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > & EnsureExecutable, state: Machine.Snapshot, @@ -6983,7 +7229,7 @@ export const plan: < ), E | InfiniteTransitionError | MachineSchemaDecodeError, never -> = internal.plan +> = internal.plan as any /** * Creates advanced stateful process logic from explicit initialization and @@ -7247,7 +7493,8 @@ export const start: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: & Machine< @@ -7263,7 +7510,8 @@ export const start: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > & EnsureExecutable, ...args: [...Machine.InputArgs] @@ -7276,7 +7524,8 @@ export const start: < | InfiniteTransitionError | MachineSchemaDecodeError | StoppedError, - Output + Output, + Machine.EmittedEventOf >, | InitialE | E @@ -7290,7 +7539,7 @@ export const start: < Machine.EventOf, Machine.EmitOf > -> = internal.start +> = internal.start as any /** * Starts a fresh managed runtime from a decoded logical snapshot. @@ -7356,7 +7605,8 @@ export const resume: < FinalStates extends Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: & Machine< @@ -7372,7 +7622,8 @@ export const resume: < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > & EnsureExecutable, snapshot: Machine.Snapshot @@ -7385,7 +7636,8 @@ export const resume: < | InfiniteTransitionError | MachineSchemaDecodeError | StoppedError, - Output + Output, + Machine.EmittedEventOf >, MachineSchemaDecodeError, ExcludeCompatibleRuntime< @@ -7393,4 +7645,4 @@ export const resume: < Machine.EventOf, Machine.EmitOf > -> = internal.resume +> = internal.resume as any diff --git a/src/internal/machine/atom.ts b/src/internal/machine/atom.ts index 1a26427..682e65b 100644 --- a/src/internal/machine/atom.ts +++ b/src/internal/machine/atom.ts @@ -72,9 +72,9 @@ type MachineStartError = | Machine.StoppedError | RuntimeError -const runMachineAtomEffect = ( +const runMachineAtomEffect = ( get: Atom.AtomContext, - start: Effect.Effect, StartError, Requirements> + start: Effect.Effect, StartError, Requirements> ): Effect.Effect => Effect.scoped( Effect.acquireRelease(start, (ref) => ref.stop).pipe( @@ -96,7 +96,8 @@ const startMachineAtomEffect = < FinalStates extends Machine.Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( get: Atom.AtomContext, machine: @@ -113,7 +114,8 @@ const startMachineAtomEffect = < Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > & EnsureExecutable, args: [...Machine.Machine.InputArgs] @@ -122,11 +124,12 @@ const startMachineAtomEffect = < Machine.Machine.Snapshot, Machine.Machine.EventInputOf, MachineRuntimeError, - Output + Output, + Machine.Machine.EmittedEventOf >, MachineStartError, MachineRequirements, Machine.Machine.EmitOf> -> => runMachineAtomEffect(get, internalMachine.start(machine, ...args)) +> => runMachineAtomEffect(get, internalMachine.start(machine as any, ...args) as any) const resumeMachineAtomEffect = ( get: Atom.AtomContext, @@ -137,6 +140,24 @@ const resumeMachineAtomEffect = ( type RefState = Ref extends Machine.MachineRef ? State : never type RefError = Ref extends Machine.MachineRef ? Error : never type RefOutput = Ref extends Machine.MachineRef ? Output : never +type RefEmitted = Ref extends Machine.MachineRef ? Emitted : never + +export const emissions = ( + self: MachineAtom +): Stream.Stream => + Atom.toStreamResult(self.ref).pipe(Stream.flatMap((ref) => ref.emissions)) + +export const childEmissions = ( + self: ChildMachineAtom +): Stream.Stream>, StartError, AtomRegistry.AtomRegistry> => + Atom.toStreamResult(self.ref).pipe( + Stream.flatMap( + Option.match({ + onNone: () => Stream.empty, + onSome: (ref) => ref.emissions + }) + ) + ) const makeRuntimeResultAtom = ( snapshot: Atom.Atom, StartError>> @@ -588,7 +609,8 @@ type ResumedMachineAtomOf = Machine Machine.Machine.EventInput>, MachineRuntimeError, Machine.Machine.Services>, Machine.Machine.Output, - Machine.MachineSchemaDecodeError | RuntimeError + Machine.MachineSchemaDecodeError | RuntimeError, + Machine.Machine.EmittedEvent > export const make: { @@ -605,7 +627,8 @@ export const make: { FinalStates extends Machine.Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: & Machine.Machine< @@ -621,7 +644,8 @@ export const make: { Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > & EnsureNoExternalRequirements< MachineRequirements< @@ -638,7 +662,8 @@ export const make: { Machine.Machine.EventInputOf, MachineRuntimeError, Output, - MachineStartError + MachineStartError, + Machine.Machine.EmittedEventOf > } = ((machine: Machine.Machine.Any, ...args: ReadonlyArray) => { const ref = Atom.make((get) => startMachineAtomEffect(get, machine as any, args as [])) diff --git a/src/internal/machine/command.ts b/src/internal/machine/command.ts index 63d19ea..915dff2 100644 --- a/src/internal/machine/command.ts +++ b/src/internal/machine/command.ts @@ -47,8 +47,8 @@ export const makeCollector = (machine: Machine.Any): Collected => emit: (event) => { emittedEvents.push(decodeEmitSync(machine, event)) }, - sendTo: (child: unknown, event: unknown) => { - commands.push({ _tag: "SendTo", child: child as any, event }) + sendTo: (target: unknown, event: unknown) => { + commands.push({ _tag: "SendTo", target: target as any, event }) }, stop: (child: unknown) => { commands.push({ _tag: "Stop", child: child as any }) diff --git a/src/internal/machine/commandRuntime.ts b/src/internal/machine/commandRuntime.ts index 9ab3baa..5cbc202 100644 --- a/src/internal/machine/commandRuntime.ts +++ b/src/internal/machine/commandRuntime.ts @@ -10,6 +10,9 @@ import type { RuntimeCommand } from "./command.js" import { decodeEmit, decodeEvent } from "./protocol.js" import type { ProcessScope } from "./runtime.js" +const isActorRef = (target: unknown): target is { readonly send: (event: unknown) => Effect.Effect } => + typeof target === "object" && target !== null && "send" in target && typeof target.send === "function" + export const makeLiveRuntime = ( machine: Machine.Any, scope: ProcessScope @@ -18,9 +21,9 @@ export const makeLiveRuntime = ( decodeEvent(machine, event).pipe( Effect.flatMap((event) => scope.self.send(event as Events)) ), - sendParent: (event) => + emit: (event) => decodeEmit(machine, event).pipe( - Effect.flatMap((event) => scope.sendParent(event)) + Effect.flatMap((event) => scope.emit(event)) ) }) @@ -30,7 +33,9 @@ export const runCommands = ( ) => Effect.forEach(commands, (command) => command._tag === "SendTo" - ? scope.sendTo(command.child as never, command.event) + ? isActorRef(command.target) + ? command.target.send(command.event) + : scope.sendTo(command.target as never, command.event) : scope.stopChild(command.child as never), { discard: true }) export const runEmittedEvents = ( @@ -38,6 +43,6 @@ export const runEmittedEvents = ( runtime: Runtime ) => Effect.all( - Array.from(events, (event) => runtime.sendParent(event)), + Array.from(events, (event) => runtime.emit(event)), { discard: true } ) diff --git a/src/internal/machine/configuration.ts b/src/internal/machine/configuration.ts index e7e34cc..284a27f 100644 --- a/src/internal/machine/configuration.ts +++ b/src/internal/machine/configuration.ts @@ -8,7 +8,7 @@ import * as Cause from "effect/Cause" import * as Effect from "effect/Effect" import * as Option from "effect/Option" import { hasProperty } from "effect/Predicate" -import type { Machine } from "../../Machine.js" +import type { ActorRef, Machine } from "../../Machine.js" import { MachineSchemaDecodeError } from "./errors.js" import { decodeBoundary, @@ -252,8 +252,26 @@ export interface ActiveConfiguration { readonly values: ReadonlyMap readonly outputs: ReadonlyMap readonly history: ReadonlyMap + readonly actorScope?: PlanningActorScope } +export interface PlanningActorScope { + readonly self: ActorRef + readonly parent: ActorRef | undefined +} + +export const withActorScope = ( + configuration: ActiveConfiguration, + actorScope: PlanningActorScope +): ActiveConfiguration => ({ + ...configuration, + actorScope: { self: actorScope.self, parent: actorScope.parent } +}) + +export const getActorScope = ( + configuration: ActiveConfiguration +): PlanningActorScope | undefined => configuration.actorScope + export interface FinalCompletion { readonly path: string readonly output: unknown @@ -1178,8 +1196,8 @@ export const resolveFinalOutputEffect: < const node = getNode(machine, path) const output = getStateConfigByPath(machine, path)?.output?.({ state: configuration.values.get(path), - parent: getParentValue(machine, configuration, path), - parents: getParentValues(machine, configuration, path), + containingState: getParentValue(machine, configuration, path), + ancestors: getParentValues(machine, configuration, path), event, outputs } as any) @@ -1331,8 +1349,8 @@ const resolveFinalOutputSync = diff --git a/src/internal/machine/executionPlan.ts b/src/internal/machine/executionPlan.ts index 84cfd1a..b8553f5 100644 --- a/src/internal/machine/executionPlan.ts +++ b/src/internal/machine/executionPlan.ts @@ -4,7 +4,8 @@ * @since 0.4.0 */ -import type { Machine } from "../../Machine.js" +import * as Effect from "effect/Effect" +import type { ActorRef, Machine } from "../../Machine.js" import { getTargetBuilder, type RuntimeCommand } from "./command.js" import { type ActiveConfiguration, @@ -17,10 +18,12 @@ import { isDescendantOf, normalizeConfigurationSync, normalizeTargetConfigurationSync, + type PlanningActorScope, snapshotFromConfiguration, - validateInitialConfiguration + validateInitialConfiguration, + withActorScope } from "./configuration.js" -import { InfiniteTransitionError } from "./errors.js" +import { InfiniteTransitionError, StoppedError } from "./errors.js" import * as InvocationEvent from "./invocationEvent.js" import { broadenTransitionBoundary, @@ -64,6 +67,29 @@ interface IndexedExecutionDescriptor { > } +const planningActorScopes = new WeakMap() + +const getPlanningActorScope = (machine: Machine.Any): PlanningActorScope => { + const cached = planningActorScopes.get(machine) + if (cached !== undefined) return cached + const self: ActorRef = { + id: machine.id ?? "Machine", + sessionId: "Machine.plan", + send: () => Effect.fail(new StoppedError()) + } + const scope = { self, parent: undefined } + planningActorScopes.set(machine, scope) + return scope +} + +const getActorContext = ( + machine: Machine.Any, + actorScope: PlanningActorScope | undefined +): PlanningActorScope => + actorScope === undefined + ? getPlanningActorScope(machine) + : { self: actorScope.self, parent: actorScope.parent } + const indexedStateConfigKeys: ReadonlySet = new Set([ "initial", "invoke", @@ -336,18 +362,20 @@ const makeIndexedTransitionContext = ( descriptor: IndexedExecutionDescriptor, configuration: OwnedIndexedState, sourceIndex: number, - event: any + event: any, + actorScope?: PlanningActorScope ): any => { const source = descriptor.nodes[sourceIndex]! const parentIndex = descriptor.parentIndices[sourceIndex]! - const parents: Record = {} + const ancestors: Record = {} for (const ancestorIndex of descriptor.ancestorIndices[sourceIndex]!) { - parents[descriptor.nodes[ancestorIndex]!.path] = configuration.values[ancestorIndex] + ancestors[descriptor.nodes[ancestorIndex]!.path] = configuration.values[ancestorIndex] } return { + ...getActorContext(machine, actorScope), state: configuration.values[sourceIndex], - parent: parentIndex < 0 ? undefined : configuration.values[parentIndex], - parents, + containingState: parentIndex < 0 ? undefined : configuration.values[parentIndex], + ancestors, event, snapshot: snapshotFromIndexedState(descriptor, configuration), target: getTargetBuilder(machine, source.path) @@ -407,7 +435,7 @@ const collectIndexedTransition = ( ;(emittedEvents ??= []).push(decodeEmitSync(machine, event)) }, sendTo: (child: unknown, event: unknown) => { - ;(commands ??= []).push({ _tag: "SendTo", child: child as any, event }) + ;(commands ??= []).push({ _tag: "SendTo", target: child as any, event }) }, stop: (child: unknown) => { ;(commands ??= []).push({ _tag: "Stop", child: child as any }) @@ -425,7 +453,8 @@ const selectIndexedEventTransitions = ( machine: Machine.Any, descriptor: IndexedExecutionDescriptor, configuration: OwnedIndexedState, - event: any + event: any, + actorScope?: PlanningActorScope ): ReadonlyArray => { const selected: Array = [] for (const leafIndex of configuration.activeLeaves) { @@ -446,7 +475,8 @@ const selectIndexedEventTransitions = ( descriptor, configuration, sourceIndex, - event + event, + actorScope ) }) } @@ -629,7 +659,8 @@ const planIndexedFlatState = ( descriptor: IndexedExecutionDescriptor, configuration: OwnedIndexedState, decoded: { readonly _tag: PropertyKey }, - retainMicrosteps: boolean + retainMicrosteps: boolean, + actorScope?: PlanningActorScope ): ExecutionMacrostep => { let current = configuration let event: any = decoded @@ -681,9 +712,10 @@ const planIndexedFlatState = ( machine, transition.transition, { + ...getActorContext(machine, actorScope), state: current.values[sourceIndex], - parent: undefined, - parents: {}, + containingState: undefined, + ancestors: {}, event, snapshot: snapshotFromIndexedState(descriptor, current), target: getTargetBuilder(machine, sourcePath) @@ -774,7 +806,8 @@ const planIndexedState = ( descriptor: IndexedExecutionDescriptor, configuration: OwnedIndexedState, input: unknown, - retainMicrosteps: boolean + retainMicrosteps: boolean, + actorScope?: PlanningActorScope ): ExecutionMacrostep => { if (InvocationEvent.isInvocationEvent(input)) { // Invocation lifecycle transitions retain the generic planner as their @@ -783,7 +816,9 @@ const planIndexedState = ( // the representation boundary. const planned = planConfiguration( machine as any, - activeConfigurationFromIndexedState(descriptor, configuration), + actorScope === undefined + ? activeConfigurationFromIndexedState(descriptor, configuration) + : withActorScope(activeConfigurationFromIndexedState(descriptor, configuration), actorScope), input ) return { @@ -806,7 +841,7 @@ const planIndexedState = ( } const decoded = decodeEventSync(machine, input) if (descriptor.flat) { - return planIndexedFlatState(machine, descriptor, configuration, decoded, retainMicrosteps) + return planIndexedFlatState(machine, descriptor, configuration, decoded, retainMicrosteps, actorScope) } if (descriptor.finalIndices.some((index) => configuration.active[index] === 1)) { const active = activeConfigurationFromIndexedState(descriptor, configuration) @@ -827,7 +862,7 @@ const planIndexedState = ( } } - const selections = selectIndexedEventTransitions(machine, descriptor, configuration, decoded) + const selections = selectIndexedEventTransitions(machine, descriptor, configuration, decoded, actorScope) if (selections.length === 0) { return { next: configuration, @@ -895,7 +930,7 @@ const planIndexedState = ( } raisedIndex += 1 currentEvent = raised - const raisedSelections = selectIndexedEventTransitions(machine, descriptor, current, raised) + const raisedSelections = selectIndexedEventTransitions(machine, descriptor, current, raised, actorScope) if (raisedSelections.length === 0) continue const step = indexedMicrostep(machine, descriptor, current, raised, raisedSelections) current = step.next @@ -913,10 +948,12 @@ export interface CompiledExecutionPlan { readonly plan: ( state: unknown, event: unknown, - retainMicrosteps?: boolean + retainMicrosteps?: boolean, + actorScope?: PlanningActorScope ) => ExecutionMacrostep readonly initial?: ( - args: ReadonlyArray + args: ReadonlyArray, + actorScope?: PlanningActorScope ) => { readonly state: Machine.Snapshot readonly configuration: unknown @@ -933,7 +970,14 @@ const makeActiveExecutionPlan = (machine: Machine.Any): CompiledExecutionPlan => fromConfiguration: (configuration) => configuration, toConfiguration: (state) => state as ActiveConfiguration, snapshot: (state) => snapshotFromConfiguration(machine, state as ActiveConfiguration), - plan: (state, event) => planConfiguration(machine as any, state as ActiveConfiguration, event as any) + plan: (state, event, _retainMicrosteps, actorScope) => + planConfiguration( + machine as any, + actorScope === undefined + ? state as ActiveConfiguration + : withActorScope(state as ActiveConfiguration, actorScope), + event as any + ) }) const makeIndexedExecutionPlan = ( @@ -943,16 +987,17 @@ const makeIndexedExecutionPlan = ( fromConfiguration: (configuration) => ownedIndexedStateFromActive(indexed, configuration), toConfiguration: (state) => activeConfigurationFromIndexedState(indexed, state as OwnedIndexedState), snapshot: (state) => snapshotFromIndexedState(indexed, state as OwnedIndexedState), - plan: (state, event, retainMicrosteps = false) => - planIndexedState(machine, indexed, state as OwnedIndexedState, event, retainMicrosteps), - initial: (args) => { + plan: (state, event, retainMicrosteps = false, actorScope) => + planIndexedState(machine, indexed, state as OwnedIndexedState, event, retainMicrosteps, actorScope), + initial: (args, actorScope) => { const inputArgs = machine.input === undefined ? args : args.length === 0 ? (decodeInputSync(machine, machine.input, undefined), args) : [decodeInputSync(machine, machine.input, args[0])] const initial = machine.initial(...inputArgs as any) - const active = normalizeConfigurationSync(machine, initial as Machine.Snapshot) + const normalized = normalizeConfigurationSync(machine, initial as Machine.Snapshot) + const active = actorScope === undefined ? normalized : withActorScope(normalized, actorScope) validateInitialConfiguration(machine, active) const completed = completeConfigurationSync(machine, active, InitialEvent).configuration const configuration = ownedIndexedStateFromActive(indexed, completed) diff --git a/src/internal/machine/invocation.ts b/src/internal/machine/invocation.ts index 0ef6a48..8cc8bd3 100644 --- a/src/internal/machine/invocation.ts +++ b/src/internal/machine/invocation.ts @@ -235,9 +235,10 @@ export const startAll = ( .filter((path) => configuration.active.has(path)) .flatMap((path) => { const context = { + ...(Configuration.getActorScope(configuration) ?? { self: scope.self, parent: scope.parent }), state: configuration.values.get(path), - parent: Configuration.getParentValue(machine, configuration, path), - parents: Configuration.getParentValues(machine, configuration, path), + containingState: Configuration.getParentValue(machine, configuration, path), + ancestors: Configuration.getParentValues(machine, configuration, path), event } return InvocationEvent.definitions(Configuration.getStateConfigByPath(machine, path)?.invoke).map((definition) => diff --git a/src/internal/machine/machine.ts b/src/internal/machine/machine.ts index 3dba2c2..491724a 100644 --- a/src/internal/machine/machine.ts +++ b/src/internal/machine/machine.ts @@ -112,7 +112,8 @@ const cloneWithHandlers = ( machine.states = self.states machine.events = self.events machine.internalEvents = self.internalEvents - machine.emits = self.emits + machine.emittedEvents = self.emittedEvents + machine.parentEvents = self.parentEvents machine.input = self.input machine.id = self.id machine.initial = self.initial @@ -766,7 +767,8 @@ type MakeConfig< Input extends Schema.Top, InitialE, InitialR, - InternalEvents extends ReadonlyArray + InternalEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray > = { readonly id?: string readonly states: States & DefineStateTreeInput> @@ -779,7 +781,8 @@ type MakeConfig< NoInfer, NoInfer > - readonly emits?: Emits + readonly emittedEvents?: Machine.EventProtocol<"emitted", Emits> + readonly parentEvents?: Machine.EventProtocol<"public", ParentEvents> readonly input?: Input readonly initial: (...args: [...Machine.InputArgs]) => Machine.InitialResult } @@ -791,7 +794,8 @@ type MakeResult< Input extends Schema.Top, InitialE, InitialR, - InternalEvents extends ReadonlyArray + InternalEvents extends ReadonlyArray, + ParentEvents extends ReadonlyArray > = Machine< States, readonly [...InputEvents, ...InternalEvents], @@ -805,7 +809,8 @@ type MakeResult< Machine.TerminalOutput, Emits, never, - InputEvents + InputEvents, + ParentEvents > interface Make { @@ -816,11 +821,12 @@ interface Make { const Input extends Schema.Top = typeof Schema.Void, InitialE = never, InitialR = never, - const InternalEvents extends ReadonlyArray = readonly [] + const InternalEvents extends ReadonlyArray = readonly [], + const ParentEvents extends ReadonlyArray = readonly [] >( - config: MakeConfig, + config: MakeConfig, ..._validation: ValidateDefinedStates> - ): MakeResult + ): MakeResult < const States extends Machine.StateSchemas, const InputEvents extends ReadonlyArray, @@ -828,10 +834,11 @@ interface Make { const Input extends Schema.Top = typeof Schema.Void, InitialE = never, InitialR = never, - const InternalEvents extends ReadonlyArray = readonly [] + const InternalEvents extends ReadonlyArray = readonly [], + const ParentEvents extends ReadonlyArray = readonly [] >( config: - & Omit, "states"> + & Omit, "states"> & { readonly states: InvalidDefinedStateTreeInput } ): never } @@ -843,24 +850,27 @@ export const make: Make = (< const Input extends Schema.Top = typeof Schema.Void, InitialE = never, InitialR = never, - const InternalEvents extends ReadonlyArray = readonly [] + const InternalEvents extends ReadonlyArray = readonly [], + const ParentEvents extends ReadonlyArray = readonly [] >( config: { readonly id?: string readonly states: States readonly events: Machine.EventProtocol<"public", InputEvents> readonly internalEvents?: Machine.EventProtocol<"internal", InternalEvents> - readonly emits?: Emits + readonly emittedEvents?: Machine.EventProtocol<"emitted", Emits> + readonly parentEvents?: Machine.EventProtocol<"public", ParentEvents> readonly input?: Input readonly initial: (...args: [...Machine.InputArgs]) => Machine.InitialResult } -): MakeResult => { +): MakeResult => { StateDefinition.validateStateDefinitions(config.states, "Machine.make") const self = Object.create(Proto) self.states = config.states self.events = config.events self.internalEvents = config.internalEvents ?? Protocol.makeEventProtocol("internal", [] as const) - self.emits = config.emits ?? [] + self.emittedEvents = config.emittedEvents ?? Protocol.makeEventProtocol("emitted", [] as const) + self.parentEvents = config.parentEvents ?? Protocol.makeEventProtocol("public", [] as const) self.input = config.input self.id = config.id self.initial = config.initial @@ -872,15 +882,39 @@ export const make: Make = (< return self }) as Make -export const events = >( - ...schemas: Schemas -): Machine.EventProtocol<"public", readonly [...Schemas]> => - Protocol.makeEventProtocol<"public", readonly [...Schemas]>("public", schemas) +const flattenEventProtocolInputs = ( + kind: Kind, + inputs: ReadonlyArray> +): ReadonlyArray => + inputs.flatMap((input) => + Protocol.isEventProtocol(input, kind) + ? Protocol.eventProtocolSchemas(input) + : [input as Machine.TaggedSchema] + ) + +export const events = >>( + ...inputs: Inputs +): Machine.EventProtocol<"public", Machine.EventProtocolInputSchemasOf<"public", Inputs>> => + Protocol.makeEventProtocol( + "public", + flattenEventProtocolInputs("public", inputs) + ) as Machine.EventProtocol<"public", Machine.EventProtocolInputSchemasOf<"public", Inputs>> -export const internalEvents = >( - ...schemas: Schemas -): Machine.EventProtocol<"internal", readonly [...Schemas]> => - Protocol.makeEventProtocol<"internal", readonly [...Schemas]>("internal", schemas) +export const internalEvents = >>( + ...inputs: Inputs +): Machine.EventProtocol<"internal", Machine.EventProtocolInputSchemasOf<"internal", Inputs>> => + Protocol.makeEventProtocol( + "internal", + flattenEventProtocolInputs("internal", inputs) + ) as Machine.EventProtocol<"internal", Machine.EventProtocolInputSchemasOf<"internal", Inputs>> + +export const emittedEvents = >>( + ...inputs: Inputs +): Machine.EventProtocol<"emitted", Machine.EventProtocolInputSchemasOf<"emitted", Inputs>> => + Protocol.makeEventProtocol( + "emitted", + flattenEventProtocolInputs("emitted", inputs) + ) as Machine.EventProtocol<"emitted", Machine.EventProtocolInputSchemasOf<"emitted", Inputs>> export const encodeSnapshot: < const States extends Machine.StateSchemas, @@ -1373,7 +1407,8 @@ export const start: < | InfiniteTransitionError | MachineSchemaDecodeError | StoppedError, - Output + Output, + Machine.EmittedEventOf >, | InitialE | E @@ -1431,7 +1466,8 @@ export const resume: < | InfiniteTransitionError | MachineSchemaDecodeError | StoppedError, - Output + Output, + Machine.EmittedEventOf >, MachineSchemaDecodeError, ExcludeCompatibleRuntime< diff --git a/src/internal/machine/planner.ts b/src/internal/machine/planner.ts index 49f3d61..16a00d0 100644 --- a/src/internal/machine/planner.ts +++ b/src/internal/machine/planner.ts @@ -7,7 +7,7 @@ import * as Cause from "effect/Cause" import * as Effect from "effect/Effect" import type * as Schema from "effect/Schema" -import type { Enqueue, InitialEvent as MachineInitialEvent, Machine } from "../../Machine.js" +import type { ActorContext, ActorRef, Enqueue, InitialEvent as MachineInitialEvent, Machine } from "../../Machine.js" import { getTargetBuilder, makeCollector, type RuntimeCommand } from "./command.js" import { type ActiveConfiguration, @@ -37,7 +37,7 @@ import { snapshotFromConfigurationAtPath, validateInitialConfiguration } from "./configuration.js" -import { InfiniteTransitionError, MachineSchemaDecodeError, StartupError } from "./errors.js" +import { InfiniteTransitionError, MachineSchemaDecodeError, StartupError, StoppedError } from "./errors.js" import * as InvocationEvent from "./invocationEvent.js" import { decodeEventSync, decodeInputSync, decodeStateValueSync } from "./protocol.js" import { InitialEventTypeId } from "./symbols.js" @@ -93,6 +93,38 @@ export type TransitionHandler ) => Machine.HandlerResult +const rootActorScopes = new WeakMap>() +const machineActorScopes = new WeakMap>() + +export const withActorScope = ( + machine: Machine.Any, + actorScope: ActorContext +): Machine.Any => { + const scoped = Object.create(machine) as Machine.Any + machineActorScopes.set(scoped, { self: actorScope.self, parent: actorScope.parent }) + return scoped +} + +const getActorContext = ( + machine: Machine.Any, + configuration: ActiveConfiguration +): ActorContext => { + const scope = configuration.actorScope + if (scope !== undefined) return scope + const machineScope = machineActorScopes.get(machine) + if (machineScope !== undefined) return machineScope + const cached = rootActorScopes.get(machine) + if (cached !== undefined) return cached + const self: ActorRef = { + id: machine.id ?? "Machine", + sessionId: "Machine.plan", + send: () => Effect.fail(new StoppedError()) + } + const root = { self, parent: undefined } + rootActorScopes.set(machine, root) + return root +} + type EventTransition = | TransitionHandler | { @@ -206,7 +238,8 @@ const completeHistoryConfiguration = ( active, values, outputs: configuration.outputs, - history: configuration.history + history: configuration.history, + actorScope: configuration.actorScope } as ActiveConfiguration active.add(child.path) if (child.schema !== undefined) { @@ -215,9 +248,10 @@ const completeHistoryConfiguration = ( throw new Error(`Machine shallow history requires an initial value implementation for state "${path}"`) } const initialized = collectStateInitializer(machine, initializer, { + ...getActorContext(machine, current), state: current.values.get(path), - parent: getParentValue(machine, current, path), - parents: getParentValues(machine, current, path), + containingState: getParentValue(machine, current, path), + ancestors: getParentValues(machine, current, path), event }) values.set(child.path, decodeStateValueSync(machine, child, initialized.value)) @@ -235,16 +269,18 @@ const completeHistoryConfiguration = ( active, values, outputs: configuration.outputs, - history: configuration.history + history: configuration.history, + actorScope: configuration.actorScope } as ActiveConfiguration const initializer = valuedMissing.length === 0 ? undefined : machine.handlers[path]?.initial if (valuedMissing.length > 0 && initializer === undefined) { throw new Error(`Machine shallow history requires an initial value implementation for state "${path}"`) } const initialized = initializer === undefined ? undefined : collectStateInitializer(machine, initializer, { + ...getActorContext(machine, current), state: current.values.get(path), - parent: getParentValue(machine, current, path), - parents: getParentValues(machine, current, path), + containingState: getParentValue(machine, current, path), + ancestors: getParentValues(machine, current, path), event }) if (initialized !== undefined && (typeof initialized.value !== "object" || initialized.value === null)) { @@ -328,7 +364,7 @@ const resolveHistoryTarget = ( const collected = collectTransition(machine, fallback, { event, target: getTargetBuilder(machine, target.parent).full, - parent: target.parent + owner: target.parent }) if (collected.state === undefined || isHistoryTarget(collected.state) || !isSnapshot(collected.state)) { throw new Error(`Machine history default for "${target.path}" must return a complete snapshot containing its owner`) @@ -505,9 +541,10 @@ const makeStateActionContext = < path: string, event: Machine.LifecycleEvent ): Machine.StateActionContext => ({ + ...getActorContext(machine, configuration), state: configuration.values.get(path) as Machine.StateByIdentifier, - parent: getParentValue(machine, configuration, path) as Machine.ParentStateValue, - parents: getParentValues(machine, configuration, path) as Machine.ParentStateValues, + containingState: getParentValue(machine, configuration, path) as Machine.ParentStateValue, + ancestors: getParentValues(machine, configuration, path) as Machine.ParentStateValues, event }) @@ -524,9 +561,10 @@ const makeTransitionContext = < event: Machine.EventByTag, snapshot: Machine.Snapshot ): Machine.HandlerContext => ({ + ...getActorContext(machine, configuration), state: configuration.values.get(path) as Machine.StateByIdentifier, - parent: getParentValue(machine, configuration, path) as Machine.ParentStateValue, - parents: getParentValues(machine, configuration, path) as Machine.ParentStateValues, + containingState: getParentValue(machine, configuration, path) as Machine.ParentStateValue, + ancestors: getParentValues(machine, configuration, path) as Machine.ParentStateValues, event, snapshot, target: getTargetBuilder(machine, path) @@ -545,9 +583,10 @@ const makeDoneContext = < output: unknown, snapshot: Machine.Snapshot ): Machine.DoneContext => ({ + ...getActorContext(machine, configuration), state: configuration.values.get(path) as Machine.StateByIdentifier, - parent: getParentValue(machine, configuration, path) as Machine.ParentStateValue, - parents: getParentValues(machine, configuration, path) as Machine.ParentStateValues, + containingState: getParentValue(machine, configuration, path) as Machine.ParentStateValue, + ancestors: getParentValues(machine, configuration, path) as Machine.ParentStateValues, event, output: output as Machine.CompletionOutputByIdentifier, snapshot, @@ -639,15 +678,16 @@ const selectAlwaysTransitions = < Machine.AlwaysContext> >, context: { + ...getActorContext(machine, configuration), state: configuration.values.get(path) as Machine.StateByIdentifier< States, Machine.StateIdentifier >, - parent: getParentValue(machine, configuration, path) as Machine.ParentStateValue< + containingState: getParentValue(machine, configuration, path) as Machine.ParentStateValue< States, Machine.StateIdentifier >, - parents: getParentValues(machine, configuration, path) as Machine.ParentStateValues< + ancestors: getParentValues(machine, configuration, path) as Machine.ParentStateValues< States, Machine.StateIdentifier >, @@ -826,9 +866,10 @@ const selectInvocationTransition = < if (transition === undefined) return [] const snapshot = snapshotFromConfiguration(machine, configuration) const context = { + ...getActorContext(machine, configuration), state: configuration.values.get(event.path), - parent: getParentValue(machine, configuration, event.path), - parents: getParentValues(machine, configuration, event.path), + containingState: getParentValue(machine, configuration, event.path), + ancestors: getParentValues(machine, configuration, event.path), snapshot, target: getTargetBuilder(machine, event.path), id: event.id, @@ -1064,11 +1105,13 @@ const resolveChoiceTarget = ( ...Object.entries(extracted.values) ]), outputs: configuration.outputs, - history: configuration.history + history: configuration.history, + ...(configuration.actorScope === undefined ? {} : { actorScope: configuration.actorScope }) } const collected = collectTransition(machine, choice.transition, { - parent: getParentValue(machine, provisional, node.path), - parents: getParentValues(machine, provisional, node.path), + ...getActorContext(machine, provisional), + containingState: getParentValue(machine, provisional, node.path), + ancestors: getParentValues(machine, provisional, node.path), event, target: getTargetBuilder(machine, node.path) }) diff --git a/src/internal/machine/process.ts b/src/internal/machine/process.ts index 3c30e7e..47072d1 100644 --- a/src/internal/machine/process.ts +++ b/src/internal/machine/process.ts @@ -131,7 +131,8 @@ const makeChildlessCompiledDrain = ( planned = executionPlan.plan( configuration ?? executionPlan.fromConfiguration(Configuration.normalizeConfigurationSync(machine, current)), event, - acknowledged + acknowledged, + context.scope ) } catch (error) { return error instanceof InfiniteTransitionError || error instanceof MachineSchemaDecodeError @@ -255,7 +256,8 @@ const makeInvokingCompiledDrain = ( configuration ?? executionPlan.fromConfiguration(Configuration.normalizeConfigurationSync(machine, current)), event, - acknowledged + acknowledged, + scope ) } catch (error) { return error instanceof InfiniteTransitionError || error instanceof MachineSchemaDecodeError @@ -426,9 +428,11 @@ const makeProcessLogic: < const executionPlan = ExecutionPlan.compileExecutionPlan(machine) const initialArgs = entry._tag === "Initial" ? entry.args : [] const compiledInitial = entry._tag === "Initial" ? executionPlan.initial : undefined - const makeCompiledInitial = compiledInitial === undefined ? undefined : () => { + const makeCompiledInitial = compiledInitial === undefined ? undefined : ( + scope: internalRuntime.ProcessScope> + ) => { try { - const planned = compiledInitial(initialArgs) + const planned = compiledInitial(initialArgs, scope) const result = { state: planned.state as Machine.Snapshot, done: planned.done, @@ -455,7 +459,7 @@ const makeProcessLogic: < ) => compiledInitial === undefined ? internalRuntime.provideMachineRuntime( - internalPlanner.planInitial(machine, ...initialArgs).pipe( + internalPlanner.planInitial(internalPlanner.withActorScope(machine, scope), ...initialArgs).pipe( Effect.flatMap((planned) => { const commands = planned.commands.length === 0 ? undefined @@ -480,7 +484,7 @@ const makeProcessLogic: < ), scope ) - : Effect.try({ try: makeCompiledInitial!, catch: (error) => error as any }) + : Effect.try({ try: () => makeCompiledInitial!(scope), catch: (error) => error as any }) return ({ execution: { _tag: "Compiled", @@ -536,7 +540,10 @@ const makeProcessLogic: < try { planned = internalPlanner.planConfiguration( machine, - configuration ?? Configuration.normalizeConfigurationSync(machine, current), + Configuration.withActorScope( + configuration ?? Configuration.normalizeConfigurationSync(machine, current), + context + ), event ) } catch (error) { @@ -644,7 +651,10 @@ const makeProcessLogic: < try { planned = internalPlanner.planConfiguration( machine, - configuration ?? Configuration.normalizeConfigurationSync(machine, current), + Configuration.withActorScope( + configuration ?? Configuration.normalizeConfigurationSync(machine, current), + context + ), event ) } catch (error) { diff --git a/src/internal/machine/protocol.ts b/src/internal/machine/protocol.ts index e485ada..fcebcd2 100644 --- a/src/internal/machine/protocol.ts +++ b/src/internal/machine/protocol.ts @@ -15,7 +15,7 @@ import { MachineSchemaDecodeError } from "./errors.js" import { getStateNodeSchema, isStateInput } from "./topology.js" export interface DecodeBoundaryOptions { - readonly boundary: "input" | "event" | "emit" | "state" | "output" | "history" | "configuration" + readonly boundary: "input" | "event" | "emission" | "state" | "output" | "history" | "configuration" readonly state?: string readonly event?: string } @@ -25,6 +25,8 @@ interface MachineProtocolSchemas { readonly emit: Schema.Top readonly eventConstructors: ReadonlySet readonly trustedEvents: WeakSet + readonly emissionConstructors: ReadonlySet + readonly trustedEmissions: WeakSet } interface EventProtocolDefinition { @@ -42,6 +44,7 @@ interface EventConstruction { } interface EventConstructionDefinition { + readonly kind: Machine.EventProtocolKind readonly schema: Machine.TaggedSchema readonly input: unknown readonly inputError?: unknown @@ -124,7 +127,9 @@ const getEventProtocolDefinition = ( ): EventProtocolDefinition => { const definition = eventProtocolDefinitions.get(protocol as object) if (definition === undefined) { - throw new Error("Machine expected an event protocol created with Machine.events or Machine.internalEvents") + throw new Error( + "Machine expected an event protocol created with Machine.events, Machine.internalEvents, or Machine.emittedEvents" + ) } if (expectedKind !== undefined && definition.kind !== expectedKind) { throw new Error(`Machine expected a ${expectedKind} event protocol`) @@ -132,6 +137,12 @@ const getEventProtocolDefinition = ( return definition } +export const isEventProtocol = ( + value: unknown, + kind: Kind +): value is Machine.EventProtocol.Any => + typeof value === "object" && value !== null && eventProtocolDefinitions.get(value)?.kind === kind + export const eventProtocolSchemas = >( protocol: Machine.EventProtocol ): Schemas => getEventProtocolDefinition(protocol).schemas as Schemas @@ -142,9 +153,13 @@ export const inputEventSchemas = (machine: Machine.Any): ReadonlyArray => getEventProtocolDefinition(machine.internalEvents, "internal").schemas +export const emittedEventSchemas = (machine: Machine.Any): ReadonlyArray => + getEventProtocolDefinition(machine.emittedEvents, "emitted").schemas + export const setProtocol = (machine: Machine.Any): void => { const inputEvents = inputEventSchemas(machine) const localEvents = internalEventSchemas(machine) + const emittedEvents = emittedEventSchemas(machine) const publicTags = new Set(Reflect.ownKeys(machine.events)) for (const tag of Reflect.ownKeys(machine.internalEvents)) { if (publicTags.has(tag)) { @@ -154,9 +169,11 @@ export const setProtocol = (machine: Machine.Any): void => { const events = [...inputEvents, ...localEvents] setProtocolSchemas(machine, { event: Schema.Union(events), - emit: Schema.Union(machine.emits), + emit: Schema.Union(emittedEvents), eventConstructors: collectEventConstructors(events), - trustedEvents: new WeakSet() + trustedEvents: new WeakSet(), + emissionConstructors: collectEventConstructors(emittedEvents), + trustedEmissions: new WeakSet() }) } @@ -167,6 +184,7 @@ export const getEventName = (event: unknown): string | undefined => hasProperty(event, "_tag") ? String(event._tag) : undefined const makeEventConstruction = ( + kind: Machine.EventProtocolKind, schema: Machine.TaggedSchema, tag: PropertyKey, input: unknown @@ -186,6 +204,7 @@ const makeEventConstruction = ( _tag: tag } eventConstructionDefinitions.set(construction, { + kind, schema, input: ownedInput, ...(inputError === undefined ? {} : { inputError }) @@ -197,6 +216,7 @@ const isEventConstruction = (value: unknown): value is EventConstruction => typeof value === "object" && value !== null && eventConstructionDefinitions.has(value) const eventConstructors = ( + kind: Machine.EventProtocolKind, schemas: ReadonlyArray ): Readonly) => EventConstruction>> => { const leaves: Array = [] @@ -252,7 +272,7 @@ const eventConstructors = ( } Object.defineProperty(constructors, tag, { value: (...args: ReadonlyArray) => - makeEventConstruction(schema, tag, args.length === 0 ? {} : args[0]), + makeEventConstruction(kind, schema, tag, args.length === 0 ? {} : args[0]), enumerable: true }) } @@ -268,7 +288,7 @@ export const makeEventProtocol = < schemas: Schemas ): Machine.EventProtocol => { const ownedSchemas = Object.freeze(Array.from(schemas)) as unknown as Schemas - const protocol = eventConstructors(ownedSchemas) as Machine.EventProtocol + const protocol = eventConstructors(kind, ownedSchemas) as Machine.EventProtocol eventProtocolDefinitions.set(protocol as object, { kind, schemas: ownedSchemas }) return Object.freeze(protocol) as Machine.EventProtocol } @@ -336,13 +356,14 @@ const makeBoundarySync = ( const eventConstructionProtocolError = ( machine: Machine.Any, - construction: EventConstruction + construction: EventConstruction, + boundary: "event" | "emission" ): MachineSchemaDecodeError => new MachineSchemaDecodeError({ machineId: machine.id, - boundary: "event", + boundary, event: String(construction._tag), - cause: Cause.die(new Error("Constructed event schema does not belong to the machine event protocol")) + cause: Cause.die(new Error(`Constructed ${boundary} schema does not belong to the machine ${boundary} protocol`)) }) const eventConstructionInput = ( @@ -358,11 +379,12 @@ const eventConstructionInput = ( const eventConstructionInputError = ( machine: Machine.Any, construction: EventConstruction, + boundary: "event" | "emission", cause: unknown ): MachineSchemaDecodeError => new MachineSchemaDecodeError({ machineId: machine.id, - boundary: "event", + boundary, event: String(construction._tag), cause: Cause.die(cause) }) @@ -370,52 +392,68 @@ const eventConstructionInputError = ( const decodeEventConstruction = ( machine: Machine.Any, protocol: MachineProtocolSchemas, - construction: EventConstruction + construction: EventConstruction, + boundary: "event" | "emission" ): Effect.Effect => { const definition = eventConstructionDefinitions.get(construction) - if (definition === undefined || !protocol.eventConstructors.has(definition.schema as object)) { - return Effect.fail(eventConstructionProtocolError(machine, construction)) + const constructors = boundary === "event" ? protocol.eventConstructors : protocol.emissionConstructors + if ( + definition === undefined || + (boundary === "emission" ? definition.kind !== "emitted" : definition.kind === "emitted") || + !constructors.has(definition.schema as object) + ) { + return Effect.fail(eventConstructionProtocolError(machine, construction, boundary)) } return Effect.try({ try: () => eventConstructionInput(construction, definition), - catch: (cause) => eventConstructionInputError(machine, construction, cause) + catch: (cause) => eventConstructionInputError(machine, construction, boundary, cause) }).pipe( Effect.flatMap((input) => definition.schema.makeEffect(input as never).pipe( Effect.mapError((cause) => new MachineSchemaDecodeError({ machineId: machine.id, - boundary: "event", + boundary, event: String(construction._tag), cause: new Schema.SchemaError(cause) }) ) ) ), - Effect.tap((event) => Effect.sync(() => protocol.trustedEvents.add(event as object))) + Effect.tap((event) => + Effect.sync(() => + (boundary === "event" ? protocol.trustedEvents : protocol.trustedEmissions).add(event as object) + ) + ) ) } const decodeEventConstructionSync = ( machine: Machine.Any, protocol: MachineProtocolSchemas, - construction: EventConstruction + construction: EventConstruction, + boundary: "event" | "emission" ): unknown => { const definition = eventConstructionDefinitions.get(construction) - if (definition === undefined || !protocol.eventConstructors.has(definition.schema as object)) { - throw eventConstructionProtocolError(machine, construction) + const constructors = boundary === "event" ? protocol.eventConstructors : protocol.emissionConstructors + if ( + definition === undefined || + (boundary === "emission" ? definition.kind !== "emitted" : definition.kind === "emitted") || + !constructors.has(definition.schema as object) + ) { + throw eventConstructionProtocolError(machine, construction, boundary) } let input: unknown try { input = eventConstructionInput(construction, definition) } catch (cause) { - throw eventConstructionInputError(machine, construction, cause) + throw eventConstructionInputError(machine, construction, boundary, cause) } const event = makeBoundarySync(machine, definition.schema, input, { - boundary: "event", + boundary, event: String(construction._tag) }) - protocol.trustedEvents.add(event as object) + ;(boundary === "event" ? protocol.trustedEvents : protocol.trustedEmissions).add(event as object) return event } @@ -435,7 +473,7 @@ export const decodeEvent = , MachineSchemaDecodeError> => { const protocol = getProtocolSchemas(machine) if (isEventConstruction(event)) { - return decodeEventConstruction(machine, protocol, event) as Effect.Effect< + return decodeEventConstruction(machine, protocol, event, "event") as Effect.Effect< Machine.EventOf, MachineSchemaDecodeError > @@ -458,7 +496,7 @@ export const decodeEventSync = => { const protocol = getProtocolSchemas(machine) if (isEventConstruction(event)) { - return decodeEventConstructionSync(machine, protocol, event) as Machine.EventOf + return decodeEventConstructionSync(machine, protocol, event, "event") as Machine.EventOf } if (isTrustedEvent(protocol, event)) { return event as Machine.EventOf @@ -476,12 +514,22 @@ export const decodeEmit = , MachineSchemaDecodeError> => { + const protocol = getProtocolSchemas(machine) + if (isEventConstruction(event)) { + return decodeEventConstruction(machine, protocol, event, "emission") as Effect.Effect< + Machine.EmitOf, + MachineSchemaDecodeError + > + } + if (typeof event === "object" && event !== null && protocol.trustedEmissions.has(event)) { + return Effect.succeed(event as Machine.EmitOf) + } const eventName = getEventName(event) return decodeBoundary>( machine, - getProtocolSchemas(machine).emit, + protocol.emit, event, - eventName === undefined ? { boundary: "emit" } : { boundary: "emit", event: eventName } + eventName === undefined ? { boundary: "emission" } : { boundary: "emission", event: eventName } ) } @@ -489,12 +537,19 @@ export const decodeEmitSync = => { + const protocol = getProtocolSchemas(machine) + if (isEventConstruction(event)) { + return decodeEventConstructionSync(machine, protocol, event, "emission") as Machine.EmitOf + } + if (typeof event === "object" && event !== null && protocol.trustedEmissions.has(event)) { + return event as Machine.EmitOf + } const eventName = getEventName(event) return decodeBoundarySync>( machine, - getProtocolSchemas(machine).emit, + protocol.emit, event, - eventName === undefined ? { boundary: "emit" } : { boundary: "emit", event: eventName } + eventName === undefined ? { boundary: "emission" } : { boundary: "emission", event: eventName } ) } diff --git a/src/internal/machine/runtime.ts b/src/internal/machine/runtime.ts index 5342c94..a7c3e03 100644 --- a/src/internal/machine/runtime.ts +++ b/src/internal/machine/runtime.ts @@ -18,6 +18,7 @@ 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 { ActorRef } from "../../Machine.js" import { ChildAlreadyExistsError, StoppedError } from "./errors.js" type ChildDescriptor = { @@ -284,12 +285,13 @@ export type RuntimeOutcome = readonly snapshot: Extract, { readonly status: "stopped" }> } -export interface MachineRef { +export interface MachineRef { readonly id: string readonly sessionId: string readonly state: Effect.Effect readonly snapshot: Effect.Effect> readonly changes: Stream.Stream> + readonly emissions: Stream.Stream readonly join: Effect.Effect readonly stop: Effect.Effect readonly send: (event: Event) => Effect.Effect @@ -307,12 +309,19 @@ interface ProcessAddress { readonly send: (event: Event) => Effect.Effect } +const isProcessAddress = (value: unknown): value is ActorRef => + typeof value === "object" && value !== null && "send" in value && typeof value.send === "function" + export interface ProcessScope { readonly self: ProcessAddress readonly parent: ProcessAddress | undefined readonly spawn: ProcessSpawn readonly sendParent: (event: unknown) => Effect.Effect - readonly sendTo: (child: ChildSelector, event: unknown) => Effect.Effect + readonly emit: (event: unknown) => Effect.Effect + readonly sendTo: { + (target: ActorRef, event: TargetEvent): Effect.Effect + (child: ChildSelector, event: unknown): Effect.Effect + } readonly stopChild: (child: ChildSelector) => Effect.Effect /** @internal */ readonly failCause: (cause: Cause.Cause) => Effect.Effect @@ -708,6 +717,10 @@ class OwnedChildRuntimeImpl implements OwnedChildRuntime { ownerPath: options.path, ownerActive: true }) + const parent: ProcessAddress = { + ...this.self, + send: (event) => options.sendParent(isCurrent, event) + } const startOptions: StartInternalOptions = { detached: true, id: options.id, @@ -722,7 +735,7 @@ class OwnedChildRuntimeImpl implements OwnedChildRuntime { }, onStopSync: () => unregisterChild(this.registry, options.id, token), skipStoppedOutcome: true, - parent: this.self, + parent, runtime: this.runtime } const execution = logic.execution @@ -785,6 +798,9 @@ class OwnedChildRuntimeImpl implements OwnedChildRuntime { const noChildChanges = Stream.succeed(Option.none()).pipe(Stream.concat(Stream.never)) const noParentSend = (_event: unknown): Effect.Effect => Effect.void +const EmissionsClosed: unique symbol = Symbol("effect/Machine/EmissionsClosed") + +type LazyEmissions = PubSub.PubSub | typeof EmissionsClosed | undefined const childlessRuntime: ChildRuntime = { close: () => Effect.void, @@ -1091,6 +1107,38 @@ const startGenericInternal: < const 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 termination = yield* Deferred.make() const done = yield* Deferred.make() const awaitCompletion = Deferred.await(done).pipe(Effect.exit, Effect.asVoid) @@ -1161,7 +1209,7 @@ const startGenericInternal: < } const cleanupStartupFailure = (exit: Exit.Exit): Effect.Effect => Exit.isFailure(exit) - ? closeChildren(exit) + ? closeChildren(exit).pipe(Effect.andThen(closeEmissions())) : Effect.void const cleanup = onStopSync === undefined ? onStop ?? Effect.void : Effect.sync(onStopSync) const sendParent = overrideSendParent ?? (parent === undefined ? noParentSend : parent.send) @@ -1171,7 +1219,11 @@ const startGenericInternal: < parent, spawn, sendParent, - sendTo, + emit, + sendTo: ((target: unknown, event: unknown) => + isProcessAddress(target) ? target.send(event) : sendTo(target as ChildSelector, event)) as ProcessScope< + Event + >["sendTo"], stopChild, failCause: (cause) => Deferred.succeed(termination, { @@ -1209,7 +1261,9 @@ const startGenericInternal: < const runtimeSnapshot = snapshot.snapshot return runtimeSnapshot.status !== "active" ? publish - : publish.pipe(Effect.tap(() => notifyActiveSnapshot(onSnapshot, runtimeSnapshot))) + : publish.pipe(Effect.tap(() => + notifyActiveSnapshot(onSnapshot, runtimeSnapshot) + )) } const completeChanges = ( @@ -1348,6 +1402,7 @@ const startGenericInternal: < Effect.andThen(Queue.shutdown(queue)), Effect.andThen(closeChildren(exit)), Effect.andThen(setAndPublishSnapshot(snapshot)), + Effect.andThen(closeEmissions()), Effect.andThen(Effect.sync(() => { if (Exit.isFailure(exit)) { failAcknowledgedMessage(inFlightMessage, exit.cause) @@ -1504,6 +1559,11 @@ 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, join: Deferred.await(done), stop, send: self.send, @@ -1657,6 +1717,7 @@ class CompiledProcess implements MachineRef { private interruptRequested = false private offerRevision = 0 private inFlightMessage: ProcessMessage | undefined + private emissionsPubSub: LazyEmissions constructor( private readonly logic: ProcessLogic, @@ -1690,7 +1751,11 @@ class CompiledProcess implements MachineRef { parent, spawn: this.childRuntime.spawn, sendParent, - sendTo: this.childRuntime.sendTo, + emit: (event) => this.emitEvent(event), + sendTo: ((target: unknown, event: unknown) => + isProcessAddress(target) + ? target.send(event) + : this.childRuntime.sendTo(target as ChildSelector, event)) as ProcessScope["sendTo"], stopChild: this.childRuntime.stop, failCause: (cause: Cause.Cause) => this.failCause(cause) } @@ -1743,7 +1808,11 @@ class CompiledProcess implements MachineRef { parent, spawn: self.childRuntime.spawn, sendParent, - sendTo: self.childRuntime.sendTo, + emit: (event) => self.emitEvent(event), + sendTo: ((target: unknown, event: unknown) => + isProcessAddress(target) + ? target.send(event) + : self.childRuntime.sendTo(target as ChildSelector, event)) as ProcessScope["sendTo"], stopChild: self.childRuntime.stop, failCause: (cause: Cause.Cause) => self.failCause(cause) } @@ -1838,6 +1907,10 @@ class CompiledProcess implements MachineRef { return this.changesStream() } + get emissions(): Stream.Stream { + return this.emissionsStream() as Stream.Stream + } + get join(): Effect.Effect { return Effect.suspend(() => { if (this.lifecycle === "Completed") { @@ -2145,6 +2218,7 @@ class CompiledProcess implements MachineRef { }).pipe( Effect.andThen(this.childRuntime.close(exit)), Effect.andThen(this.setAndPublishSnapshot(snapshot)), + Effect.andThen(this.closeEmissions()), Effect.andThen(Effect.sync(() => { if (requested._tag === "Failure") { failAcknowledgedMessage(this.inFlightMessage, requested.cause) @@ -2193,6 +2267,55 @@ class CompiledProcess implements MachineRef { : PubSub.publish(snapshot.changes, Exit.succeed(undefined)).pipe(Effect.asVoid) } + private emitEvent(event: unknown): Effect.Effect { + return Effect.suspend(() => + this.emissionsPubSub === undefined || this.emissionsPubSub === EmissionsClosed + ? Effect.void + : PubSub.publish(this.emissionsPubSub, event).pipe(Effect.asVoid) + ) + } + + shutdownEmissions(): Effect.Effect { + return this.closeEmissions() + } + + private closeEmissions(): Effect.Effect { + const observed = this.emissionsPubSub + this.emissionsPubSub = EmissionsClosed + return observed === undefined || observed === EmissionsClosed ? Effect.void : PubSub.shutdown(observed) + } + + private getOrCreateEmissions(): Effect.Effect | undefined> { + return Effect.suspend(() => { + const observed = this.emissionsPubSub + 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 = this.emissionsPubSub + if (latest === EmissionsClosed) return [undefined, true] as const + if (latest !== undefined) return [latest, true] as const + this.emissionsPubSub = candidate + return [candidate, false] as const + }).pipe( + Effect.flatMap(([selected, discard]) => + discard ? PubSub.shutdown(candidate).pipe(Effect.as(selected)) : Effect.succeed(selected) + ) + ) + ) + ) + }) + } + + private emissionsStream(): Stream.Stream { + return Stream.unwrap( + this.getOrCreateEmissions().pipe( + Effect.map((emissions) => emissions === undefined ? Stream.empty : Stream.fromPubSub(emissions)) + ) + ) + } + private setAndPublishSnapshot(snapshot: RuntimeSnapshot): Effect.Effect { return Effect.suspend(() => { this.flushPendingChanges() @@ -2407,10 +2530,13 @@ const startCompactCompiledInternal: typeof startGenericInternal = Effect.fnUntra // A compiled initializer is synchronous by construction. Only startup // callbacks that themselves return Effects need the generic initialization // program; the compiled drain is still provided the complete service context. - return yield* execution.initialSync !== undefined && + const initialize = execution.initialSync !== undefined && options.onReady === undefined && options.onSnapshot === undefined ? process.initializeCompiledSync() : process.initialize() + return yield* initialize.pipe( + Effect.onExit((exit) => Exit.isFailure(exit) ? process.shutdownEmissions() : Effect.void) + ) }) as typeof startGenericInternal const startLogicInternal: typeof startGenericInternal = (( diff --git a/src/unstable/reactivity/AtomMachine.ts b/src/unstable/reactivity/AtomMachine.ts index c07e298..b96007b 100644 --- a/src/unstable/reactivity/AtomMachine.ts +++ b/src/unstable/reactivity/AtomMachine.ts @@ -7,6 +7,7 @@ import type * as Option from "effect/Option" import type * as Schema from "effect/Schema" import type * as Scope from "effect/Scope" +import type * as Stream from "effect/Stream" import type { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity" import * as internal from "../../internal/machine/atom.js" import type { ChildNotActiveError, NotReadyError } from "../../internal/machine/atom.js" @@ -89,14 +90,14 @@ type MachineStartError = * @category models * @since 0.4.0 */ -export interface MachineAtom { +export interface MachineAtom { /** * Atom containing the running machine handle once startup succeeds. * * @since 0.4.0 */ readonly ref: Atom.Atom< - AsyncResult.AsyncResult, StartError> + AsyncResult.AsyncResult, StartError> > /** @@ -163,6 +164,32 @@ type RefState = Ref extends Machine.MachineRef type RefError = Ref extends Machine.MachineRef ? Error : never type RefOutput = Ref extends Machine.MachineRef ? Output : never +type RefEmitted = 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. + * + * @category getters + * @since 0.10.0 + */ +export const emissions: ( + self: MachineAtom +) => Stream.Stream = internal.emissions + +/** + * Observes emissions from each active instance selected by a child bridge. + * + * @category getters + * @since 0.10.0 + */ +export const childEmissions: ( + self: ChildMachineAtom +) => Stream.Stream< + RefEmitted>, + StartError, + AtomRegistry.AtomRegistry +> = internal.childEmissions /** * Reactive access to one invoked child machine selected by its descriptor. @@ -546,7 +573,8 @@ type MachineAtomOf = MachineAtom< Machine.Machine.InitialServices, Machine.Machine.Services, RuntimeError - > + >, + Machine.Machine.EmittedEvent > type ResumedMachineAtomOf = MachineAtom< @@ -554,7 +582,8 @@ type ResumedMachineAtomOf = Machine Machine.Machine.EventInput>, MachineRuntimeError, Machine.Machine.Services>, Machine.Machine.Output, - Machine.MachineSchemaDecodeError | RuntimeError + Machine.MachineSchemaDecodeError | RuntimeError, + Machine.Machine.EmittedEvent > /** @@ -631,7 +660,8 @@ export const make: { FinalStates extends Machine.Machine.StateIdentifier = never, Output = never, OutputStates extends Machine.Machine.StateIdentifier = never, - InputEvents extends ReadonlyArray = Events + InputEvents extends ReadonlyArray = Events, + ParentEvents extends ReadonlyArray = readonly [] >( machine: & Machine.Machine< @@ -647,7 +677,8 @@ export const make: { Output, Emits, OutputStates, - InputEvents + InputEvents, + ParentEvents > & EnsureNoExternalRequirements< MachineRequirements< @@ -664,7 +695,8 @@ export const make: { Machine.Machine.EventInputOf, MachineRuntimeError, Output, - MachineStartError + MachineStartError, + Machine.Machine.EmittedEventOf > } = internal.make diff --git a/test/internal/machine/strategyDifferential.test.ts b/test/internal/machine/strategyDifferential.test.ts index 2545353..8e54d58 100644 --- a/test/internal/machine/strategyDifferential.test.ts +++ b/test/internal/machine/strategyDifferential.test.ts @@ -319,6 +319,56 @@ describe("machine planner and runtime strategies", () => { } }) as Effect.Effect) + it.effect("publishes and validates emitted events in generic and compiled managed runtimes", () => + Effect.gen(function*() { + class Idle extends Schema.TaggedClass("StrategyEmissionIdle")("Idle", {}) {} + class Publish extends Schema.TaggedClass("StrategyEmissionPublish")("Publish", {}) {} + class Published extends Schema.TaggedClass("StrategyEmissionPublished")("Published", { + value: Schema.Number + }) {} + const states = Machine.defineStates({ Idle }) + const Events = Machine.events(Publish) + const Emissions = Machine.emittedEvents(Published) + let value: unknown = 1 + const machine = Machine.make({ + states: states.states, + events: Events, + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + on: { + Publish: ({ parent, self }, enqueue) => { + assert.strictEqual(parent, undefined) + assert.ok(self.sessionId.startsWith("machine:")) + enqueue.emit(Emissions.Published({ value } as never)) + } + } + } + }) + + for (const strategy of ["generic", "compiled"] as const) { + value = 1 + const ref = yield* openWithRuntimeStrategy(machine, strategy) + const observed = yield* ref.emissions.pipe( + Stream.take(1), + Stream.runCollect, + Effect.forkChild({ startImmediately: true }) + ) + yield* ref.send(Events.Publish()) + assert.deepStrictEqual(Array.from(yield* Fiber.join(observed)), [new Published({ value: 1 })]) + yield* ref.stop + + value = "invalid" + const invalid = yield* openWithRuntimeStrategy(machine, strategy) + yield* invalid.send(Events.Publish()) + const error = yield* Effect.flip(invalid.join) + assert.instanceOf(error, Machine.MachineSchemaDecodeError) + assert.strictEqual(error.boundary, "emission") + assert.strictEqual(error.event, "Published") + } + }) as Effect.Effect) + it.effect("matches acknowledged probe delivery in generic and compiled managed runtimes", () => Effect.gen(function*() { const machine = makeFlatMachine() @@ -495,15 +545,17 @@ describe("machine planner and runtime strategies", () => { const current = generation return Machine.logic({ initial: "active", - run: ({ sendParent, setState }) => - (current === 1 ? Deferred.succeed(firstStarted, undefined) : Effect.void).pipe( - Effect.andThen(Effect.never), - Effect.onInterrupt(() => - setState("stale").pipe( - Effect.andThen(sendParent(new Stale({}))) + run: ({ parent, sendTo, setState }) => + parent === undefined ? + Effect.die("worker expected an owning actor") : + (current === 1 ? Deferred.succeed(firstStarted, undefined) : Effect.void).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => + setState("stale").pipe( + Effect.andThen(sendTo(parent, new Stale({}))) + ) ) ) - ) }) }, onFailure: () => undefined, diff --git a/test/machine/ActorEvents.test.ts b/test/machine/ActorEvents.test.ts new file mode 100644 index 0000000..4c6fb94 --- /dev/null +++ b/test/machine/ActorEvents.test.ts @@ -0,0 +1,181 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Fiber, Option, Schema, Stream } from "effect" +import { Machine } from "../../src/index.js" + +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("publishes root emissions as a hot, non-replayed Effect Stream", () => + Effect.gen(function*() { + class Idle extends Schema.TaggedClass("EmissionRootIdle")("Idle", {}) {} + class Publish extends Schema.TaggedClass("EmissionRootPublish")("Publish", { + value: Schema.Number + }) {} + class Published extends Schema.TaggedClass("EmissionRootPublished")("Published", { + value: Schema.Number + }) {} + + const states = Machine.defineStates({ Idle }) + const Events = Machine.events(Publish) + const Emissions = Machine.emittedEvents(Published) + const machine = Machine.make({ + states: states.states, + events: Events, + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + on: { + Publish: ({ event }, enqueue) => enqueue.emit(Emissions.Published({ value: event.value })) + } + } + }) + + const ref = yield* Machine.start(machine) + const early = yield* ref.emissions.pipe( + Stream.take(2), + Stream.runCollect, + Effect.map(Array.from), + Effect.forkChild({ startImmediately: true }) + ) + const first = yield* collectNext(ref.emissions) + yield* ref.send(Events.Publish({ value: 1 })) + yield* Fiber.join(first) + + const late = yield* collectNext(ref.emissions) + yield* ref.send(Events.Publish({ value: 2 })) + + const earlyValues = yield* Fiber.join(early) + const lateValues = yield* Fiber.join(late) + assert.deepStrictEqual((earlyValues as Array).map(({ value }) => value), [1, 2]) + assert.deepStrictEqual((lateValues as Array).map(({ value }) => value), [2]) + + yield* ref.stop + assert.deepStrictEqual(Array.from(yield* Stream.runCollect(ref.emissions)), []) + })) + + it.effect("fails the actor with a typed machine error when an emission cannot be decoded", () => + Effect.gen(function*() { + class Idle extends Schema.TaggedClass("InvalidEmissionIdle")("Idle", {}) {} + class Publish extends Schema.TaggedClass("InvalidEmissionPublish")("Publish", {}) {} + class Published extends Schema.TaggedClass("InvalidEmissionPublished")("Published", { + value: Schema.Number + }) {} + + const states = Machine.defineStates({ Idle }) + const Events = Machine.events(Publish) + const Emissions = Machine.emittedEvents(Published) + const machine = Machine.make({ + states: states.states, + events: Events, + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + on: { + Publish: (_, enqueue) => enqueue.emit(Emissions.Published({ value: "invalid" } as never)) + } + } + }) + + const ref = yield* Machine.start(machine) + const observed = yield* ref.emissions.pipe( + Stream.runCollect, + Effect.map(Array.from), + Effect.forkChild({ startImmediately: true }) + ) + yield* ref.send(Events.Publish()) + const error = yield* Effect.flip(ref.join) + + assert.instanceOf(error, Machine.MachineSchemaDecodeError) + assert.strictEqual(error.boundary, "emission") + assert.strictEqual(error.event, "Published") + assert.deepStrictEqual(yield* Fiber.join(observed), []) + })) + + it.effect("types a child parent reference from parentEvents and keeps emissions external", () => + Effect.gen(function*() { + class Waiting extends Schema.TaggedClass("ParentEventsWaiting")("Waiting", {}) {} + class Reported extends Schema.TaggedClass("ParentEventsReported")("Reported", {}) {} + class Trigger extends Schema.TaggedClass("ParentEventsTrigger")("Trigger", {}) {} + class ChildReported extends Schema.TaggedClass("ParentEventsChildReported")("ChildReported", { + value: Schema.Number + }) {} + class Notice extends Schema.TaggedClass("ParentEventsNotice")("Notice", { + value: Schema.Number + }) {} + class Awaiting extends Schema.TaggedClass("ParentEventsAwaiting")("Awaiting", {}) {} + class Finished extends Schema.TaggedClass("ParentEventsFinished")("Finished", { + source: Schema.String + }) {} + + const ParentEvents = Machine.events(ChildReported) + const ChildEvents = Machine.events(Trigger) + const ChildEmissions = Machine.emittedEvents(Notice) + const childStates = Machine.defineStates({ Waiting, Reported }) + let rootHadParent = true + const childMachine = Machine.make({ + states: childStates.states, + events: ChildEvents, + parentEvents: ParentEvents, + emittedEvents: ChildEmissions, + initial: () => childStates.initial.Waiting(new Waiting({})) + }).handle({ + Waiting: { + on: { + Trigger: ({ parent, target }, enqueue) => { + rootHadParent = parent !== undefined + enqueue.emit(ChildEmissions.Notice({ value: 1 })) + if (parent !== undefined) { + enqueue.sendTo(parent, ParentEvents.ChildReported({ value: 1 })) + } + return target.full.Reported(new Reported({})) + } + } + }, + Reported: {} + }) + + const root = yield* Machine.start(childMachine) + const rootChanged = yield* root.changes.pipe( + Stream.filter((snapshot) => snapshot.state.path === "Reported"), + Stream.take(1), + Stream.runDrain, + Effect.forkChild({ startImmediately: true }) + ) + yield* root.send(ChildEvents.Trigger()) + yield* Fiber.join(rootChanged) + assert.isFalse(rootHadParent) + yield* root.stop + + const Child = Machine.child("reporter", childMachine) + const parentStates = Machine.defineStates({ + Awaiting, + Finished: { schema: Finished, type: "final", output: Schema.String } + }) + const parentMachine = Machine.make({ + states: parentStates.states, + events: Machine.events(ParentEvents, Notice), + initial: () => parentStates.initial.Awaiting(new Awaiting({})) + }).handle({ + Awaiting: { + invoke: Machine.invoke({ child: Child }), + on: { + ChildReported: ({ target }) => target.full.Finished(new Finished({ source: "parent event" })), + Notice: ({ target }) => target.full.Finished(new Finished({ source: "emission" })) + } + }, + Finished: { output: ({ state }) => state.source } + }) + + const parent = yield* Machine.start(parentMachine) + const child = Option.getOrThrow(yield* parent.child(Child)) + const notice = yield* collectNext(child.emissions) + yield* child.send(ChildEvents.Trigger()) + + const notices = yield* Fiber.join(notice) + assert.deepStrictEqual((notices as Array).map(({ value }) => value), [1]) + assert.strictEqual(yield* parent.join, "parent event") + })) +}) diff --git a/test/machine/Choice.test.ts b/test/machine/Choice.test.ts index 3523be9..733a98d 100644 --- a/test/machine/Choice.test.ts +++ b/test/machine/Choice.test.ts @@ -30,8 +30,8 @@ const machine = Machine.make({ Routing: { choice: { targets: ["Flow.Approved", "Flow.Rejected"], - transition: ({ parent, target }) => { - return parent.score >= 70 + transition: ({ containingState, target }) => { + return containingState.score >= 70 ? target.local.Approved(new Approved({})) : target.local.Rejected(new Rejected({})) } diff --git a/test/machine/Machine.test.ts b/test/machine/Machine.test.ts index bc6f227..779b776 100644 --- a/test/machine/Machine.test.ts +++ b/test/machine/Machine.test.ts @@ -1900,9 +1900,9 @@ describe("Machine", () => { states: { entering: { on: { - Authorize: ({ event, parent, parents, target }) => { - assert.deepStrictEqual(parent, payment) - assert.deepStrictEqual(parents, { payment }) + Authorize: ({ event, containingState, ancestors, target }) => { + assert.deepStrictEqual(containingState, payment) + assert.deepStrictEqual(ancestors, { payment }) return target.local.authorized(new AuthorizedPayment({ code: event.code })) } } @@ -1961,8 +1961,8 @@ describe("Machine", () => { } }, authorized: { - output: ({ parents, state }) => { - assert.deepStrictEqual(parents, { payment }) + output: ({ ancestors, state }) => { + assert.deepStrictEqual(ancestors, { payment }) return state.code } } @@ -4513,7 +4513,7 @@ describe("Machine", () => { assert.strictEqual(yield* actor.join, "loaded") })) - it.effect("routes sendParent from an active invoked machine", () => + it.effect("routes sendTo(parent) from an active invoked machine", () => Effect.gen(function*() { const childStarted = yield* Deferred.make() const machine = Machine.make({ @@ -4527,11 +4527,13 @@ describe("Machine", () => { address: Machine.childAddress("request-parent"), logic: Machine.logic({ initial: undefined, - run: ({ sendParent }) => - Deferred.succeed(childStarted, void 0).pipe( - Effect.andThen(sendParent(new RequestSucceeded({ value: "child" }))), - Effect.andThen(Effect.never) - ) + run: ({ parent, sendTo }) => + parent === undefined ? + Effect.die("child expected an owning actor") : + Deferred.succeed(childStarted, void 0).pipe( + Effect.andThen(sendTo(parent, new RequestSucceeded({ value: "child" }))), + Effect.andThen(Effect.never) + ) }), onFailure: () => undefined }), @@ -4550,7 +4552,7 @@ describe("Machine", () => { assert.strictEqual(yield* actor.join, "child") })) - it.effect("drops sendParent from a stale invoked machine finalizer", () => + it.effect("drops sendTo(parent) from a stale invoked machine finalizer", () => Effect.gen(function*() { const childStarted = yield* Deferred.make() const machine = Machine.make({ @@ -4569,11 +4571,13 @@ describe("Machine", () => { address: Machine.childAddress("stale-request"), logic: Machine.logic({ initial: undefined, - run: ({ sendParent }) => - Deferred.succeed(childStarted, void 0).pipe( - Effect.andThen(Effect.never), - Effect.onInterrupt(() => sendParent(new RequestSucceeded({ value: "stale" }))) - ) + run: ({ parent, sendTo }) => + parent === undefined ? + Effect.die("child expected an owning actor") : + Deferred.succeed(childStarted, void 0).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => sendTo(parent, new RequestSucceeded({ value: "stale" }))) + ) }), onFailure: () => undefined }), @@ -4991,9 +4995,9 @@ describe("Machine", () => { }), states: { entering: { - entry: ({ parents, state }) => { + entry: ({ ancestors, state }) => { assert.deepStrictEqual(state, entering) - assert.deepStrictEqual(parents, { payment }) + assert.deepStrictEqual(ancestors, { payment }) }, invoke: Machine.invoke({ id: "request", diff --git a/test/machine/RuntimeDifferential.test.ts b/test/machine/RuntimeDifferential.test.ts index 9bf15c6..b97b34e 100644 --- a/test/machine/RuntimeDifferential.test.ts +++ b/test/machine/RuntimeDifferential.test.ts @@ -170,22 +170,33 @@ describe("pure planning and managed runtime differential", () => { Advance: ({ state, target }) => target.branch.Running.Right(new Right({ value: state.value + 10 })), Bump: ({ state, target }) => target.branch.Running.Right(new Right({ value: state.value + 100 })), Inspect: (context) => { - const { state, parent, parents, snapshot } = context + const { state, containingState, ancestors, snapshot } = context if (snapshot.path !== "Running") throw new Error("expected Running snapshot") - const expectedKeys = ["state", "parent", "parents", "event", "snapshot", "target"] + const expectedKeys = [ + "self", + "parent", + "state", + "containingState", + "ancestors", + "event", + "snapshot", + "target" + ] const spread = { ...context } assert.deepStrictEqual(Object.keys(context), expectedKeys) assert.deepStrictEqual(Object.keys(spread), expectedKeys) + assert.strictEqual(spread.self, context.self) + assert.strictEqual(spread.parent, context.parent) assert.strictEqual(spread.state, state) - assert.strictEqual(spread.parent, parent) - assert.strictEqual(spread.parents, parents) + assert.strictEqual(spread.containingState, containingState) + assert.strictEqual(spread.ancestors, ancestors) assert.strictEqual(spread.event, context.event) assert.strictEqual(spread.snapshot, snapshot) assert.strictEqual(spread.target, context.target) observations.push({ state: state.value, - parent: parent._tag, - parents: parents.Running._tag, + parent: containingState._tag, + parents: ancestors.Running._tag, left: snapshot.states.Left.value.value, right: snapshot.states.Right.value.value }) @@ -425,7 +436,7 @@ describe("pure planning and managed runtime differential", () => { states: states.states, events: Machine.events(Begin), internalEvents: Machine.internalEvents(RaisedOne, RaisedTwo), - emits: [Notice], + emittedEvents: Machine.emittedEvents(Notice), initial: () => states.initial.Idle(new Idle({})) }).handle({ Idle: { diff --git a/test/machine/StructuralStates.test.ts b/test/machine/StructuralStates.test.ts index 0e757a6..eb841b1 100644 --- a/test/machine/StructuralStates.test.ts +++ b/test/machine/StructuralStates.test.ts @@ -108,9 +108,9 @@ const makeMachine = () => states: { Paused: { on: { - Play: ({ parent, state, target }) => { + Play: ({ containingState, state, target }) => { assert.strictEqual(state, undefined) - return target.local.Playing.from({ position: Math.min(0, parent.duration) }) + return target.local.Playing.from({ position: Math.min(0, containingState.duration) }) } } }, diff --git a/test/machine/support/activityLifecycleModel.ts b/test/machine/support/activityLifecycleModel.ts index 4eabcc0..754ab69 100644 --- a/test/machine/support/activityLifecycleModel.ts +++ b/test/machine/support/activityLifecycleModel.ts @@ -82,8 +82,8 @@ export const makeActivityProbe: Effect.Effect = Effect.gen(functi logic: (owner, behavior) => Machine.logic({ initial: () => initial(owner), - run: ({ sendParent, state }) => - state.pipe( + run: ({ parent, sendTo, state }) => + parent === undefined ? Effect.die("activity expected an owning actor") : state.pipe( Effect.flatMap(({ epoch, release }) => { switch (behavior._tag) { case "Blocked": @@ -95,7 +95,7 @@ export const makeActivityProbe: Effect.Effect = Effect.gen(functi case "StaleOnCancel": return Effect.never.pipe( Effect.onInterrupt(() => - sendParent(behavior.event(epoch)).pipe( + sendTo(parent, behavior.event(epoch)).pipe( Effect.catchTag("StoppedError", () => Effect.void) ) ), diff --git a/test/machine/support/runtimeDifferential.ts b/test/machine/support/runtimeDifferential.ts index ceb10f5..72f994d 100644 --- a/test/machine/support/runtimeDifferential.ts +++ b/test/machine/support/runtimeDifferential.ts @@ -1,5 +1,5 @@ import { assert } from "@effect/vitest" -import { Effect, Fiber, Stream } from "effect" +import { Cause, Effect, Fiber, Stream } from "effect" import { isDeepStrictEqual } from "node:util" import { Machine } from "../../../src/index.js" import { MachineTest } from "../../../src/testing/index.js" @@ -28,7 +28,11 @@ const assertRuntimeSnapshot = Effect.fn(function*( expected: DifferentialBoundary, label: string ) { - assert.notStrictEqual(actual.status, "error", `${label} unexpectedly failed`) + assert.notStrictEqual( + actual.status, + "error", + `${label} unexpectedly failed${actual.status === "error" ? `: ${Cause.pretty(actual.cause)}` : ""}` + ) assert.notStrictEqual(actual.status, "stopped", `${label} unexpectedly stopped`) if (actual.status === "error" || actual.status === "stopped") return assert.strictEqual(actual.status, expected.done ? "done" : "active", `${label} status`) diff --git a/test/testing/Probe.test.ts b/test/testing/Probe.test.ts index ca474c8..c14f369 100644 --- a/test/testing/Probe.test.ts +++ b/test/testing/Probe.test.ts @@ -52,6 +52,7 @@ describe("MachineTest probe", () => { state: ref.state, snapshot: ref.snapshot, changes: ref.changes, + emissions: ref.emissions, join: ref.join, stop: ref.stop, send: ref.send, diff --git a/test/testing/Runtime.test.ts b/test/testing/Runtime.test.ts index b674d97..5dffc6b 100644 --- a/test/testing/Runtime.test.ts +++ b/test/testing/Runtime.test.ts @@ -427,6 +427,7 @@ describe("MachineTest runtime commands", () => { state: Effect.succeed(2), snapshot: Effect.succeed(second), changes: Stream.make(initial, first, second), + emissions: Stream.empty, join: Effect.never, stop: Effect.void, send: () => Effect.void, diff --git a/test/unstable/cluster/ClusterMachine.test.ts b/test/unstable/cluster/ClusterMachine.test.ts index fa3c3fc..5e61d2f 100644 --- a/test/unstable/cluster/ClusterMachine.test.ts +++ b/test/unstable/cluster/ClusterMachine.test.ts @@ -59,7 +59,7 @@ const makeCounter = (state: { id: "Counter", states: CounterStates.states, events: Machine.events(Increment, Fail, Finish, RaiseFromAction, SpawnFromAction), - emits: [Changed], + emittedEvents: Machine.emittedEvents(Changed), initial: () => CounterStates.initial.Count(new Count({ value: 0 })) }).handle({ Count: { diff --git a/typetest/machine/ActorEvents.tst.ts b/typetest/machine/ActorEvents.tst.ts new file mode 100644 index 0000000..d0a4ebf --- /dev/null +++ b/typetest/machine/ActorEvents.tst.ts @@ -0,0 +1,121 @@ +import { Effect, Schema, Stream } from "effect" +import { AtomRegistry } from "effect/unstable/reactivity" +import { describe, expect, it } from "tstyche" +import { Machine } from "../../src/index.js" +import { AtomMachine } from "../../src/unstable/reactivity/index.js" + +describe("machine actor event channels", () => { + class Idle extends Schema.TaggedClass("ActorEventsIdle")("Idle", {}) {} + class Ping extends Schema.TaggedClass("ActorEventsPing")("Ping", {}) {} + class Local extends Schema.TaggedClass("ActorEventsLocal")("Local", {}) {} + class ParentNotice extends Schema.TaggedClass("ActorEventsParentNotice")("ParentNotice", { + value: Schema.Number + }) {} + class OtherParentEvent extends Schema.TaggedClass("ActorEventsOtherParent")( + "OtherParentEvent", + {} + ) {} + class Published extends Schema.TaggedClass("ActorEventsPublished")("Published", {}) {} + class ValuedPublished extends Schema.TaggedClass("ActorEventsValuedPublished")( + "ValuedPublished", + { value: Schema.Number } + ) {} + + const ParentEvents = Machine.events(ParentNotice) + const Events = Machine.events(Ping) + const InternalEvents = Machine.internalEvents(Local) + const Emissions = Machine.emittedEvents(Published, ValuedPublished) + const states = Machine.defineStates({ Idle }) + const childMachine = Machine.make({ + states: states.states, + events: Events, + internalEvents: InternalEvents, + parentEvents: ParentEvents, + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + on: { + Ping: () => undefined + } + } + }) + const Child = Machine.child("child", childMachine) + + it("types self, parent, raised events, and emissions as separate channels", () => { + Machine.make({ + states: states.states, + events: Events, + internalEvents: InternalEvents, + parentEvents: ParentEvents, + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + on: { + Ping: ({ parent, self }, enqueue) => { + expect(self.send).type.toBeCallableWith(Events.Ping()) + expect(self.send).type.not.toBeCallableWith(InternalEvents.Local()) + expect(enqueue.sendTo).type.toBeCallableWith(self, Events.Ping()) + expect(enqueue.sendTo).type.not.toBeCallableWith(self, ParentEvents.ParentNotice({ value: 1 })) + + if (parent !== undefined) { + expect(enqueue.sendTo).type.toBeCallableWith(parent, ParentEvents.ParentNotice({ value: 1 })) + expect(enqueue.sendTo).type.not.toBeCallableWith(parent, Events.Ping()) + } + + expect(enqueue.raise).type.toBeCallableWith(InternalEvents.Local()) + expect(enqueue.emit).type.toBeCallableWith(Emissions.Published()) + expect(enqueue.emit).type.toBeCallableWith(Emissions.ValuedPublished({ value: 1 })) + expect(enqueue.emit).type.not.toBeCallableWith(Events.Ping()) + } + } + } + }) + }) + + it("composes builder protocols and checks required parent inputs", () => { + const compatible = Machine.make({ + states: states.states, + events: Machine.events(Ping, ParentEvents), + initial: () => states.initial.Idle(new Idle({})) + }) + compatible.handle({ + Idle: { + invoke: { + child: Child, + onDone: () => undefined, + onFailure: () => undefined, + onSnapshot: () => undefined + } + } + }) + + const incompatible = Machine.make({ + states: states.states, + events: Machine.events(Ping, OtherParentEvent), + initial: () => states.initial.Idle(new Idle({})) + }) + expect(incompatible.handle).type.not.toBeCallableWith({ + Idle: { + invoke: { + child: Child, + onDone: () => undefined, + onFailure: () => undefined, + onSnapshot: () => undefined + } + } + }) + }) + + it("infers emitted streams through MachineRef and AtomMachine", () => { + const started = Machine.start(childMachine) + type Ref = Effect.Success + expect().type.toBe>() + + const atom = AtomMachine.make(childMachine) + const atomEmissions = AtomMachine.emissions(atom) + expect>().type.toBe() + expect>().type.toBe() + }) +}) diff --git a/typetest/machine/Choice.tst.ts b/typetest/machine/Choice.tst.ts index 3c50d6e..3d52962 100644 --- a/typetest/machine/Choice.tst.ts +++ b/typetest/machine/Choice.tst.ts @@ -45,8 +45,8 @@ describe("Machine choice pseudo-states", () => { targets: ["Flow.Approved", "Flow.Rejected"], transition: (context) => { expect(context).type.not.toHaveProperty("state") - expect(context.parent).type.toBe() - expect(context.parents.Flow).type.toBe() + expect(context.containingState).type.toBe() + expect(context.ancestors.Flow).type.toBe() expect(context.event).type.toBe>() return context.target.local.Approved(new Approved({})) } diff --git a/typetest/machine/History.tst.ts b/typetest/machine/History.tst.ts index 8737ddb..8093e98 100644 --- a/typetest/machine/History.tst.ts +++ b/typetest/machine/History.tst.ts @@ -204,8 +204,8 @@ describe("Machine history states", () => { checkout: { history: { recent: { - default: ({ parent, target }) => { - expect(parent).type.toBe<"checkout">() + default: ({ owner, target }) => { + expect(owner).type.toBe<"checkout">() expect(target).type.toBe< Machine.Machine.HistoryDefaultTargetBuilder >() @@ -225,10 +225,10 @@ describe("Machine history states", () => { }, states: { payment: { - initial: ({ state, parent, parents }) => { + initial: ({ state, containingState, ancestors }) => { expect(state).type.toBe() - expect(parent).type.toBe() - expect(parents).type.toBe<{ readonly checkout: Checkout }>() + expect(containingState).type.toBe() + expect(ancestors).type.toBe<{ readonly checkout: Checkout }>() return new CardEntry({ cardNumber: `attempt-${state.attempt}` }) } } @@ -295,8 +295,8 @@ describe("Machine history states", () => { Workspace: { history: { resume: { - default: ({ parent, target }) => { - expect(parent).type.toBe<"App.Workspace">() + default: ({ owner, target }) => { + expect(owner).type.toBe<"App.Workspace">() expect(target).type.toBe< Machine.Machine.HistoryDefaultTargetBuilder >() diff --git a/typetest/machine/Machine.tst.ts b/typetest/machine/Machine.tst.ts index 701e1c2..c7c4b24 100644 --- a/typetest/machine/Machine.tst.ts +++ b/typetest/machine/Machine.tst.ts @@ -229,14 +229,14 @@ describe("Machine", () => { it("machine contexts expose type-safe parent state values", () => { const nested = null as unknown as SignedOutContext - expect(nested.parent).type.toBe() - expect(nested.parents).type.toBe<{ + expect(nested.containingState).type.toBe() + expect(nested.ancestors).type.toBe<{ readonly up: Up readonly "up.auth": Auth }>() - expect(nested.parents.up).type.toBe() - expect(nested.parents["up.auth"]).type.toBe() - expect(nested.parents).type.not.toHaveProperty("up.sync") + expect(nested.ancestors.up).type.toBe() + expect(nested.ancestors["up.auth"]).type.toBe() + expect(nested.ancestors).type.not.toHaveProperty("up.sync") expect(nested).type.not.toHaveProperty("action") type NestedParents = { @@ -249,7 +249,7 @@ describe("Machine", () => { readonly [typeof SignIn], [], "up.auth.signedOut" - >["parents"] + >["ancestors"] >().type.toBe() expect< Machine.Machine.InvokeContext< @@ -257,7 +257,7 @@ describe("Machine", () => { readonly [typeof SignIn], [], "up.auth.signedOut" - >["parent"] + >["containingState"] >().type.toBe() expect< Machine.Machine.InvokeContext< @@ -265,7 +265,7 @@ describe("Machine", () => { readonly [typeof SignIn], [], "up.auth.signedOut" - >["parents"] + >["ancestors"] >().type.toBe() expect< Machine.Machine.AlwaysContext< @@ -273,7 +273,7 @@ describe("Machine", () => { readonly [typeof SignIn], [], "up.auth.signedOut" - >["parents"] + >["ancestors"] >().type.toBe() expect< Machine.Machine.DoneContext< @@ -281,26 +281,26 @@ describe("Machine", () => { readonly [typeof SignIn], [], "up.auth.signedOut" - >["parents"] + >["ancestors"] >().type.toBe() expect< Machine.Machine.FinalOutputContext< typeof UpStates.states, readonly [typeof SignIn], "up.auth.signedOut" - >["parents"] + >["ancestors"] >().type.toBe() expect< Machine.Machine.ParallelOutputContext< typeof UpStates.states, readonly [typeof SignIn], "up.auth.signedOut" - >["parents"] + >["ancestors"] >().type.toBe() const root = null as unknown as SignInContext - expect(root.parent).type.toBe() - expect(root.parents).type.toBe<{}>() + expect(root.containingState).type.toBe() + expect(root.ancestors).type.toBe<{}>() expect< Machine.Machine.ParentStateValue< @@ -458,7 +458,7 @@ describe("Machine", () => { const machine = Machine.make({ states: UpStates.states, events: Machine.events(SignIn), - emits: [SignInCompleted], + emittedEvents: Machine.emittedEvents(SignInCompleted), initial: () => UpStates.initial.down(new Down({})) }).handle({ down: { @@ -820,7 +820,7 @@ describe("Machine", () => { const child = Machine.make({ states: childStates.states, events: Machine.events(SignIn), - emits: [SignIn], + emittedEvents: Machine.emittedEvents(SignIn), input: ChildInput, initial: () => childStates.initial.done(new Down({})) }).handle({ @@ -859,7 +859,7 @@ describe("Machine", () => { const incompatibleEmits = Machine.make({ states: childStates.states, events: Machine.events(SignIn), - emits: [Down], + emittedEvents: Machine.emittedEvents(Down), input: ChildInput, initial: () => childStates.initial.done(new Down({})) }).handle({ diff --git a/typetest/machine/StructuralStates.tst.ts b/typetest/machine/StructuralStates.tst.ts index cd29753..c9d5b11 100644 --- a/typetest/machine/StructuralStates.tst.ts +++ b/typetest/machine/StructuralStates.tst.ts @@ -139,10 +139,10 @@ describe("structural active state types", () => { expect(state).type.toBe() }, on: { - Select: ({ parent, parents, state, target }) => { + Select: ({ containingState, ancestors, state, target }) => { expect(state).type.toBe() - expect(parent).type.toBe() - expect(parents).type.toBe<{}>() + expect(containingState).type.toBe() + expect(ancestors).type.toBe<{}>() expect(target.local).type.not.toHaveProperty("with") expect(target.local.Empty.from).type.toBeCallableWith() expect(target.local.Empty.from).type.not.toBeCallableWith({}) @@ -165,14 +165,14 @@ describe("structural active state types", () => { states: { Paused: { on: { - Play: ({ parent, parents, state, target }) => { + Play: ({ containingState, ancestors, state, target }) => { expect(state).type.toBe() - expect(parent).type.toBe() - expect(parents).type.toBe<{ readonly "player.transport.Ready": Ready }>() + expect(containingState).type.toBe() + expect(ancestors).type.toBe<{ readonly "player.transport.Ready": Ready }>() expect(target.local).type.toHaveProperty("with") expect(target.local.Playing.from).type.toBeCallableWith({ position: 0 }) return target.local.with.from( - { duration: parent.duration }, + { duration: containingState.duration }, (ready) => ready.Playing.from({ position: 0 }) ) }