Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions .changeset/live-machine-inspection.md
Original file line number Diff line number Diff line change
@@ -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.
50 changes: 50 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
50 changes: 50 additions & 0 deletions docs/agent-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:

Expand Down Expand Up @@ -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.
Expand Down
190 changes: 190 additions & 0 deletions src/Machine.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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.
*
Expand Down Expand Up @@ -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>
}
Expand Down
26 changes: 26 additions & 0 deletions src/internal/machine/atom.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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> =>
Expand Down
Loading