diff --git a/.changeset/live-machine-inspection.md b/.changeset/live-machine-inspection.md new file mode 100644 index 0000000..074bd9f --- /dev/null +++ b/.changeset/live-machine-inspection.md @@ -0,0 +1,20 @@ +--- +"@typeonce/effect-machine": minor +--- + +Add live, root-scoped machine inspection through `Machine.prepare(machine).inspection` and `AtomMachine.inspection(machineAtom)`. + +The hot Effect `Stream` observes ordered creation, initialization, mailbox delivery and processing, state changes, emissions, Effect and timer activities, and termination for a prepared root and all locally owned descendants: + +```ts +const prepared = yield * Machine.prepare(machine) + +yield * prepared.inspection.pipe( + Stream.runForEach((event) => Console.log(event.sequence, event.subject.id, event._tag)), + Effect.forkScoped({ startImmediately: true }) +) + +const ref = yield * prepared.start +``` + +Inspection is non-replayed, never fails, and completes with the root. Its session ids and ordering are local to one prepared ownership tree; distributed identity and delivery remain an Effect Cluster concern. diff --git a/README.md b/README.md index 621dc52..e833883 100644 --- a/README.md +++ b/README.md @@ -227,6 +227,56 @@ const ref = yield * prepared.start not observe startup emissions. Preparation does not retain or replay an emission: the observer is simply subscribed before initialization begins. +### Inspect a live machine tree + +`Machine.prepare(machine).inspection` is the operational counterpart to the +domain-facing `changes` and `emissions` streams. It observes the prepared root +and every locally owned child, `Logic` process, Effect, and timer in one total +publication order: + +```ts +const prepared = yield * Machine.prepare(checkout) + +yield * prepared.inspection.pipe( + Stream.runForEach((record) => Console.log(record.sequence, record.subject.id, record._tag)), + Effect.forkScoped({ startImmediately: true }) +) + +const checkoutRef = yield * prepared.start +``` + +For a handled input, the stream may expose values such as: + +```ts +{ _tag: "EventSent", sequence: 2, deliveryId: 0, + subject: { id: "checkout", sessionId: "machine:0", kind: "Machine" }, + source: undefined, target: { id: "checkout", sessionId: "machine:0" }, + event: CheckoutEvents.Submit(), causedBy: undefined } + +{ _tag: "EventProcessed", sequence: 4, macrostepId: 0, + deliveryId: 0, handled: true, configurationChanged: true, + before: { status: "active", state: /* ... */ }, + after: { status: "active", state: /* ... */ }, microsteps: [/* ... */] } +``` + +The closed `Machine.Inspection.Event` union also reports creation, +initialization and startup failure, direct `Logic` state updates, outward +emissions, Effect/timer activity lifecycles, and termination. Records erase +unrelated child protocols to `unknown`; application-level observation remains +typed through each reference's `changes` and `emissions`. + +The stream is hot, non-replayed, never fails, and completes after the root +terminates. Subscribe before `prepared.start` to capture initialization. Local +session ids are unique only inside that prepared ownership tree: `machine:0` +is the root and later ids identify its descendants. They are intentionally not +distributed identities. Cluster placement, routing, and correlation continue +to use Cluster entity, runner, and request identities at the integration +boundary. + +`AtomMachine.inspection(machineAtom)` provides the same root-scoped stream and +starts a fresh atom-backed machine only after its inspection subscription is +installed. + Invalid event and emission constructions fail the machine with a typed `MachineSchemaDecodeError`; they do not throw from the constructor call. diff --git a/docs/agent-guide.md b/docs/agent-guide.md index c59e26e..c3c2b97 100644 --- a/docs/agent-guide.md +++ b/docs/agent-guide.md @@ -548,6 +548,51 @@ yield* prepared.emissions.pipe( const ref = yield* prepared.start ``` +`prepared.inspection` is a third, operational stream. It covers the root and +its complete local ownership tree rather than one machine protocol. Subscribe +before `prepared.start` when creation and initialization records matter: + +```ts +const prepared = yield* Machine.prepare(machine) +yield* prepared.inspection.pipe( + Stream.runForEach((event) => Console.log(event.sequence, event.subject.id, event._tag)), + Effect.forkScoped({ startImmediately: true }) +) +const ref = yield* prepared.start +``` + +`Machine.Inspection.Event` is a closed union: + +- `Created`, `Initialized`, and `StartFailed` describe process startup; +- `EventSent` records accepted mailbox delivery and `EventProcessed` records + the committed macrostep, including retained transitions, raised events, + emissions, commands, and entry/exit paths for each microstep; +- `StateChanged` describes direct updates made by generic `Logic`; +- `Emitted` records actual outward notification publication; +- `ActivityStarted` and `ActivityStopped` describe Effect and timer invokes; +- `Terminated` carries the final `done`, `error`, or `stopped` snapshot. + +Every record has a root-local `sequence`, `rootSessionId`, and `subject`. +`deliveryId` correlates acceptance with processing; `macrostepId` correlates +work caused by one statechart input. `source` is present for sends originating +inside the inspected tree. `origin` distinguishes a root, state-owned invoke, +and explicit spawn. Child machine and generic process protocols are erased to +`unknown` because one stream can contain unrelated types. + +Inspection is hot, non-replayed, never fails, and completes with the prepared +root. It is not a replacement for `changes`, which retains the latest lifecycle +snapshot, or `emissions`, which remains the typed domain-notification channel. +Invalid decoded inputs or emissions still fail the owning machine through its +typed `MachineSchemaDecodeError`; inspection never turns validation into a +throw or a stream failure. + +Session ids are deterministic and unique only inside one prepared local tree +(`machine:0`, `machine:1`, ...). Do not persist them as globally unique actor +ids. Distributed identity, placement, delivery, and request correlation belong +to Effect Cluster and its entity, runner, shard, and request identifiers. A +Cluster adapter may translate local inspection records into telemetry, but the +core machine stream does not claim cross-node identity or ordering. + For child-to-parent input, export a public builder protocol and reuse it at both composition boundaries: @@ -590,6 +635,11 @@ Atom-backed machines retain the same transient semantics. Use return streams requiring the corresponding `AtomRegistry`; emissions are not stored as atom state. +Use `AtomMachine.inspection(machineAtom)` for root-scoped operational records. +It installs the subscription before a fresh bridge starts, so initialization, +owned children, and activities are visible without storing inspection records +in 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. diff --git a/src/Machine.ts b/src/Machine.ts index 48a7cc5..cab0a81 100644 --- a/src/Machine.ts +++ b/src/Machine.ts @@ -7,6 +7,7 @@ import type * as Cause from "effect/Cause" import type * as Duration from "effect/Duration" import type * as Effect from "effect/Effect" +import type * as Exit from "effect/Exit" import type * as Option from "effect/Option" import type { Pipeable } from "effect/Pipeable" import { hasProperty } from "effect/Predicate" @@ -1664,6 +1665,187 @@ export type RuntimeSnapshot = readonly state: State } +/** + * Live, root-scoped records describing one prepared machine tree. + * + * Inspection records deliberately erase machine-specific values to `unknown`: + * one stream may contain unrelated root, child-machine, and `Logic` protocols. + * The record structure remains a closed discriminated union, while typed + * application observation continues through `changes` and `emissions`. + * + * @category models + * @since 0.13.0 + */ +export declare namespace Inspection { + /** Read-only identity of a local machine endpoint. */ + export interface Endpoint { + readonly id: string + readonly sessionId: string + } + + /** Read-only identity of a runtime represented in the inspection tree. */ + export interface Subject extends Endpoint { + readonly kind: "Machine" | "Logic" + } + + /** Causal origin of an inspected runtime. */ + export type Origin = + | { readonly _tag: "Root" } + | { + readonly _tag: "Invoke" + readonly ownerPath: string + readonly invokeId: string + } + | { + readonly _tag: "Spawn" + readonly address: string | undefined + } + + /** Causal owner of a send or emission. */ + export type Causation = + | { readonly _tag: "Initialization" } + | { readonly _tag: "Macrostep"; readonly macrostepId: number } + | { readonly _tag: "Activity"; readonly activitySessionId: string } + + /** Closed command projection safe for observation. */ + export type Command = + | { + readonly _tag: "SendTo" + readonly target: Endpoint | { readonly id: string } + readonly event: unknown + } + | { + readonly _tag: "Stop" + readonly target: Endpoint | { readonly id: string } + } + + /** One statechart microstep in a committed macrostep. */ + export interface Microstep { + readonly event: unknown + readonly transitions: ReadonlyArray + readonly raisedEvents: ReadonlyArray + readonly emittedEvents: ReadonlyArray + readonly commands: ReadonlyArray + readonly exitPaths: ReadonlyArray + readonly entryPaths: ReadonlyArray + readonly changed: boolean + } + + /** Common ordering and identity fields for every record. */ + export interface Base { + /** Total publication order within this prepared root. */ + readonly sequence: number + /** Session id of the prepared root that owns this inspection stream. */ + readonly rootSessionId: string + /** Runtime instance described by this record. */ + readonly subject: Subject + } + + /** Announces allocation of a root or owned process identity. */ + export interface Created extends Base { + readonly _tag: "Created" + readonly parent: Subject | undefined + readonly origin: Origin + /** Compiled statechart definition; absent for generic `Logic`. */ + readonly definition: Machine.Any | undefined + } + + /** Announces successful initialization. */ + export interface Initialized extends Base { + readonly _tag: "Initialized" + readonly snapshot: RuntimeSnapshot + readonly initialEntryPaths: ReadonlyArray + readonly microsteps: ReadonlyArray + } + + /** Announces failure before an initial runtime snapshot exists. */ + export interface StartFailed extends Base { + readonly _tag: "StartFailed" + readonly cause: Cause.Cause + } + + /** Announces acceptance of an event by a local mailbox. */ + export interface EventSent extends Base { + readonly _tag: "EventSent" + readonly deliveryId: number + readonly source: Subject | undefined + readonly target: Endpoint + readonly event: unknown + readonly causedBy: Causation | undefined + } + + /** Announces complete processing of one statechart mailbox event. */ + export interface EventProcessed extends Base { + readonly _tag: "EventProcessed" + readonly macrostepId: number + readonly deliveryId: number + readonly source: Subject | undefined + readonly event: unknown + readonly before: RuntimeSnapshot + readonly after: RuntimeSnapshot + readonly handled: boolean + readonly configurationChanged: boolean + readonly microsteps: ReadonlyArray + } + + /** Announces a direct state update made by generic `Logic`. */ + export interface StateChanged extends Base { + readonly _tag: "StateChanged" + readonly before: unknown + readonly after: unknown + readonly causedByDeliveryId: number | undefined + } + + /** Announces actual publication on a machine's domain emission stream. */ + export interface Emitted extends Base { + readonly _tag: "Emitted" + readonly emission: unknown + readonly causedBy: Causation | undefined + } + + /** Identity of one Effect or timer invocation run. */ + export interface Activity { + readonly id: string + readonly sessionId: string + readonly owner: Subject + readonly ownerPath: string + readonly kind: "Effect" | "Timer" + } + + /** Announces an Effect or timer invocation starting. */ + export interface ActivityStarted extends Base { + readonly _tag: "ActivityStarted" + readonly activity: Activity + } + + /** Announces an Effect or timer invocation outcome. */ + export interface ActivityStopped extends Base { + readonly _tag: "ActivityStopped" + readonly activity: Activity + /** Success or failure from the invoke; interruption means its owner stopped it. */ + readonly exit: Exit.Exit + } + + /** Announces a terminal local runtime snapshot. */ + export interface Terminated extends Base { + readonly _tag: "Terminated" + readonly snapshot: RuntimeSnapshot + } + + /** Complete live inspection protocol. */ + export type Event = + | Created + | Initialized + | StartFailed + | EventSent + | EventProcessed + | StateChanged + | Emitted + | ActivityStarted + | ActivityStopped + | Terminated +} + /** * Represents a classified terminal outcome derived from a runtime snapshot. * @@ -1726,6 +1908,14 @@ export interface Prepared + /** + * Streams ordered operational records for this root and every locally owned + * descendant. The stream is hot, non-replayed, never fails, and completes + * after the prepared root terminates. Subscribe before evaluating `start` to + * observe initialization. + */ + readonly inspection: Stream.Stream + /** Initializes this machine once and returns its running reference. */ readonly start: Effect.Effect, StartError, StartRequirements> } diff --git a/src/internal/machine/atom.ts b/src/internal/machine/atom.ts index 8549989..e29fa71 100644 --- a/src/internal/machine/atom.ts +++ b/src/internal/machine/atom.ts @@ -139,6 +139,32 @@ export const emissions = ( ) } +export const inspection = ( + self: MachineAtom +): Stream.Stream => { + const prepared = preparedByMachineAtom.get(self as object) + if (prepared === undefined) return Stream.empty + return Stream.unwrap( + Effect.gen(function*() { + const registry = yield* AtomRegistry.AtomRegistry + const releasePrepared = yield* Effect.sync(() => registry.mount(prepared)) + yield* Effect.addFinalizer(() => Effect.sync(releasePrepared)) + const machine = yield* Atom.getResult(prepared) + const pull = yield* Stream.toPull(machine.inspection) + const firstPull = yield* pull.pipe(Effect.forkScoped({ startImmediately: true })) + const releaseRef = yield* Effect.sync(() => registry.mount(self.ref)) + yield* Effect.addFinalizer(() => Effect.sync(releaseRef)) + yield* Atom.getResult(self.ref) + let first = true + return Stream.fromPull(Effect.succeed(Effect.suspend(() => { + if (!first) return pull + first = false + return Fiber.join(firstPull) + }))) + }) + ) +} + export const childEmissions = ( self: ChildMachineAtom ): Stream.Stream>, StartError, AtomRegistry.AtomRegistry> => diff --git a/src/internal/machine/commandRuntime.ts b/src/internal/machine/commandRuntime.ts index da2ffe1..04df1f4 100644 --- a/src/internal/machine/commandRuntime.ts +++ b/src/internal/machine/commandRuntime.ts @@ -10,18 +10,13 @@ import type { RuntimeCommand } from "./command.js" import { decodeEmit, decodeEvent } from "./protocol.js" import type { ProcessScope } from "./runtime.js" -const isMachineTarget = ( - 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 ): Runtime => ({ raise: (event) => decodeEvent(machine, event).pipe( - Effect.flatMap((event) => scope.self.send(event as Events)) + Effect.flatMap((event) => scope.sendTo(scope.self, event as Events)) ), emit: (event) => decodeEmit(machine, event).pipe( @@ -35,9 +30,7 @@ export const runCommands = ( ) => Effect.forEach(commands, (command) => command._tag === "SendTo" - ? isMachineTarget(command.target) - ? command.target.send(command.event) - : scope.sendTo(command.target as never, command.event) + ? scope.sendTo(command.target as never, command.event) : scope.stopChild(command.child as never), { discard: true }) export const runEmittedEvents = ( diff --git a/src/internal/machine/executionPlan.ts b/src/internal/machine/executionPlan.ts index 38c04ff..9c30115 100644 --- a/src/internal/machine/executionPlan.ts +++ b/src/internal/machine/executionPlan.ts @@ -402,6 +402,7 @@ const emptyExecutionValues: ReadonlyArray = Object.freeze([]) export interface ExecutionMicrostep { readonly next: State readonly event: unknown + readonly transitions?: ReadonlyArray readonly commands: ReadonlyArray readonly raisedEvents: ReadonlyArray readonly emittedEvents: ReadonlyArray @@ -588,13 +589,22 @@ const indexedMicrostep = ( descriptor: IndexedExecutionDescriptor, state: OwnedIndexedState, event: any, - selections: ReadonlyArray + selections: ReadonlyArray, + retainTransitions: boolean ): ExecutionMicrostep => { + const retain = (transition: IndexedEvaluatedTransition): Machine.RetainedTransition => ({ + source: transition.selection.sourcePath, + trigger: transition.selection.trigger, + reenter: transition.selection.transition.reenter, + target: transition.unresolvedTarget === undefined ? undefined : getTargetNodePath(transition.unresolvedTarget), + resolvedTarget: transition.target === undefined ? undefined : getTargetNodePath(transition.target) + }) if (selections.length === 1) { const transition = collectIndexedEvaluatedTransition(machine, descriptor, state, selections[0]!) return { next: transition.next, event, + ...(retainTransitions ? { transitions: [retain(transition)] } : undefined), commands: transition.commands, raisedEvents: transition.raisedEvents, emittedEvents: transition.emittedEvents, @@ -645,6 +655,7 @@ const indexedMicrostep = ( return { next, event, + ...(retainTransitions ? { transitions: transitions.map(retain) } : undefined), commands, raisedEvents, emittedEvents, @@ -762,6 +773,17 @@ const planIndexedFlatState = ( const step: ExecutionMicrostep = { next, event, + ...(retainMicrosteps + ? { + transitions: [{ + source: sourcePath, + trigger: { type: "event", event: event._tag }, + reenter: transition.reenter, + target: target === undefined ? undefined : getTargetNodePath(target as any), + resolvedTarget: target === undefined ? undefined : getTargetNodePath(target as any) + }] + } + : undefined), commands: transitionResult.commands, raisedEvents: transitionResult.raisedEvents, emittedEvents: transitionResult.emittedEvents, @@ -828,6 +850,7 @@ const planIndexedState = ( microsteps: planned.microsteps.map((step) => ({ next: ownedIndexedStateFromActive(descriptor, step.next), event: step.event, + transitions: step.transitions, commands: step.commands, raisedEvents: step.raisedEvents, emittedEvents: step.emittedEvents, @@ -874,7 +897,7 @@ const planIndexedState = ( } } - const first = indexedMicrostep(machine, descriptor, configuration, decoded, selections) + const first = indexedMicrostep(machine, descriptor, configuration, decoded, selections, retainMicrosteps) let current = first.next let currentEvent: any = decoded const commands = [...first.commands] @@ -932,7 +955,7 @@ const planIndexedState = ( currentEvent = raised const raisedSelections = selectIndexedEventTransitions(machine, descriptor, current, raised, machineReferences) if (raisedSelections.length === 0) continue - const step = indexedMicrostep(machine, descriptor, current, raised, raisedSelections) + const step = indexedMicrostep(machine, descriptor, current, raised, raisedSelections, retainMicrosteps) current = step.next commands.push(...step.commands) raisedEvents.push(...step.raisedEvents) diff --git a/src/internal/machine/inspectionRuntime.ts b/src/internal/machine/inspectionRuntime.ts new file mode 100644 index 0000000..6f64479 --- /dev/null +++ b/src/internal/machine/inspectionRuntime.ts @@ -0,0 +1,102 @@ +/** + * Root-scoped live inspection publication and evidence projection. + * + * @since 0.13.0 + */ + +import * as Effect from "effect/Effect" +import * as PubSub from "effect/PubSub" +import * as Stream from "effect/Stream" +import type { Inspection } from "../../Machine.js" + +export type Draft = Event extends Inspection.Event + ? Omit + : never + +export interface Runtime { + readonly rootSessionId: string + readonly stream: Stream.Stream + readonly isActive: () => boolean + readonly nextDeliveryId: () => number + readonly nextMacrostepId: () => number + readonly publishUnsafe: (event: Draft) => void + readonly close: Effect.Effect +} + +export const make = (rootSessionId: string): Effect.Effect => + Effect.gen(function*() { + const pubsub = yield* PubSub.unbounded() + let subscribers = 0 + let sequence = 0 + let deliveryId = 0 + let macrostepId = 0 + return { + rootSessionId, + stream: Stream.fromPubSub(pubsub).pipe( + Stream.onStart(Effect.sync(() => { + subscribers += 1 + })), + Stream.ensuring(Effect.sync(() => { + subscribers -= 1 + })) + ), + isActive: () => subscribers > 0, + nextDeliveryId: () => deliveryId++, + nextMacrostepId: () => macrostepId++, + publishUnsafe: (event) => { + if (subscribers === 0) return + PubSub.publishUnsafe(pubsub, { + ...event, + sequence: sequence++, + rootSessionId + } as Inspection.Event) + }, + close: PubSub.shutdown(pubsub) + } + }) + +export const endpoint = ( + target: { readonly id: string; readonly sessionId: string } +): Inspection.Endpoint => ({ + id: target.id, + sessionId: target.sessionId +}) + +const commands = (values: ReadonlyArray): ReadonlyArray => + values.flatMap((command): ReadonlyArray => { + if (typeof command !== "object" || command === null || !("_tag" in command)) return [] + if (command._tag === "SendTo" && "target" in command && "event" in command) { + const target = command.target + return typeof target === "object" && target !== null && "id" in target + ? [{ + _tag: "SendTo", + target: "sessionId" in target ? endpoint(target as any) : { id: String(target.id) }, + event: command.event + }] + : [] + } + if (command._tag === "Stop" && "child" in command) { + const child = command.child + if (typeof child === "string") return [{ _tag: "Stop", target: { id: child } }] + return typeof child === "object" && child !== null && "id" in child + ? [{ _tag: "Stop", target: "sessionId" in child ? endpoint(child as any) : { id: String(child.id) } }] + : [] + } + return [] + }) + +export const microsteps = (plan: unknown): ReadonlyArray => { + if (typeof plan !== "object" || plan === null || !("microsteps" in plan) || !Array.isArray(plan.microsteps)) { + return [] + } + return plan.microsteps.map((step: any) => ({ + event: step.event, + transitions: Array.isArray(step.transitions) ? step.transitions : [], + raisedEvents: Array.isArray(step.raisedEvents) ? step.raisedEvents : [], + emittedEvents: Array.isArray(step.emittedEvents) ? step.emittedEvents : [], + commands: commands(Array.isArray(step.commands) ? step.commands : []), + exitPaths: Array.isArray(step.exitPaths) ? step.exitPaths : [], + entryPaths: Array.isArray(step.entryPaths) ? step.entryPaths : [], + changed: step.changed === true + })) +} diff --git a/src/internal/machine/invocation.ts b/src/internal/machine/invocation.ts index 9769aa4..81fa5a9 100644 --- a/src/internal/machine/invocation.ts +++ b/src/internal/machine/invocation.ts @@ -6,7 +6,7 @@ import * as Cause from "effect/Cause" import * as Effect from "effect/Effect" -import type { ChildMachine, Logic, Machine } from "../../Machine.js" +import type { ChildMachine, Inspection, Logic, Machine } from "../../Machine.js" import * as Configuration from "./configuration.js" import { InfiniteTransitionError, MachineSchemaDecodeError, StoppedError } from "./errors.js" import * as InvocationEvent from "./invocationEvent.js" @@ -23,6 +23,7 @@ export interface AnyConfig { readonly onDone?: unknown readonly onFailure?: unknown readonly onSnapshot?: unknown + readonly activityKind?: Inspection.Activity["kind"] } /** @internal */ @@ -57,7 +58,8 @@ const resolveOne = ( >, onDone: raw.onDone, onFailure: raw.onFailure, - onSnapshot: raw.onSnapshot + onSnapshot: raw.onSnapshot, + activityKind: "Effect" } } if ("after" in raw) { @@ -72,7 +74,8 @@ const resolveOne = ( any, any >, - onDone: raw.onDone + onDone: raw.onDone, + activityKind: "Timer" } } if ("logic" in raw) { @@ -115,8 +118,18 @@ const runSequentialDiscard = ( const sendLifecycle = ( scope: Runtime.ProcessScope, - event: InvocationEvent.InvocationEvent -): Effect.Effect => scope.self.send(event).pipe(Effect.catchTag("StoppedError", () => Effect.void)) + event: InvocationEvent.InvocationEvent, + activitySessionId?: string +): Effect.Effect => + (activitySessionId === undefined + ? scope.self.send(event) + : scope.self.sendInspected === undefined + ? scope.self.send(event) + : scope.self.sendInspected( + event, + scope.self.inspectionSubject, + { _tag: "Activity", activitySessionId } + )).pipe(Effect.catchTag("StoppedError", () => Effect.void)) const isFrameworkFailure = (error: unknown): boolean => error instanceof InfiniteTransitionError || error instanceof MachineSchemaDecodeError || error instanceof StoppedError @@ -131,7 +144,8 @@ const startResolved = ( src: () => Runtime.ProcessLogic, onDone: unknown, onFailure: unknown, - onSnapshot: unknown + onSnapshot: unknown, + activityKind?: Inspection.Activity["kind"] ): Effect.Effect => Effect.suspend(() => { const key = makeKey(path, invokeId) @@ -141,8 +155,9 @@ const startResolved = ( id: childId, duplicateId: invokeId, ...(descriptor === undefined ? undefined : { descriptor }), + ...(activityKind === undefined ? undefined : { activityKind }), sendParent: (isCurrent, event) => isCurrent() ? scope.self.send(event) : Effect.void, - onOutcome: (isCurrent, outcome) => { + onOutcome: (isCurrent, outcome, activitySessionId) => { if (outcome._tag === "Stopped" || !isCurrent()) return Effect.void if (outcome._tag === "Done") { if (onDone === undefined) { @@ -152,14 +167,22 @@ const startResolved = ( ) )) } - return sendLifecycle(scope, InvocationEvent.done(path, invokeId, outcome.output)) + return sendLifecycle( + scope, + InvocationEvent.done(path, invokeId, outcome.output), + activityKind === undefined ? undefined : activitySessionId + ) } if ( outcome._tag === "Failure" && onFailure !== undefined && (descriptor === undefined || !isFrameworkFailure(outcome.error)) ) { - return sendLifecycle(scope, InvocationEvent.failure(path, invokeId, outcome.error)) + return sendLifecycle( + scope, + InvocationEvent.failure(path, invokeId, outcome.error), + activityKind === undefined ? undefined : activitySessionId + ) } return scope.failCause(outcome.cause) }, @@ -189,7 +212,8 @@ const start = ( config.src, config.onDone, config.onFailure, - config.onSnapshot + config.onSnapshot, + config.activityKind ) } @@ -213,7 +237,8 @@ const startStaticChild = ( descriptor[ChildMachineLogicTypeId], raw.onDone, raw.onFailure, - raw.onSnapshot + raw.onSnapshot, + undefined ) } diff --git a/src/internal/machine/process.ts b/src/internal/machine/process.ts index 8b83f71..d5ff71d 100644 --- a/src/internal/machine/process.ts +++ b/src/internal/machine/process.ts @@ -73,8 +73,7 @@ const acknowledgedPlan = ( commands: planned.commands, emittedEvents: planned.emittedEvents, microsteps: planned.microsteps.map((microstep) => { - const { transitions: _, ...evidence } = microstep - return { ...evidence, next: snapshot(microstep.next) } + return { ...microstep, transitions: microstep.transitions ?? [], next: snapshot(microstep.next) } }), done: planned.done, output: planned.output @@ -433,6 +432,7 @@ const makeProcessLogic: < ) => { try { const planned = compiledInitial(initialArgs, scope) + scope.inspectInitial(planned.initialEntryPaths) const result = { state: planned.state as Machine.Snapshot, done: planned.done, @@ -461,6 +461,7 @@ const makeProcessLogic: < ? internalRuntime.provideMachineRuntime( internalPlanner.planInitial(internalPlanner.withMachineReferences(machine, scope), ...initialArgs).pipe( Effect.flatMap((planned) => { + scope.inspectInitial(planned.initialEntryPaths, planned.microsteps) const commands = planned.commands.length === 0 ? undefined : CommandRuntime.runCommands(planned.commands, scope) @@ -486,6 +487,7 @@ const makeProcessLogic: < ) : Effect.try({ try: () => makeCompiledInitial!(scope), catch: (error) => error as any }) return ({ + inspection: { kind: "Machine", definition: machine }, execution: { _tag: "Compiled", childless: !hasInvokes, diff --git a/src/internal/machine/runtime.ts b/src/internal/machine/runtime.ts index 08b12db..a02b2e0 100644 --- a/src/internal/machine/runtime.ts +++ b/src/internal/machine/runtime.ts @@ -18,8 +18,9 @@ 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 { MachineTarget } from "../../Machine.js" +import type { Inspection, Machine as MachineDefinition, MachineTarget } from "../../Machine.js" import { ChildAlreadyExistsError, StoppedError } from "./errors.js" +import * as InspectionRuntime from "./inspectionRuntime.js" type ChildDescriptor = { readonly id: string @@ -69,7 +70,8 @@ const AcknowledgedMessageTypeId: unique symbol = Symbol("effect/Machine/Acknowle export interface AcknowledgedMessage { readonly [AcknowledgedMessageTypeId]: true readonly event: Event - readonly deferred: Deferred.Deferred, unknown> + readonly deferred?: Deferred.Deferred, unknown> + readonly inspection?: InspectedDelivery } /** @internal */ @@ -90,7 +92,10 @@ const succeedAcknowledgedMessage = ( delivery: AcknowledgedDelivery ): void => { if (message !== undefined && isAcknowledgedMessage(message)) { - Deferred.doneUnsafe(message.deferred, Effect.succeed(delivery as AcknowledgedDelivery)) + if (message.deferred !== undefined) { + Deferred.doneUnsafe(message.deferred, Effect.succeed(delivery as AcknowledgedDelivery)) + } + message.inspection?.complete(delivery as AcknowledgedDelivery) } } @@ -99,16 +104,32 @@ const failAcknowledgedMessage = ( cause: Cause.Cause ): void => { if (message !== undefined && isAcknowledgedMessage(message)) { - Deferred.doneUnsafe(message.deferred, Effect.failCause(cause)) + if (message.deferred !== undefined) Deferred.doneUnsafe(message.deferred, Effect.failCause(cause)) } } const stopAcknowledgedMessage = (message: ProcessMessage | undefined): void => { if (message !== undefined && isAcknowledgedMessage(message)) { - Deferred.doneUnsafe(message.deferred, Effect.fail(new StoppedError())) + if (message.deferred !== undefined) Deferred.doneUnsafe(message.deferred, Effect.fail(new StoppedError())) } } +interface InspectedDelivery { + readonly deliveryId: number + readonly macrostepId: number + readonly source: Inspection.Subject | undefined + readonly event: unknown + readonly causedBy: Inspection.Causation | undefined + readonly complete: (delivery: AcknowledgedDelivery) => void +} + +type InspectedOffer = ( + event: Event, + source: Inspection.Subject | undefined, + causedBy: Inspection.Causation | undefined, + deferred?: Deferred.Deferred, unknown> +) => Effect.Effect + type ChildObservation = Option.Option> type ChildObservationBatch = [ChildObservation, ...Array] @@ -295,6 +316,10 @@ export interface MachineRef readonly stop: Effect.Effect readonly send: (event: Event) => Effect.Effect + /** @internal */ + readonly inspectionSubject?: Inspection.Subject + /** @internal */ + readonly sendInspected?: ProcessAddress["sendInspected"] readonly [acknowledgedSend]?: ( event: Event ) => Effect.Effect, Error | StoppedError> @@ -315,19 +340,36 @@ export interface PreparedProcess< readonly sessionId: string readonly changes: Stream.Stream, StartError> readonly emissions: Stream.Stream + readonly inspection: Stream.Stream readonly start: Effect.Effect, StartError, StartRequirements> } interface ProcessAddress { readonly id: string readonly sessionId: string + readonly inspectionSubject?: Inspection.Subject readonly stop: Effect.Effect readonly send: (event: Event) => Effect.Effect + readonly sendInspected?: ( + event: Event, + source: Inspection.Subject | undefined, + causedBy: Inspection.Causation | undefined + ) => Effect.Effect } -const isProcessAddress = (value: unknown): value is MachineTarget => +const isMachineTarget = (value: unknown): value is MachineTarget => typeof value === "object" && value !== null && "send" in value && typeof value.send === "function" +const sendMachineTarget = ( + target: MachineTarget, + event: unknown, + source: Inspection.Subject | undefined, + causedBy: Inspection.Causation | undefined +): Effect.Effect => + "inspectionSubject" in target && "sendInspected" in target && typeof target.sendInspected === "function" + ? target.sendInspected(event, source, causedBy) + : target.send(event) + export interface ProcessScope { readonly self: ProcessAddress readonly parent: ProcessAddress | undefined @@ -341,6 +383,11 @@ export interface ProcessScope { readonly stopChild: (child: ChildSelector) => Effect.Effect /** @internal */ readonly failCause: (cause: Cause.Cause) => Effect.Effect + /** @internal */ + readonly inspectInitial: ( + initialEntryPaths: ReadonlyArray, + microsteps?: ReadonlyArray + ) => void } export interface ProcessContext extends ProcessScope { @@ -493,6 +540,11 @@ export interface ProcessLogic< > { /** @internal */ readonly execution?: ProcessExecution + /** @internal */ + readonly inspection?: { + readonly kind: Inspection.Subject["kind"] + readonly definition?: MachineDefinition.Any + } initial(scope: ProcessScope): Effect.Effect run(context: ProcessContext): Effect.Effect } @@ -613,6 +665,7 @@ export const watch = ( interface ProcessRuntime { readonly nextSessionId: Effect.Effect + inspection?: InspectionRuntime.Runtime } const makeProcessRuntime: Effect.Effect = Effect.sync(() => { @@ -622,6 +675,85 @@ const makeProcessRuntime: Effect.Effect = Effect.sync(() => { } }) +const inspectionSubject = ( + logic: ProcessLogic, + id: string, + sessionId: string +): Inspection.Subject => ({ + id, + sessionId, + kind: logic.inspection?.kind ?? "Logic" +}) + +const messageCausation = ( + message: ProcessMessage | undefined, + initializing: boolean +): Inspection.Causation | undefined => + initializing + ? { _tag: "Initialization" } + : message !== undefined && isAcknowledgedMessage(message) && message.inspection !== undefined + ? { _tag: "Macrostep", macrostepId: message.inspection.macrostepId } + : undefined + +const makeInspectedMessage = ( + inspection: InspectionRuntime.Runtime, + subject: Inspection.Subject, + event: Event, + source: Inspection.Subject | undefined, + causedBy: Inspection.Causation | undefined, + deferred?: Deferred.Deferred, unknown> +): AcknowledgedMessage => { + const deliveryId = inspection.nextDeliveryId() + const macrostepId = inspection.nextMacrostepId() + const inspected: InspectedDelivery = { + deliveryId, + macrostepId, + source, + event, + causedBy, + complete: (delivery) => { + const microsteps = InspectionRuntime.microsteps(delivery.plan) + inspection.publishUnsafe({ + _tag: "EventProcessed", + subject, + macrostepId, + deliveryId, + source, + event, + before: { status: "active", state: delivery.before }, + after: { status: "active", state: delivery.after }, + handled: microsteps.some((microstep) => microstep.transitions.length > 0), + configurationChanged: microsteps.some((microstep) => microstep.changed), + microsteps + }) + } + } + return { + [AcknowledgedMessageTypeId]: true, + event, + ...(deferred === undefined ? undefined : { deferred }), + inspection: inspected + } +} + +const publishInspectedSent = ( + inspection: InspectionRuntime.Runtime, + subject: Inspection.Subject, + message: ProcessMessage +): void => { + if (!isAcknowledgedMessage(message) || message.inspection === undefined) return + const delivery = message.inspection + inspection.publishUnsafe({ + _tag: "EventSent", + subject, + deliveryId: delivery.deliveryId, + source: delivery.source, + target: InspectionRuntime.endpoint(subject), + event: delivery.event, + causedBy: delivery.causedBy + }) +} + interface StartInternalOptions { readonly detached?: boolean readonly id?: string @@ -642,6 +774,14 @@ interface StartInternalOptions { readonly parent?: ProcessAddress readonly runtime: ProcessRuntime readonly sendParent?: (event: unknown) => Effect.Effect + readonly origin?: Inspection.Origin + readonly activity?: { + readonly id: string + readonly owner: Inspection.Subject + readonly ownerPath: string + readonly kind: Inspection.Activity["kind"] + } + readonly inspectionRoot?: boolean } /** @internal */ @@ -653,7 +793,8 @@ export interface OwnedChildSpawnOptions { readonly descriptor?: ChildDescriptor readonly onOutcome: ( isCurrent: () => boolean, - outcome: RuntimeOutcome + outcome: RuntimeOutcome, + activitySessionId: string | undefined ) => Effect.Effect readonly onSnapshot?: ( isCurrent: () => boolean, @@ -663,6 +804,7 @@ export interface OwnedChildSpawnOptions { isCurrent: () => boolean, event: unknown ) => Effect.Effect + readonly activityKind?: Inspection.Activity["kind"] } /** @internal */ @@ -684,7 +826,12 @@ interface ChildRuntime { readonly changes: ( child: ChildSelector ) => Stream.Stream>> - readonly sendTo: (child: ChildSelector, event: unknown) => Effect.Effect + readonly sendTo: ( + child: ChildSelector, + event: unknown, + source?: Inspection.Subject, + causedBy?: Inspection.Causation + ) => Effect.Effect readonly stop: (child: ChildSelector) => Effect.Effect readonly owned: OwnedChildRuntime } @@ -737,13 +884,15 @@ class OwnedChildRuntimeImpl implements OwnedChildRuntime { }) const parent: ProcessAddress = { ...this.self, - send: (event) => options.sendParent(isCurrent, event) + send: (event) => options.sendParent(isCurrent, event), + sendInspected: (event, source, causedBy) => + isCurrent() ? sendMachineTarget(this.self, event, source, causedBy) : Effect.void } const startOptions: StartInternalOptions = { detached: true, id: options.id, sendParent: (event) => options.sendParent(isCurrent, event), - onOutcome: (outcome) => options.onOutcome(isCurrent, outcome), + onOutcome: (outcome) => options.onOutcome(isCurrent, outcome, startedChild?.sessionId), ...(options.onSnapshot === undefined ? undefined : { onSnapshot: (snapshot) => options.onSnapshot!(isCurrent, snapshot) }), @@ -754,7 +903,19 @@ class OwnedChildRuntimeImpl implements OwnedChildRuntime { onStopSync: () => unregisterChild(this.registry, options.id, token), skipStoppedOutcome: true, parent, - runtime: this.runtime + runtime: this.runtime, + origin: { _tag: "Invoke", ownerPath: options.path, invokeId: options.duplicateId }, + ...(options.activityKind === undefined || this.runtime.inspection === undefined || + this.self.inspectionSubject === undefined + ? undefined + : { + activity: { + id: options.duplicateId, + owner: this.self.inspectionSubject, + ownerPath: options.path, + kind: options.activityKind + } + }) } const execution = logic.execution const synchronous = this.services !== undefined && options.onSnapshot === undefined && @@ -816,6 +977,8 @@ class OwnedChildRuntimeImpl implements OwnedChildRuntime { const noChildChanges = Stream.succeed(Option.none()).pipe(Stream.concat(Stream.never)) const noParentSend = (_event: unknown): Effect.Effect => Effect.void +const noInspectInitial = (_paths: ReadonlyArray, _microsteps?: ReadonlyArray): void => {} +const noCausation = (): Inspection.Causation | undefined => undefined const EmissionsClosed: unique symbol = Symbol("effect/Machine/EmissionsClosed") type LazyEmissions = PubSub.PubSub | typeof EmissionsClosed | undefined @@ -996,7 +1159,12 @@ const makeChildRuntimeSync = ( ) } - const sendTo = (child: ChildSelector, event: unknown): Effect.Effect => { + const sendTo = ( + child: ChildSelector, + event: unknown, + source?: Inspection.Subject, + causedBy?: Inspection.Causation + ): Effect.Effect => { const id = typeof child === "string" ? child : child.id return Effect.suspend(() => { if (registry.closed) { @@ -1004,7 +1172,7 @@ const makeChildRuntimeSync = ( } const entry = registry.children.get(id) return entry !== undefined && matchesChild(entry, child) - ? entry.ref.send(event) + ? sendMachineTarget(entry.ref, event, source, causedBy) : Effect.void }) } @@ -1100,7 +1268,8 @@ const makeChildRuntimeSync = ( ), onStop: unregister(key, token), parent: self, - runtime + runtime, + origin: { _tag: "Spawn", address: spawnOptions?.id } }).pipe( Effect.onExit((exit) => Exit.isFailure(exit) @@ -1172,6 +1341,13 @@ const startGenericInternal: < const sessionId = options.sessionId ?? (yield* runtime.nextSessionId) const id = requestedId ?? sessionId + const inspector = runtime.inspection + const subject = inspector === undefined ? undefined : inspectionSubject(logic, id, sessionId) + const activity: Inspection.Activity | undefined = inspector === undefined || options.activity === undefined + ? undefined + : { ...options.activity, sessionId } + let initialEntryPaths: ReadonlyArray | undefined + let initialMicrosteps: ReadonlyArray | undefined const queue = yield* Queue.unbounded>() const emissions = options.emissions ?? makeEmissionRuntime() const termination = yield* Deferred.make() @@ -1180,6 +1356,26 @@ const startGenericInternal: < let initializing = true let inFlightMessage: ProcessMessage | undefined const requestStop = Deferred.succeed(termination, { _tag: "Stopped" }).pipe(Effect.asVoid) + const offerDirect = (message: ProcessMessage): Effect.Effect => + Queue.offer(queue, message).pipe( + Effect.flatMap((accepted) => accepted ? Effect.void : Effect.fail(new StoppedError())) + ) + const offerInspected: InspectedOffer | undefined = inspector === undefined ? undefined : ( + event, + source, + causedBy, + deferred?: Deferred.Deferred, unknown> + ) => + Effect.suspend(() => { + const message: ProcessMessage = inspector.isActive() + ? makeInspectedMessage(inspector, subject!, event, source, causedBy, deferred) + : deferred === undefined + ? event + : { [AcknowledgedMessageTypeId]: true as const, event, deferred } + return offerDirect(message).pipe( + Effect.tap(() => Effect.sync(() => publishInspectedSent(inspector, subject!, message))) + ) + }) const self: ProcessAddress = { id, sessionId, @@ -1193,10 +1389,12 @@ const startGenericInternal: < ? requestStop : requestStop.pipe(Effect.andThen(Effect.never)) ), - send: (event: Event) => - Queue.offer(queue, event).pipe( - Effect.flatMap((accepted) => accepted ? Effect.void : Effect.fail(new StoppedError())) - ) + send: inspector === undefined + ? (event) => offerDirect(event) + : (event) => offerInspected!(event, undefined, undefined), + ...(inspector === undefined + ? undefined + : { inspectionSubject: subject!, sendInspected: offerInspected! }) } const sendAcknowledged: | ((event: Event) => Effect.Effect, Error | StoppedError>) @@ -1205,19 +1403,12 @@ const startGenericInternal: < : (event) => Effect.uninterruptibleMask((restore) => Deferred.make, unknown>().pipe( - Effect.flatMap((deferred) => - Queue.offer(queue, { - [AcknowledgedMessageTypeId]: true as const, - event, - deferred - }).pipe( - Effect.flatMap((accepted) => - accepted - ? restore(Deferred.await(deferred)) - : Effect.fail(new StoppedError()) - ) - ) - ), + Effect.flatMap((deferred) => { + const offered = inspector === undefined + ? offerDirect({ [AcknowledgedMessageTypeId]: true as const, event, deferred }) + : offerInspected!(event, undefined, undefined, deferred) + return offered.pipe(Effect.andThen(restore(Deferred.await(deferred)))) + }), Effect.map((delivery) => delivery as AcknowledgedDelivery) ) ) as Effect.Effect, Error | StoppedError> @@ -1242,29 +1433,87 @@ const startGenericInternal: < stop: stopChild } = yield* makeChildRuntime(self, runtime)) } - const cleanupStartupFailure = (exit: Exit.Exit): Effect.Effect => - Exit.isFailure(exit) - ? closeChildren(exit).pipe(Effect.andThen(emissions.close())) - : Effect.void + const cleanupStartupFailure = (exit: Exit.Exit): Effect.Effect => { + if (Exit.isSuccess(exit)) return Effect.void + if (inspector !== undefined) { + inspector.publishUnsafe( + activity === undefined + ? { _tag: "StartFailed", subject: subject!, cause: exit.cause } + : { _tag: "ActivityStopped", subject: activity.owner, activity, exit } + ) + } + return closeChildren(exit).pipe( + Effect.andThen(emissions.close()), + Effect.andThen(options.inspectionRoot === true && inspector !== undefined ? inspector.close : Effect.void) + ) + } const cleanup = onStopSync === undefined ? onStop ?? Effect.void : Effect.sync(onStopSync) - const sendParent = overrideSendParent ?? (parent === undefined ? noParentSend : parent.send) + const currentCausation = inspector === undefined + ? noCausation + : (): Inspection.Causation | undefined => messageCausation(inFlightMessage, initializing) + const sendParent = overrideSendParent ?? (parent === undefined + ? noParentSend + : inspector === undefined + ? parent.send + : (event) => sendMachineTarget(parent, event, subject, currentCausation())) + const emit = inspector === undefined + ? emissions.emit + : (event: unknown) => + emissions.emit(event).pipe( + Effect.tap(() => + Effect.sync(() => + inspector.publishUnsafe({ + _tag: "Emitted", + subject: subject!, + emission: event, + causedBy: currentCausation() + }) + ) + ) + ) + const sendToTarget: ProcessScope["sendTo"] = inspector === undefined + ? ((target: unknown, event: unknown) => + isMachineTarget(target) ? target.send(event) : sendTo(target as ChildSelector, event)) as ProcessScope< + Event + >["sendTo"] + : ((target: unknown, event: unknown) => + isMachineTarget(target) + ? sendMachineTarget(target, event, subject, currentCausation()) + : sendTo(target as ChildSelector, event, subject, currentCausation())) as ProcessScope["sendTo"] const scope: ProcessScope = { self, parent, spawn, sendParent, - emit: emissions.emit, - sendTo: ((target: unknown, event: unknown) => - isProcessAddress(target) ? target.send(event) : sendTo(target as ChildSelector, event)) as ProcessScope< - Event - >["sendTo"], + emit, + sendTo: sendToTarget, stopChild, failCause: (cause) => Deferred.succeed(termination, { _tag: "Failure", cause: cause as Cause.Cause - }).pipe(Effect.asVoid) + }).pipe(Effect.asVoid), + inspectInitial: inspector === undefined + ? noInspectInitial + : (paths, microsteps = []) => { + initialEntryPaths = paths + initialMicrosteps = microsteps + } + } + + if (inspector !== undefined) { + inspector.publishUnsafe( + activity === undefined + ? { + _tag: "Created", + subject: subject!, + parent: parent?.inspectionSubject, + origin: options.origin ?? { _tag: "Root" }, + definition: logic.inspection?.definition + } + : { _tag: "ActivityStarted", subject: activity.owner, activity } + ) } const initial = yield* logic.initial(scope).pipe( @@ -1282,6 +1531,15 @@ const startGenericInternal: < state: initial } }) + if (activity === undefined) { + inspector?.publishUnsafe({ + _tag: "Initialized", + subject: subject!, + snapshot: { status: "active", state: initial }, + initialEntryPaths: initialEntryPaths ?? [], + microsteps: InspectionRuntime.microsteps({ microsteps: initialMicrosteps ?? [] }) + }) + } const publishSnapshot: ( snapshot: VersionedSnapshot ) => Effect.Effect> = onSnapshot === undefined @@ -1296,9 +1554,7 @@ 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 = ( @@ -1402,7 +1658,7 @@ const startGenericInternal: < Effect.asVoid ) - const setActiveState = (state: State) => + const setActiveStateDirect = (state: State) => updateSnapshot((snapshot) => Effect.succeed( snapshot.status === "active" @@ -1414,6 +1670,40 @@ const startGenericInternal: < ) ).pipe(Effect.asVoid) + const setActiveState = inspector === undefined ? + setActiveStateDirect : + (state: State) => + SynchronizedRef.get(current).pipe( + Effect.flatMap((before) => + updateSnapshot((snapshot) => + Effect.succeed( + snapshot.status === "active" + ? { + status: "active", + state + } + : undefined + ) + ).pipe( + Effect.tap((after) => + Effect.sync(() => { + if (logic.inspection?.kind === "Machine" || after === undefined) return + inspector.publishUnsafe({ + _tag: "StateChanged", + subject: subject!, + before: before.snapshot.state, + after: state, + causedByDeliveryId: inFlightMessage !== undefined && isAcknowledgedMessage(inFlightMessage) + ? inFlightMessage.inspection?.deliveryId + : undefined + }) + }) + ) + ) + ), + Effect.asVoid + ) + const terminalizeWith = ( snapshot: RuntimeSnapshot, exit: Exit.Exit, @@ -1426,6 +1716,22 @@ const startGenericInternal: < Effect.exit, Effect.asVoid ) + const closeEmissionsAndInspect = inspector === undefined + ? emissions.close() + : emissions.close().pipe( + Effect.andThen(Effect.sync(() => + inspector.publishUnsafe( + activity === undefined + ? { _tag: "Terminated", subject: subject!, snapshot } + : { + _tag: "ActivityStopped", + subject: activity.owner, + activity, + exit: snapshot.status === "stopped" ? Exit.interrupt() : exit + } + ) + )) + ) return Effect.uninterruptible( Effect.sync(() => { while (true) { @@ -1437,7 +1743,7 @@ const startGenericInternal: < Effect.andThen(Queue.shutdown(queue)), Effect.andThen(closeChildren(exit)), Effect.andThen(setAndPublishSnapshot(snapshot)), - Effect.andThen(emissions.close()), + Effect.andThen(closeEmissionsAndInspect), Effect.andThen(Effect.sync(() => { if (Exit.isFailure(exit)) { failAcknowledgedMessage(inFlightMessage, exit.cause) @@ -1448,6 +1754,7 @@ const startGenericInternal: < })), Effect.andThen(notifyOutcome), Effect.andThen(cleanup), + Effect.andThen(options.inspectionRoot === true && inspector !== undefined ? inspector.close : Effect.void), Effect.andThen(completeDone) ) ) @@ -1530,25 +1837,79 @@ const startGenericInternal: < inFlightMessage = undefined } } + const receive = inspector === undefined || logic.execution?._tag === "Compiled" + ? Queue.take(queue).pipe(Effect.map(messageEvent)) + : Queue.take(queue).pipe( + Effect.tap((message) => + Effect.sync(() => { + inFlightMessage = message + }) + ), + Effect.map(messageEvent) + ) + const poll = inspector === undefined || logic.execution?._tag === "Compiled" + ? Queue.poll(queue).pipe(Effect.map(Option.map(messageEvent))) + : Queue.poll(queue).pipe( + Effect.tap((message) => + Effect.sync(() => { + if (Option.isSome(message)) inFlightMessage = message.value + }) + ), + Effect.map(Option.map(messageEvent)) + ) + const updateStateDirect = (f: (state: State) => Effect.Effect) => + updateSnapshot((snapshot) => + snapshot.status === "active" + ? f(snapshot.state).pipe( + Effect.map((state) => ({ + status: "active" as const, + state + })) + ) + : Effect.succeed(undefined) + ).pipe(Effect.asVoid) + const updateState: ProcessContext["updateState"] = inspector === undefined + ? updateStateDirect + : (f) => + SynchronizedRef.get(current).pipe( + Effect.flatMap((before) => + updateSnapshot((snapshot) => + snapshot.status === "active" + ? f(snapshot.state).pipe( + Effect.map((state) => ({ + status: "active" as const, + state + })) + ) + : Effect.succeed(undefined) + ).pipe( + Effect.tap((after) => + Effect.sync(() => { + if (logic.inspection?.kind === "Machine" || after?.status !== "active") return + inspector.publishUnsafe({ + _tag: "StateChanged", + subject: subject!, + before: before.snapshot.state, + after: after.state, + causedByDeliveryId: inFlightMessage !== undefined && isAcknowledgedMessage(inFlightMessage) + ? inFlightMessage.inspection?.deliveryId + : undefined + }) + }) + ) + ) + ), + Effect.asVoid + ) const context: ProcessContext = { ...scope, ...(logic.execution?._tag === "Compiled" && !logic.execution.childless ? { ownedChildren } : undefined), ...acknowledgedContext, - receive: Queue.take(queue).pipe(Effect.map(messageEvent)), - poll: Queue.poll(queue).pipe(Effect.map(Option.map(messageEvent))), + receive, + poll, state: SynchronizedRef.get(current).pipe(Effect.map((current) => current.snapshot.state)), setState: setActiveState, - updateState: (f) => - updateSnapshot((snapshot) => - snapshot.status === "active" - ? f(snapshot.state).pipe( - Effect.map((state) => ({ - status: "active" as const, - state - })) - ) - : Effect.succeed(undefined) - ).pipe(Effect.asVoid) + updateState } const getOrCreateChanges = SynchronizedRef.modifyEffect( @@ -1591,6 +1952,9 @@ const startGenericInternal: < const ref: MachineRef = { id, sessionId, + ...(inspector === undefined + ? undefined + : { inspectionSubject: subject!, sendInspected: offerInspected! }), state: SynchronizedRef.get(current).pipe(Effect.map((current) => current.snapshot.state)), snapshot: SynchronizedRef.get(current).pipe(Effect.map((current) => current.snapshot)), changes: changesStream, @@ -1715,7 +2079,15 @@ type CompiledRunState = "Initializing" | "Idle" | "Draining" class CompiledProcess implements MachineRef { readonly id: string readonly sessionId: string - readonly send: (event: unknown) => Effect.Effect; + readonly inspectionSubject?: Inspection.Subject + readonly send: (event: unknown) => Effect.Effect + sendInspected( + event: unknown, + source: Inspection.Subject | undefined, + causedBy: Inspection.Causation | undefined + ): Effect.Effect { + return this.offerEvent(event, source, causedBy) + } [acknowledgedSend]( event: unknown ): Effect.Effect, unknown | StoppedError> { @@ -1750,6 +2122,10 @@ class CompiledProcess implements MachineRef { private inFlightMessage: ProcessMessage | undefined private readonly externalEmissions: EmissionRuntime | undefined private emissionsPubSub: LazyEmissions + private readonly activity?: Inspection.Activity + private initialEntryPaths?: ReadonlyArray + private initialMicrosteps?: ReadonlyArray + private initializing?: boolean constructor( private readonly logic: ProcessLogic, @@ -1759,26 +2135,106 @@ class CompiledProcess implements MachineRef { ) { this.sessionId = sessionId this.id = options.id ?? sessionId + const inspector = options.runtime.inspection + if (inspector !== undefined) { + this.inspectionSubject = inspectionSubject(logic, this.id, sessionId) + this.initializing = true + if (options.activity !== undefined) this.activity = { ...options.activity, sessionId } + } this.externalEmissions = options.emissions - this.send = (event) => this.offerMessage(event) + this.send = inspector === undefined + ? (event) => this.offerMessage(event) + : (event) => this.offerEvent(event, undefined, undefined) this.address = { id: this.id, sessionId, stop: Effect.suspend(() => this.stopFromProcess()), - send: this.send + send: this.send, + ...(inspector === undefined + ? undefined + : { + inspectionSubject: this.inspectionSubject!, + sendInspected: ( + event: unknown, + source: Inspection.Subject | undefined, + causedBy: Inspection.Causation | undefined + ) => this.offerEvent(event, source, causedBy) + }) } } + private get inspector(): InspectionRuntime.Runtime | undefined { + return this.options.runtime.inspection + } + + private causation(): Inspection.Causation | undefined { + return messageCausation(this.inFlightMessage, this.initializing === true) + } + + private publishCreated(): void { + const inspector = this.inspector + const subject = this.inspectionSubject + if (inspector === undefined || subject === undefined) return + inspector.publishUnsafe( + this.activity === undefined + ? { + _tag: "Created", + subject, + parent: this.options.parent?.inspectionSubject, + origin: this.options.origin ?? { _tag: "Root" }, + definition: this.logic.inspection?.definition + } + : { _tag: "ActivityStarted", subject: this.activity.owner, activity: this.activity } + ) + } + + private publishInitialized(state: unknown): void { + this.initializing = false + if (this.activity !== undefined) return + const inspector = this.inspector + const subject = this.inspectionSubject + if (inspector === undefined || subject === undefined) return + inspector.publishUnsafe({ + _tag: "Initialized", + subject, + snapshot: { status: "active", state }, + initialEntryPaths: this.initialEntryPaths ?? [], + microsteps: InspectionRuntime.microsteps({ microsteps: this.initialMicrosteps ?? [] }) + }) + } + + private publishStartFailed(cause: Cause.Cause): void { + this.initializing = false + const inspector = this.inspector + const subject = this.inspectionSubject + if (inspector === undefined || subject === undefined) return + inspector.publishUnsafe( + this.activity === undefined + ? { _tag: "StartFailed", subject, cause } + : { + _tag: "ActivityStopped", + subject: this.activity.owner, + activity: this.activity, + exit: Exit.failCause(cause) + } + ) + } + private get execution(): CompiledProcessExecution { return this.logic.execution as CompiledProcessExecution } initializeCompiledSync(): Effect.Effect, unknown> { + this.publishCreated() if (!this.execution.childless) { this.childRuntime = makeChildRuntimeSync(this.address, this.options.runtime, this.services) } const parent = this.options.parent - const sendParent = this.options.sendParent ?? (parent === undefined ? noParentSend : parent.send) + const sendParent = this.options.sendParent ?? (parent === undefined + ? noParentSend + : this.inspector === undefined + ? parent.send + : (event: unknown) => sendMachineTarget(parent, event, this.inspectionSubject, this.causation())) this.processScope = { self: this.address, parent, @@ -1786,11 +2242,24 @@ class CompiledProcess implements MachineRef { sendParent, 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"], + isMachineTarget(target) + ? this.inspector === undefined + ? target.send(event) + : sendMachineTarget(target, event, this.inspectionSubject, this.causation()) + : this.childRuntime.sendTo( + target as ChildSelector, + event, + this.inspectionSubject, + this.causation() + )) as ProcessScope["sendTo"], stopChild: this.childRuntime.stop, - failCause: (cause: Cause.Cause) => this.failCause(cause) + failCause: (cause: Cause.Cause) => this.failCause(cause), + inspectInitial: this.inspector === undefined + ? noInspectInitial + : (paths, microsteps = []) => { + this.initialEntryPaths = paths + this.initialMicrosteps = microsteps + } } const compiledInitial = this.execution.initialSync! let initialized: CompiledInitialized @@ -1798,6 +2267,7 @@ class CompiledProcess implements MachineRef { initialized = compiledInitial(this.processScope) } catch (error) { this.runState = "Idle" + this.publishStartFailed(Cause.fail(error)) return Effect.fail(error) } this.runState = "Idle" @@ -1807,6 +2277,7 @@ class CompiledProcess implements MachineRef { changes: undefined, snapshot: { status: "active", state: initialized.state } } + this.publishInitialized(initialized.state) this.compiledContext = new CompiledProcessContextImpl(this.processScope, this.childRuntime.owned, this) if ("executionState" in initialized) { this.compiledContext.executionState = initialized.executionState @@ -1831,11 +2302,16 @@ class CompiledProcess implements MachineRef { initialize(): Effect.Effect, unknown, any> { const self = this return Effect.gen(function*() { + self.publishCreated() if (!self.execution.childless) { self.childRuntime = yield* makeChildRuntime(self.address, self.options.runtime, self.services) } const parent = self.options.parent - const sendParent = self.options.sendParent ?? (parent === undefined ? noParentSend : parent.send) + const sendParent = self.options.sendParent ?? (parent === undefined + ? noParentSend + : self.inspector === undefined + ? parent.send + : (event: unknown) => sendMachineTarget(parent, event, self.inspectionSubject, self.causation())) self.processScope = { self: self.address, parent, @@ -1843,11 +2319,24 @@ class CompiledProcess implements MachineRef { sendParent, 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"], + isMachineTarget(target) + ? self.inspector === undefined + ? target.send(event) + : sendMachineTarget(target, event, self.inspectionSubject, self.causation()) + : self.childRuntime.sendTo( + target as ChildSelector, + event, + self.inspectionSubject, + self.causation() + )) as ProcessScope["sendTo"], stopChild: self.childRuntime.stop, - failCause: (cause: Cause.Cause) => self.failCause(cause) + failCause: (cause: Cause.Cause) => self.failCause(cause), + inspectInitial: self.inspector === undefined + ? noInspectInitial + : (paths, microsteps = []) => { + self.initialEntryPaths = paths + self.initialMicrosteps = microsteps + } } const cleanupStartupFailure = (exit: Exit.Exit): Effect.Effect => @@ -1867,7 +2356,10 @@ class CompiledProcess implements MachineRef { ) : compiledInitial(self.processScope) const initialized = yield* initializeEffect.pipe( - Effect.onExit(cleanupStartupFailure), + Effect.onExit((exit) => { + if (Exit.isFailure(exit)) self.publishStartFailed(exit.cause) + return cleanupStartupFailure(exit) + }), Effect.ensuring(Effect.sync(() => { self.runState = "Idle" })) @@ -1879,6 +2371,7 @@ class CompiledProcess implements MachineRef { changes: undefined, snapshot: { status: "active", state: initial } } + self.publishInitialized(initial) if (self.execution.drain._tag === "Process") { self.processContext = { ...self.processScope, @@ -2022,11 +2515,7 @@ class CompiledProcess implements MachineRef { return Effect.uninterruptibleMask((restore) => Deferred.make, unknown>().pipe( Effect.flatMap((deferred) => - this.offerMessage({ - [AcknowledgedMessageTypeId]: true as const, - event, - deferred - }).pipe( + this.offerEvent(event, undefined, undefined, deferred).pipe( Effect.andThen(restore(Deferred.await(deferred))) ) ) @@ -2034,6 +2523,31 @@ class CompiledProcess implements MachineRef { ) } + private offerEvent( + event: unknown, + source: Inspection.Subject | undefined, + causedBy: Inspection.Causation | undefined, + deferred?: Deferred.Deferred, unknown> + ): Effect.Effect { + return Effect.suspend(() => { + const inspector = this.inspector + const subject = this.inspectionSubject + const message: ProcessMessage = inspector?.isActive() === true && subject !== undefined + ? makeInspectedMessage( + inspector, + subject, + event, + source, + causedBy, + deferred + ) + : deferred === undefined + ? event + : { [AcknowledgedMessageTypeId]: true as const, event, deferred } + return this.offerMessage(message) + }) + } + private offerMessage(message: ProcessMessage): Effect.Effect { return Effect.uninterruptible( Effect.suspend(() => { @@ -2041,6 +2555,9 @@ class CompiledProcess implements MachineRef { return Effect.fail(new StoppedError()) } offerCompactMailbox(this.mailbox, message) + const inspector = this.inspector + const subject = this.inspectionSubject + if (inspector !== undefined && subject !== undefined) publishInspectedSent(inspector, subject, message) this.offerRevision += 1 if (this.runState === "Draining") { return Effect.void @@ -2096,6 +2613,7 @@ class CompiledProcess implements MachineRef { this.lifecycle !== "Active" || (this.options.onOutcome !== undefined && this.options.skipStoppedOutcome !== true) || this.options.onStop !== undefined || + this.inspector !== undefined || this.current.changes !== undefined || this.current.terminalizing || this.current.snapshot.status !== "active" ) { @@ -2245,13 +2763,31 @@ class CompiledProcess implements MachineRef { Effect.exit, Effect.asVoid ) + const inspector = this.inspector + const subject = this.inspectionSubject + const closeEmissionsAndInspect = inspector === undefined || subject === undefined + ? this.closeEmissions() + : this.closeEmissions().pipe( + Effect.andThen(Effect.sync(() => + inspector.publishUnsafe( + this.activity === undefined + ? { _tag: "Terminated", subject, snapshot } + : { + _tag: "ActivityStopped", + subject: this.activity.owner, + activity: this.activity, + exit: snapshot.status === "stopped" ? Exit.interrupt() : exit + } + ) + )) + ) return Effect.uninterruptible( Effect.sync(() => { closeCompactMailbox(this.mailbox) }).pipe( Effect.andThen(this.childRuntime.close(exit)), Effect.andThen(this.setAndPublishSnapshot(snapshot)), - Effect.andThen(this.closeEmissions()), + Effect.andThen(closeEmissionsAndInspect), Effect.andThen(Effect.sync(() => { if (requested._tag === "Failure") { failAcknowledgedMessage(this.inFlightMessage, requested.cause) @@ -2262,6 +2798,11 @@ class CompiledProcess implements MachineRef { })), Effect.andThen(notifyOutcome), Effect.andThen(this.options.onStop ?? Effect.void), + Effect.andThen( + this.options.inspectionRoot === true && this.inspector !== undefined + ? this.inspector.close + : Effect.void + ), Effect.andThen(Effect.sync(() => { this.options.onStopSync?.() this.runState = "Idle" @@ -2301,12 +2842,26 @@ class CompiledProcess implements MachineRef { } private emitEvent(event: unknown): Effect.Effect { - if (this.externalEmissions !== undefined) return this.externalEmissions.emit(event) - return Effect.suspend(() => - this.emissionsPubSub === undefined || this.emissionsPubSub === EmissionsClosed - ? Effect.void - : PubSub.publish(this.emissionsPubSub, event).pipe(Effect.asVoid) - ) + const publish = this.externalEmissions !== undefined ? + this.externalEmissions.emit(event) : + Effect.suspend(() => + this.emissionsPubSub === undefined || this.emissionsPubSub === EmissionsClosed + ? Effect.void + : PubSub.publish(this.emissionsPubSub, event).pipe(Effect.asVoid) + ) + const inspector = this.inspector + const subject = this.inspectionSubject + if (inspector === undefined || subject === undefined) return publish + return publish.pipe(Effect.tap(() => + Effect.sync(() => + inspector.publishUnsafe({ + _tag: "Emitted", + subject, + emission: event, + causedBy: this.causation() + }) + ) + )) } shutdownEmissions(): Effect.Effect { @@ -2570,7 +3125,17 @@ const startCompactCompiledInternal: typeof startGenericInternal = Effect.fnUntra ? process.initializeCompiledSync() : process.initialize() return yield* initialize.pipe( - Effect.onExit((exit) => Exit.isFailure(exit) ? process.shutdownEmissions() : Effect.void) + Effect.onExit((exit) => + Exit.isFailure(exit) + ? process.shutdownEmissions().pipe( + Effect.andThen( + options.inspectionRoot === true && options.runtime.inspection !== undefined + ? options.runtime.inspection.close + : Effect.void + ) + ) + : Effect.void + ) ) }) as typeof startGenericInternal @@ -2689,6 +3254,8 @@ const prepareProcessWithStrategy = Effect.fnUntraced(function*< ) { const runtime = yield* makeProcessRuntime const sessionId = yield* runtime.nextSessionId + const inspection = yield* InspectionRuntime.make(sessionId) + runtime.inspection = inspection const emissions = makeEmissionRuntime() const started = yield* Deferred.make, InitialError>() const internalOptions: StartInternalOptions = options === undefined @@ -2696,14 +3263,18 @@ const prepareProcessWithStrategy = Effect.fnUntraced(function*< detached: true, emissions, runtime, - sessionId + sessionId, + inspectionRoot: true, + origin: { _tag: "Root" } } : { ...options, detached: true, emissions, runtime, - sessionId + sessionId, + inspectionRoot: true, + origin: { _tag: "Root" } } const initialize = strategy === "generic" ? startGenericInternal(logic, internalOptions) @@ -2724,6 +3295,7 @@ const prepareProcessWithStrategy = Effect.fnUntraced(function*< Deferred.await(started).pipe(Effect.map((ref) => ref.changes)) ), emissions: emissions.stream as Stream.Stream, + inspection: inspection.stream, start } }) diff --git a/src/unstable/reactivity/AtomMachine.ts b/src/unstable/reactivity/AtomMachine.ts index 99026db..2738d96 100644 --- a/src/unstable/reactivity/AtomMachine.ts +++ b/src/unstable/reactivity/AtomMachine.ts @@ -179,6 +179,18 @@ export const emissions: ( self: MachineAtom ) => Stream.Stream = internal.emissions +/** + * Observes the ordered local inspection records for the machine atom's root + * ownership tree. The stream starts before machine initialization, is hot and + * non-replayed, and completes with the root machine. + * + * @category getters + * @since 0.13.0 + */ +export const inspection: ( + self: MachineAtom +) => Stream.Stream = internal.inspection + /** * Observes emissions from each active instance selected by a child bridge. * diff --git a/test/internal/machine/strategyDifferential.test.ts b/test/internal/machine/strategyDifferential.test.ts index 812977e..b6c2933 100644 --- a/test/internal/machine/strategyDifferential.test.ts +++ b/test/internal/machine/strategyDifferential.test.ts @@ -73,7 +73,7 @@ describe("machine planner and runtime strategies", () => { assert.deepStrictEqual(planned.microsteps[0]?.entryPaths, ["Count"]) })) - it.effect("keeps indexed execution microsteps narrower than diagnostic planner microsteps", () => + it.effect("retains indexed execution microstep evidence without widening frozen execution values", () => Effect.gen(function*() { const machine = makeFlatMachine() const initial = yield* Machine.planInitial(machine) @@ -422,6 +422,30 @@ describe("machine planner and runtime strategies", () => { } }) as Effect.Effect) + it.effect("publishes equivalent live inspection records from generic and compiled runtimes", () => + Effect.scoped(Effect.gen(function*() { + const machine = makeFlatMachine() + const results: Array> = [] + + for (const strategy of ["generic", "compiled"] as const) { + const prepared = yield* prepareWithRuntimeStrategy(machine, strategy) + const observed = yield* prepared.inspection.pipe( + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }) + ) + yield* Effect.yieldNow + const ref = yield* prepared.start + for (const event of [new Noop({}), new Increment({}), new Reenter({}), new Finish({})]) { + yield* ref.send(event) + yield* Effect.yieldNow + } + yield* ref.join + results.push(Array.from(yield* Fiber.join(observed))) + } + + assert.deepStrictEqual(results[0], results[1]) + }) as Effect.Effect)) + it.effect("matches acknowledged probe delivery in generic and compiled managed runtimes", () => Effect.gen(function*() { const machine = makeFlatMachine() diff --git a/test/internal/machine/support/strategyDifferential.ts b/test/internal/machine/support/strategyDifferential.ts index 591b442..37894ff 100644 --- a/test/internal/machine/support/strategyDifferential.ts +++ b/test/internal/machine/support/strategyDifferential.ts @@ -123,6 +123,7 @@ export const prepareWithRuntimeStrategy = ( ): Effect.Effect< { readonly emissions: import("effect/Stream").Stream + readonly inspection: import("effect/Stream").Stream readonly start: Effect.Effect, unknown> }, unknown diff --git a/test/machine/LiveInspection.test.ts b/test/machine/LiveInspection.test.ts new file mode 100644 index 0000000..624f852 --- /dev/null +++ b/test/machine/LiveInspection.test.ts @@ -0,0 +1,238 @@ +import { assert, describe, it } from "@effect/vitest" +import { Cause, Effect, Exit, Fiber, Option, Schema, Stream } from "effect" +import { Machine } from "../../src/index.js" + +class Idle extends Schema.TaggedClass("LiveInspectionIdle")("Idle", {}) {} +class Increment extends Schema.TaggedClass("LiveInspectionIncrement")("Increment", { + by: Schema.Number +}) {} +class Notice extends Schema.TaggedClass("LiveInspectionNotice")("Notice", { + value: Schema.Number +}) {} + +const states = Machine.defineStates({ Idle }) +const Events = Machine.events(Increment) +const Emissions = Machine.emittedEvents(Notice) + +const machine = Machine.make({ + id: "counter", + states: states.states, + events: Events, + emittedEvents: Emissions, + initial: () => states.initial.Idle(new Idle({})) +}).handle({ + Idle: { + on: { + Increment: ({ event, target }, enqueue) => { + enqueue.emit(Emissions.Notice({ value: event.by })) + return target.full.Idle(new Idle({})) + } + } + } +}) + +describe("Machine live inspection", () => { + it.effect("observes a prepared root from creation through termination", () => + Effect.scoped(Effect.gen(function*() { + const prepared = yield* Machine.prepare(machine) + const collected = yield* prepared.inspection.pipe( + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }) + ) + yield* Effect.yieldNow + + const ref = yield* prepared.start + yield* ref.send(Events.Increment({ by: 2 })) + yield* Effect.yieldNow + yield* ref.stop + + const records = Array.from(yield* Fiber.join(collected)) + assert.deepStrictEqual(records.map(({ _tag }) => _tag), [ + "Created", + "Initialized", + "EventSent", + "Emitted", + "EventProcessed", + "Terminated" + ]) + assert.deepStrictEqual(records.map(({ sequence }) => sequence), [0, 1, 2, 3, 4, 5]) + assert.ok(records.every(({ rootSessionId }) => rootSessionId === prepared.sessionId)) + + const created = records[0] + assert.strictEqual(created?._tag, "Created") + if (created?._tag === "Created") { + assert.deepStrictEqual(created.subject, { + id: "counter", + sessionId: prepared.sessionId, + kind: "Machine" + }) + assert.deepStrictEqual(created.origin, { _tag: "Root" }) + assert.strictEqual(created.parent, undefined) + assert.strictEqual(created.definition, machine) + } + + const processed = records.find((record) => record._tag === "EventProcessed") + assert.ok(processed !== undefined && processed._tag === "EventProcessed") + if (processed?._tag === "EventProcessed") { + assert.strictEqual(processed.handled, true) + assert.strictEqual(processed.configurationChanged, false) + assert.strictEqual(processed.microsteps.length, 1) + assert.deepStrictEqual(processed.microsteps[0]?.transitions, [{ + source: "Idle", + trigger: { type: "event", event: "Increment" }, + reenter: false, + target: "Idle", + resolvedTarget: "Idle" + }]) + } + + const emitted = records.find((record) => record._tag === "Emitted") + assert.ok(emitted !== undefined && emitted._tag === "Emitted") + if (emitted?._tag === "Emitted") { + assert.deepStrictEqual(emitted.emission, new Notice({ value: 2 })) + assert.deepStrictEqual(emitted.causedBy, { _tag: "Macrostep", macrostepId: 0 }) + } + }))) + + it.effect("is hot, non-replayed, and completes when startup fails", () => + Effect.scoped(Effect.gen(function*() { + const invalid = Machine.make({ + states: states.states, + events: Machine.events(), + initial: () => { + throw new Error("boom") + } + }).handle({ Idle: {} }) + const prepared = yield* Machine.prepare(invalid) + const collected = yield* prepared.inspection.pipe( + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }) + ) + yield* Effect.yieldNow + yield* Effect.exit(prepared.start) + + assert.deepStrictEqual(Array.from(yield* Fiber.join(collected)).map(({ _tag }) => _tag), [ + "Created", + "StartFailed" + ]) + assert.deepStrictEqual(Array.from(yield* Stream.runCollect(prepared.inspection)), []) + }))) + + it.effect("represents Effect invokes as owned activities rather than child machines", () => + Effect.scoped(Effect.gen(function*() { + const active = Machine.make({ + id: "activity-root", + states: states.states, + events: Machine.events(), + initial: () => states.initial.Idle(new Idle({})) + }).handle({ + Idle: { + invoke: Machine.invoke({ id: "worker", effect: Effect.never }) + } + }) + const prepared = yield* Machine.prepare(active) + const collected = yield* prepared.inspection.pipe( + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }) + ) + yield* Effect.yieldNow + const ref = yield* prepared.start + yield* Effect.yieldNow + yield* ref.stop + + const records = Array.from(yield* Fiber.join(collected)) + assert.deepStrictEqual(records.map(({ _tag }) => _tag), [ + "Created", + "Initialized", + "ActivityStarted", + "ActivityStopped", + "Terminated" + ]) + const started = records.find((record) => record._tag === "ActivityStarted") + assert.ok(started !== undefined && started._tag === "ActivityStarted") + if (started?._tag === "ActivityStarted") { + assert.deepStrictEqual(started.activity, { + id: "worker", + sessionId: "machine:1", + owner: started.subject, + ownerPath: "Idle", + kind: "Effect" + }) + } + const stopped = records.find((record) => record._tag === "ActivityStopped") + assert.ok(stopped !== undefined && stopped._tag === "ActivityStopped") + if (stopped?._tag === "ActivityStopped") { + assert.ok(Exit.isFailure(stopped.exit)) + if (Exit.isFailure(stopped.exit)) assert.ok(Cause.hasInterruptsOnly(stopped.exit.cause)) + } + }))) + + it.effect("correlates an explicit child-to-parent send with both local subjects", () => + Effect.scoped(Effect.gen(function*() { + class ChildIdle extends Schema.TaggedClass("InspectionChildIdle")("ChildIdle", {}) {} + class Trigger extends Schema.TaggedClass("InspectionChildTrigger")("Trigger", {}) {} + class ChildReady extends Schema.TaggedClass("InspectionChildReady")("ChildReady", {}) {} + class ParentIdle extends Schema.TaggedClass("InspectionParentIdle")("ParentIdle", {}) {} + class ParentDone extends Schema.TaggedClass("InspectionParentDone")("ParentDone", {}) {} + + const ParentEvents = Machine.events(ChildReady) + const ChildEvents = Machine.events(Trigger) + const childStates = Machine.defineStates({ ChildIdle }) + const childMachine = Machine.make({ + id: "child-machine", + states: childStates.states, + events: ChildEvents, + parentEvents: ParentEvents, + initial: () => childStates.initial.ChildIdle(new ChildIdle({})) + }).handle({ + ChildIdle: { + on: { + Trigger: ({ parent, target }, enqueue) => { + if (parent !== undefined) enqueue.sendTo(parent, ParentEvents.ChildReady()) + return target.none() + } + } + } + }) + const Child = Machine.child("child", childMachine) + const parentStates = Machine.defineStates({ + ParentIdle, + ParentDone: { schema: ParentDone, type: "final" } + }) + const parentMachine = Machine.make({ + id: "parent-machine", + states: parentStates.states, + events: Machine.events(ParentEvents), + initial: () => parentStates.initial.ParentIdle(new ParentIdle({})) + }).handle({ + ParentIdle: { + invoke: Machine.invoke({ child: Child }), + on: { + ChildReady: ({ target }) => target.full.ParentDone(new ParentDone({})) + } + }, + ParentDone: {} + }) + + const prepared = yield* Machine.prepare(parentMachine) + const collected = yield* prepared.inspection.pipe( + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }) + ) + yield* Effect.yieldNow + const parent = yield* prepared.start + yield* Effect.yieldNow + const child = Option.getOrThrow(yield* parent.child(Child)) + yield* child.send(ChildEvents.Trigger()) + yield* parent.join + + const sent = Array.from(yield* Fiber.join(collected)).filter((record) => record._tag === "EventSent") + const toParent = sent.find((record) => record.subject.id === "parent-machine") + assert.ok(toParent !== undefined && toParent._tag === "EventSent") + if (toParent?._tag === "EventSent") { + assert.strictEqual(toParent.source?.id, "child") + assert.strictEqual(toParent.target.id, "parent-machine") + assert.deepStrictEqual(toParent.causedBy, { _tag: "Macrostep", macrostepId: 0 }) + } + }))) +}) diff --git a/test/unstable/reactivity/AtomMachine.test.ts b/test/unstable/reactivity/AtomMachine.test.ts index a6fac47..5472eaa 100644 --- a/test/unstable/reactivity/AtomMachine.test.ts +++ b/test/unstable/reactivity/AtomMachine.test.ts @@ -84,6 +84,22 @@ const makeCounterMachine = () => }) describe("AtomMachine", () => { + it.effect("observes prepared live inspection before atom startup", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const bridge = AtomMachine.make(makeCounterMachine()) + const observed = yield* AtomMachine.inspection(bridge).pipe( + Stream.take(2), + Stream.runCollect, + Effect.provideService(AtomRegistry.AtomRegistry, registry), + Effect.forkScoped({ startImmediately: true }) + ) + + const records = Array.from(yield* Fiber.join(observed)) + assert.deepStrictEqual(records.map(({ _tag }) => _tag), ["Created", "Initialized"]) + assert.ok(records.every(({ rootSessionId }) => rootSessionId === "machine:0")) + }))) + it.effect("observes initial emissions when the emission stream starts the machine", () => Effect.scoped(Effect.gen(function*() { class Idle extends Schema.TaggedClass("AtomPreparedIdle")("Idle", {}) {} diff --git a/typetest/machine/Inspection.tst.ts b/typetest/machine/Inspection.tst.ts index 3d0dae6..c9a164f 100644 --- a/typetest/machine/Inspection.tst.ts +++ b/typetest/machine/Inspection.tst.ts @@ -1,4 +1,4 @@ -import { Effect, Schema } from "effect" +import { Effect, Schema, Stream } from "effect" import { describe, expect, it } from "tstyche" import { Machine } from "../../src/index.js" @@ -31,6 +31,30 @@ describe("Machine inspection", () => { } }) + it("exposes one closed operational protocol from prepared machines", () => { + const FlatStates = Machine.defineStates({ Idle }) + const executable = Machine.make({ + states: FlatStates.states, + events: Machine.events(Reset), + initial: () => FlatStates.initial.Idle(new Idle({})) + }).handle({ Idle: { on: { Reset: ({ target }) => target.none() } } }) + Effect.gen(function*() { + const prepared = yield* Machine.prepare(executable) + expect(prepared.inspection).type.toBe>() + const event = yield* prepared.inspection.pipe(Stream.runHead) + if (event._tag === "Some") { + switch (event.value._tag) { + case "Created": + expect(event.value.definition).type.toBe() + break + case "EventProcessed": + expect(event.value.microsteps).type.toBe>() + break + } + } + }) + }) + it("preserves state paths for structural inspection", () => { const nodes = Machine.stateNodes(machine) expect(nodes[0]!.path).type.toBe<"root" | "root.idle" | "root.recent">()