typeonce-dev · SandroMaglione · Sep 11, 2026 · Sep 10, 2026
diff --git a/.changeset/quiet-machines-wait.md b/.changeset/quiet-machines-wait.md
@@ -0,0 +1,9 @@
+---
+"@typeonce/effect-machine": minor
+---
+
+Add `Machine.waitFor(ref, predicate)` for external Effects and tests awaiting a current or subsequent published snapshot. Type predicates narrow the result. Unmatched failures preserve their Cause, stopping fails with `StoppedError`, and completion without a match fails with `Cause.NoSuchElementError`. Compose `Effect.timeout` to bound the wait; cancellation releases observation without stopping the machine.
+
+Effect, Stream, and timer invocations now run inside `Machine.invoke` spans with machine, state, source, and invocation identity. Use ordinary Effect tracing configuration to export them. This preserves resource ownership and does not propagate each sender's trace context through the mailbox.
+
+Fix indexed self-transitions and reentry for states without a value schema, so their invocations restart consistently with the generic runtime.
@@ -397,6 +397,43 @@ address it from Effects; resolvers use `enqueue.sendTo` and `enqueue.stop`.
 Duplicate active IDs fail with `ChildAlreadyExistsError` and preserve the
 existing child. Parent-protocol compatibility is checked at each spawn.
 
+## External observation and tracing
+
+Use `Machine.waitFor(ref, predicate)` when an external Effect needs a particular
+current or subsequent published snapshot. Type predicates narrow the result.
+It does not send an event or prove that a particular event caused the snapshot.
+Prefer `ref.join` for final output, invocation `onDone` / `onFailure` for workflow
+behavior, atom selectors for UI reads, and `ref.changes` for continuous observation.
+
+```ts
+const ready = yield * Machine.waitFor(ref, isReady).pipe(Effect.timeout("5 seconds"))
+```
+
+The predicate runs first, including for terminal snapshots. An unmatched failure
+preserves its `Cause`; an unmatched stop fails with `StoppedError`; completion
+without a match fails with `Cause.NoSuchElementError`. A predicate exception is
+a defect. Observation is lazy, has no default timeout, and releases its
+subscription on completion or interruption without stopping the machine. It does
+not replay historical snapshots or expose intermediate macrostep configurations.
+
+Effect, Stream, and timer invocations run inside an Effect span named
+`Machine.invoke`. Application spans inside that work inherit the invocation span.
+Configure the application's ordinary Effect tracer and exporter; no Machine-specific
+exporter or inspection subscription is required. Standard Effect tracing and
+sampling controls apply, including `Effect.withTracerEnabled(false)`.
+
+Spans identify `machine.id`, `machine.sessionId`, `machine.state.path`,
+`machine.invoke.id`, `machine.invoke.source`, `machine.invoke.kind`, and
+`machine.invoke.sessionId`. Source is the registered name; invocation IDs can be
+overridden independently. Payloads are not recorded automatically. Execution spans
+end when the Effect, Stream consumption, or timer settles, including cancellation;
+they do not change the lifetime of resources held in an enclosing Scope. Source
+construction keeps its existing startup failure boundary. This does not add
+machine-lifetime spans or propagate the trace context of each sender through the
+mailbox. Tracing can add diagnostic Cause annotations without changing failure
+values or interruption semantics. Even without an exporter, tracing has runtime
+cost.
+
 ## Reactivity
 
 `AtomMachine` runs one lazy machine instance per `AtomRegistry`:

