Add live machine inspection

Author: SandroMaglione

.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.

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.
 

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.

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<State, Error = never, Output = never> =
     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<Machine.RetainedTransition>
+    readonly raisedEvents: ReadonlyArray<unknown>
+    readonly emittedEvents: ReadonlyArray<unknown>
+    readonly commands: ReadonlyArray<Command>
+    readonly exitPaths: ReadonlyArray<string>
+    readonly entryPaths: ReadonlyArray<string>
+    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<unknown, unknown, unknown>
+    readonly initialEntryPaths: ReadonlyArray<string>
+    readonly microsteps: ReadonlyArray<Microstep>
+  }
+
+  /** Announces failure before an initial runtime snapshot exists. */
+  export interface StartFailed extends Base {
+    readonly _tag: "StartFailed"
+    readonly cause: Cause.Cause<unknown>
+  }
+
+  /** 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<unknown, unknown, unknown>
+    readonly after: RuntimeSnapshot<unknown, unknown, unknown>
+    readonly handled: boolean
+    readonly configurationChanged: boolean
+    readonly microsteps: ReadonlyArray<Microstep>
+  }
+
+  /** 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<unknown, unknown>
+  }
+
+  /** Announces a terminal local runtime snapshot. */
+  export interface Terminated extends Base {
+    readonly _tag: "Terminated"
+    readonly snapshot: RuntimeSnapshot<unknown, unknown, unknown>
+  }
+
+  /** 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<out State, in Event, out Error, out Output, out Emitte
   /** Streams ephemeral notifications published after subscription. */
   readonly emissions: Stream.Stream<Emitted>
 
+  /**
+   * 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<Inspection.Event>
+
   /** Initializes this machine once and returns its running reference. */
   readonly start: Effect.Effect<MachineRef<State, Event, Error, Output, Emitted>, StartError, StartRequirements>
 }

src/internal/machine/atom.ts

@@ -139,6 +139,32 @@ export const emissions = <State, Event, Error, Output, StartError, Emitted>(
   )
 }
 
+export const inspection = <State, Event, Error, Output, StartError, Emitted>(
+  self: MachineAtom<State, Event, Error, Output, StartError, Emitted>
+): Stream.Stream<Machine.Inspection.Event, StartError, AtomRegistry.AtomRegistry> => {
+  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 = <Child extends Machine.ChildMachine.Any, StartError>(
   self: ChildMachineAtom<Child, StartError>
 ): Stream.Stream<RefEmitted<Machine.ChildMachine.Ref<Child>>, StartError, AtomRegistry.AtomRegistry> =>