@@ -474,3 +474,23 @@ const testProgram = Effect.gen(function*() {
 Use pure planner traces for state and transition rules. Start a live machine
 and use `MachineTest.probe` when a test depends on timers, invoked work, raised
 events, or runtime scheduling.
+
+## Choose observation deliberately
+
+Use atom selectors to render state, invocation outcomes to express workflow
+behavior, `ref.join` to await final output, and `ref.changes` for ongoing
+observation. Reach for `Machine.waitFor(ref, predicate)` only when an external
+Effect or test must wait for a specific current or subsequent snapshot, such as
+a long-lived connection becoming ready. Compose `Effect.timeout` when needed.
+
+Do not use `send` followed by `waitFor` as an acknowledgement protocol. The current
+snapshot might already match, and a match does not identify which event caused
+it. Keep workflow steps in the machine. A matching terminal snapshot succeeds;
+an unmatched error preserves its Cause, stopping fails with `StoppedError`, and
+completion without a match fails with `Cause.NoSuchElementError`.
+
+Invoked Effects, Streams, and timers already receive a `Machine.invoke` Effect
+span with machine, state, source, and invocation identity. Use ordinary Effect
+tracing configuration and application spans such as `Effect.fn("Payments.charge")`.
+Do not add manual spans around every transition or introduce a second exporter.
+Sender trace context is not automatically carried through machine mailboxes.
@@ -14,6 +14,7 @@ The command reports:
 - repeated child lookup and delivery to one running child;
 - machine start-and-stop throughput;
 - parent-with-child start-and-stop throughput;
+- complete Effect invocation throughput with standard tracing enabled and disabled;
 - generic and compiled raw-process lifecycle throughput;
 - heap and resident-memory growth at 100, 500, and 1,000 live units, including
   a raw generic process, a raw compiled process, an idle statechart, two

@@ -26,6 +26,22 @@ if (machineRuntimePath === undefined) {
 const machineRuntime = await import(pathToFileURL(machineRuntimePath).href)
 const { Effect, Fiber, Option, Schema, Stream } = effect
 
+const invocationLifecycleMachine = benchmarkApi.effectLifecycle?.(Effect.void, Schema.Void)
+const invocationLifecycleBenchmarks = invocationLifecycleMachine === undefined ? [] : [true, false].map((enabled) => ({
+  id: `effect-invocation-tracing-${enabled ? "enabled" : "disabled"}`,
+  label: `Complete an Effect invocation with tracing ${enabled ? "enabled" : "disabled"}`,
+  unit: "invocations/s",
+  operations: () => 1,
+  expected: () => 1,
+  start: () => undefined,
+  run: () => Effect.runPromise(Effect.gen(function*() {
+    const ref = yield* Machine.start(invocationLifecycleMachine)
+    yield* ref.join
+    return 1
+  }).pipe(Effect.withTracerEnabled(enabled))),
+  stop: () => undefined
+}))
+
 const CounterState = Schema.TaggedUnion({
   Count: {
     value: Schema.Number
@@ -616,6 +632,7 @@ export const effectMachineAdapter = {
   stopObservedCounter,
   stopCounters,
   additionalMachineBenchmarks: [
+    ...invocationLifecycleBenchmarks,
     {
       id: "hierarchical-plan-counter",
       label: "Plan transitions through a compound state",

@@ -65,6 +65,17 @@ export const makeEffectMachineBenchmarkApi = (Machine) => {
 
   return {
     snapshot,
+    effectLifecycle: hasHandlerInitial ? (work, output) => {
+      const root = Machine.state({ states: { Working: {}, Complete: { type: "final", output } } })
+      const targets = Machine.targets(root)
+      return Machine.make({ root, events: Machine.events({}), effects: { work } }).handle({
+        initial: { target: targets.root.Working },
+        states: {
+          Working: { invoke: { src: "work", onDone: { target: targets.root.Complete } } },
+          Complete: { output: () => undefined }
+        }
+      })
+    } : undefined,
     make: (config) => {
       if (!hasRoot) return Machine.make(config)
       const { states: root, initial, ...rest } = config

@@ -9186,6 +9186,53 @@ export const watch: <State, Event, Error = never, Output = never>(
   ref: MachineRef<State, Event, Error, Output>
 ) => Stream.Stream<RuntimeOutcome<State, Error, Output>> = internal.watch
 
+/**
+ * Waits for the first current or subsequent published snapshot matching a predicate.
+ *
+ * Use for external Effect coordination and tests. Prefer atom selectors for UI
+ * reads, invocation outcomes for workflow behavior, and `ref.join` for output.
+ * This observes state; it does not acknowledge or correlate a sent event.
+ *
+ * The predicate runs before terminal classification, so explicitly matching a
+ * done, error, or stopped snapshot succeeds. Otherwise an error preserves its
+ * Cause, stopping fails with `StoppedError`, and completion without a match
+ * fails with `Cause.NoSuchElementError`. Predicate exceptions are defects.
+ *
+ * Evaluation subscribes lazily. Interruption releases only the subscription,
+ * never the machine. There is no default timeout; compose `Effect.timeout`.
+ * Type predicates narrow the returned snapshot. Intermediate microsteps and
+ * historical snapshots are not observed.
+ *
+ * @example
+ * ```ts
+ * const snapshot = yield* Machine.waitFor(ref, isReady).pipe(Effect.timeout("5 seconds"))
+ * ```
+ * @category combinators
+ * @since 0.37.0
+ */
+export const waitFor: {
+  <State, Error, Output, Narrowed extends RuntimeSnapshot<State, Error, Output>>(
+    predicate: (snapshot: RuntimeSnapshot<State, Error, Output>) => snapshot is Narrowed
+  ): <Event, Emitted>(ref: MachineRef<State, Event, Error, Output, Emitted>) => Effect.Effect<
+    Narrowed,
+    Error | StoppedError | Cause.NoSuchElementError
+  >
+  <State, Error, Output>(
+    predicate: (snapshot: RuntimeSnapshot<State, Error, Output>) => boolean
+  ): <Event, Emitted>(ref: MachineRef<State, Event, Error, Output, Emitted>) => Effect.Effect<
+    RuntimeSnapshot<State, Error, Output>,
+    Error | StoppedError | Cause.NoSuchElementError
+  >
+  <State, Event, Error, Output, Emitted, Narrowed extends RuntimeSnapshot<State, Error, Output>>(
+    ref: MachineRef<State, Event, Error, Output, Emitted>,
+    predicate: (snapshot: RuntimeSnapshot<State, Error, Output>) => snapshot is Narrowed
+  ): Effect.Effect<Narrowed, Error | StoppedError | Cause.NoSuchElementError>
+  <State, Event, Error, Output, Emitted>(
+    ref: MachineRef<State, Event, Error, Output, Emitted>,
+    predicate: (snapshot: RuntimeSnapshot<State, Error, Output>) => boolean
+  ): Effect.Effect<RuntimeSnapshot<State, Error, Output>, Error | StoppedError | Cause.NoSuchElementError>
+} = internal.waitFor
+
 /**
  * Prepares a fresh machine without initializing it.
  *

@@ -131,6 +131,7 @@ export const invocation = (
   const field = { effects: "effect", streams: "stream", timers: "after", logic: "logic" }[source.kind]
   return {
     ...rest,
+    sourceName: src,
     id: rest.id ?? src,
     [field]: !parameterized && (source.kind === "timers" || source.kind === "logic") ? source.value : resolve
   }

@@ -537,7 +537,8 @@ const normalizeIndexedTargetStateSync = (
   const targetIndex = isTarget(target) ? descriptor.indexByPath.get(String(target.path)) : undefined
   if (
     targetIndex === activeLeafIndex && current.active[activeLeafIndex] === 1 && isTarget(target) &&
-    target[TargetSnapshotTypeId] === undefined && target.values === undefined && current.completedOrder.length === 0
+    target[TargetSnapshotTypeId] === undefined && target.values === undefined && current.completedOrder.length === 0 &&
+    descriptor.valued[activeLeafIndex]
   ) {
     const values = current.values.slice()
     values[activeLeafIndex] = decodeStateValueSync(
@@ -863,7 +864,8 @@ const planIndexedFlatState = (
           ? target[TargetSnapshotTypeId] === undefined && target.values === undefined
           : !("state" in target) && !("states" in target) && !("completed" in target) && !("history" in target)
         if (
-          targetIndex === sourceIndex && isSimpleTarget && current.completedOrder.length === 0
+          targetIndex === sourceIndex && isSimpleTarget && current.completedOrder.length === 0 &&
+          descriptor.valued[sourceIndex]
         ) {
           updateOwnedIndexedValue(
             current,

@@ -19,6 +19,7 @@ import { ChildMachineLogicTypeId } from "./symbols.js"
 /** @internal */
 export interface AnyConfig {
   readonly id: string
+  readonly sourceName?: string | undefined
   readonly address?: string
   readonly descriptor?: ChildMachine.Any
   readonly src: () => Runtime.ProcessLogic<any, any, any, any, any, any>
@@ -82,6 +83,7 @@ const resolveOne = (
   if ("effect" in raw) {
     return {
       id: String(raw.id),
+      sourceName: raw.sourceName,
       src: () =>
         oneShot(raw.effect(context) as Effect.Effect<any, any, any>) as unknown as Runtime.ProcessLogic<
           any,
@@ -100,6 +102,7 @@ const resolveOne = (
   if ("after" in raw) {
     return {
       id: String(raw.id),
+      sourceName: raw.sourceName,
       src: () =>
         oneShot(Effect.sleep(resolveValue(raw.after, context) as any)) as unknown as Runtime.ProcessLogic<
           any,
@@ -117,6 +120,7 @@ const resolveOne = (
     const id = String(raw.id)
     return {
       id,
+      sourceName: raw.sourceName,
       src: () =>
         streamLogic(raw.stream(context), path, id) as unknown as Runtime.ProcessLogic<any, any, any, any, any, any>,
       onDone: raw.onDone,
@@ -192,11 +196,37 @@ const startResolved = (
   onDone: unknown,
   onFailure: unknown,
   onSnapshot: unknown,
-  activityKind?: Inspection.Activity["kind"]
+  activityKind?: Inspection.Activity["kind"],
+  sourceName?: string
 ): Effect.Effect<void, any, any> =>
   Effect.suspend(() => {
     const key = makeKey(path, invokeId)
-    return ownedChildren.spawn(src, {
+    // Source construction keeps its startup failure boundary. Only the hosted
+    // program is wrapped; child statecharts retain their execution descriptors.
+    const tracedSource = activityKind === undefined ? src : () => {
+      const logic = src()
+      return {
+        ...logic,
+        run: (context: Runtime.ProcessContext<any, any>) =>
+          Effect.withSpan(
+            Effect.suspend(() => logic.run(context)),
+            "Machine.invoke",
+            {
+              attributes: {
+                "machine.id": scope.self.id,
+                "machine.sessionId": scope.self.sessionId,
+                "machine.state.path": path,
+                "machine.invoke.id": invokeId,
+                "machine.invoke.source": sourceName ?? invokeId,
+                "machine.invoke.kind": activityKind,
+                "machine.invoke.sessionId": context.self.sessionId
+              },
+              captureStackTrace: false
+            }
+          )
+      }
+    }
+    return ownedChildren.spawn(tracedSource, {
       key,
       path,
       id: childId,
@@ -260,7 +290,8 @@ const start = (
     config.onDone,
     config.onFailure,
     config.onSnapshot,
-    config.activityKind
+    config.activityKind,
+    config.sourceName
   )
 }
 

@@ -16,6 +16,7 @@ interface Outcomes {
 
 export type InvocationDefinition =
   & Outcomes
+  & { readonly sourceName?: string }
   & (
     | { readonly id: string; readonly effect: (context: Context) => Effect.Effect<unknown, unknown, unknown> }
     | { readonly id: string; readonly stream: (context: Context) => Stream.Stream<unknown, unknown, unknown> }

@@ -34,6 +34,7 @@ import type { ChildAlreadyExistsError, InfiniteTransitionError, StartupError } f
 import type { CapturedStateConfig } from "./implementation.js"
 import * as InitialDeclaration from "./initialDeclaration.js"
 import * as InvocationDefinition from "./invocationDefinition.js"
+import * as Observation from "./observation.js"
 import * as internalPlanner from "./planner.js"
 import * as internalProcess from "./process.js"
 import * as Protocol from "./protocol.js"
@@ -1411,6 +1412,8 @@ export const watch = <State, Event, Error = never, Output = never>(
   ref: MachineRef<State, Event, Error, Output>
 ): Stream.Stream<RuntimeOutcome<State, Error, Output>> => internalRuntime.watch(ref)
 
+export const waitFor = Observation.waitFor
+
 export const prepare = internalProcess.prepare
 
 export const start: <

@@ -0,0 +1,39 @@
+/** Observation of current and subsequent published runtime snapshots. */
+import * as Cause from "effect/Cause"
+import * as Effect from "effect/Effect"
+import { dual } from "effect/Function"
+import * as Option from "effect/Option"
+import * as Stream from "effect/Stream"
+import type * as Machine from "../../Machine.js"
+import { StoppedError } from "./errors.js"
+
+export const waitFor: typeof Machine.waitFor = dual(2, <State, Event, Error, Output, Emitted>(
+  ref: Machine.MachineRef<State, Event, Error, Output, Emitted>,
+  predicate: (snapshot: Machine.RuntimeSnapshot<State, Error, Output>) => boolean
+): Effect.Effect<Machine.RuntimeSnapshot<State, Error, Output>, Error | StoppedError | Cause.NoSuchElementError> =>
+  Effect.suspend(() => {
+    let result = Option.none<Machine.RuntimeSnapshot<State, Error, Output>>()
+    const absent = () => new Cause.NoSuchElementError(`Machine "${ref.id}" ended without a matching snapshot`)
+    return Stream.runForEachWhile(
+      ref.changes,
+      (snapshot) =>
+        Effect.suspend((): Effect.Effect<boolean, Error | StoppedError | Cause.NoSuchElementError> => {
+          if (predicate(snapshot)) {
+            result = Option.some(snapshot)
+            return Effect.succeed(false)
+          }
+          switch (snapshot.status) {
+            case "active":
+              return Effect.succeed(true)
+            case "error":
+              return Effect.failCause(snapshot.cause)
+            case "stopped":
+              return Effect.fail(new StoppedError())
+            case "done":
+              return Effect.fail(absent())
+          }
+        })
+    ).pipe(
+      Effect.flatMap(() => Option.isSome(result) ? Effect.succeed(result.value) : Effect.fail(absent()))
+    )
+  }))