Bind started machines to the caller's Scope

.changeset/scope-bound-machine-start.md

@@ -0,0 +1,21 @@
+---
+"@typeonce/effect-machine": minor
+---
+
+Bind started machines to the caller's `Scope`.
+
+`Machine.start`, `Machine.resume`, and `prepared.start` now require `Scope.Scope`, like `Effect.forkScoped`. Closing that scope is equivalent to `ref.stop`: active invocations are interrupted, owned children stop, the snapshot becomes `stopped`, and later sends fail with `StoppedError`. A machine that stops or completes earlier releases its place in the scope, so long-lived scopes do not accumulate finalizers. For `prepared.start`, the scope that first evaluates it owns the machine. `AtomMachine` is unaffected.
+
+Wrap programs that start machines in `Effect.scoped`, or own a scope explicitly when a machine must outlive the caller:
+
+```ts
+const program = Effect.scoped(Effect.gen(function*() {
+  const ref = yield* Machine.start(machine)
+  yield* ref.send(Events.Go())
+}))
+
+const detached = Effect.gen(function*() {
+  const scope = yield* Scope.make()
+  return yield* Machine.start(machine).pipe(Scope.provide(scope))
+})
+```

packages/effect-machine/README.md

@@ -75,6 +75,11 @@ Root data is shared state. `update: "root"` replaces that value while
 retaining the active child configuration and its running work. Ordinary
 transitions target a descendant; the root itself is not a destination.
 
+A started machine is owned by the enclosing `Scope`. Closing it is equivalent
+to `ref.stop`: invocations are interrupted, owned children stop, and later
+sends fail with `StoppedError`. Own a `Scope` explicitly when a machine must
+outlive the program that started it.
+
 ## Modeling workflow
 
 Put data on the lowest state that owns it. Use tagged schemas or `fields` for
@@ -567,7 +572,7 @@ Snapshots can be validated for storage or transport:
 ```ts
 const encoded = yield * Machine.encodeSnapshot(machine, snapshot)
 const decoded = yield * Machine.decodeSnapshot(machine, encoded)
-const ref = yield * Machine.resume(machine, decoded)
+const ref = yield * Machine.resume(machine, decoded) // owned by the current Scope
 ```
 
 Decoded snapshots are local runtime values and may contain class instances or

packages/effect-machine/docs/agent-guide.md

@@ -81,14 +81,17 @@ through the exported descriptor:
 ```ts
 import { Effect } from "effect"
 
-const program = Effect.gen(function*() {
+const program = Effect.scoped(Effect.gen(function*() {
   const counter = yield* Machine.start(CounterMachine)
 
   yield* counter.send(CounterEvents.Start())
   yield* counter.send(CounterEvents.Increment())
-})
+}))
 ```
 
+`Machine.start` requires a `Scope`. Closing it stops the machine and everything
+it owns; use `ref.stop` to stop earlier.
+
 ## Make impossible states unrepresentable
 
 A finite state describes how the machine behaves now. State data holds values

packages/effect-machine/perf/runtime/counter.mjs

@@ -24,7 +24,13 @@ if (machineRuntimePath === undefined) {
   throw new Error("Effect Machine benchmark could not locate the internal runtime module")
 }
 const machineRuntime = await import(pathToFileURL(machineRuntimePath).href)
-const { Effect, Fiber, Option, Schema, Stream } = effect
+const { Context, Effect, Fiber, Option, Schema, Scope, Stream } = effect
+
+// Benchmarks own started machines explicitly and stop them through `ref.stop`.
+// One long-lived Scope in one shared service context stands in for the
+// enclosing program, so starts do not copy the service context per machine.
+const benchmarkScope = Scope.makeUnsafe()
+const runPromise = Effect.runPromiseWith(Context.make(Scope.Scope, benchmarkScope))
 
 const invocationLifecycleMachine = benchmarkApi.effectLifecycle?.(Effect.void, Schema.Void)
 const invocationLifecycleBenchmarks = invocationLifecycleMachine === undefined ? [] : [true, false].map((enabled) => ({
@@ -34,7 +40,7 @@ const invocationLifecycleBenchmarks = invocationLifecycleMachine === undefined ?
   operations: () => 1,
   expected: () => 1,
   start: () => undefined,
-  run: () => Effect.runPromise(Effect.gen(function*() {
+  run: () => runPromise(Effect.gen(function*() {
     const ref = yield* Machine.start(invocationLifecycleMachine)
     yield* ref.join
     return 1
@@ -327,12 +333,12 @@ const planParallelCounterBatch = (size) =>
     })
   )
 
-export const startCounter = () => Effect.runPromise(Machine.start(counterMachine))
+export const startCounter = () => runPromise(Machine.start(counterMachine))
 
-export const stopCounter = (ref) => Effect.runPromise(ref.stop)
+export const stopCounter = (ref) => runPromise(ref.stop)
 
 export const startObservedCounter = () =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       const ref = yield* Machine.start(counterMachine)
       const observer = yield* ref.changes.pipe(Stream.runDrain, Effect.forkDetach)
@@ -342,12 +348,12 @@ export const startObservedCounter = () =>
   )
 
 export const stopObservedCounter = ({ ref, observer }) =>
-  Effect.runPromise(
+  runPromise(
     ref.stop.pipe(Effect.ensuring(Fiber.interrupt(observer)))
   )
 
 export const runObservedCounterBurst = ({ ref, observer }, size) =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       for (let index = 0; index < size; index += 1) {
         yield* ref.send(incrementEvent)
@@ -372,18 +378,18 @@ const waitForCounterChild = (parent) =>
   })
 
 export const startChildCounter = () =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       const parent = yield* Machine.start(counterParentMachine)
       yield* waitForCounterChild(parent)
       return parent
     })
   )
 
-export const stopChildCounter = (parent) => Effect.runPromise(parent.stop)
+export const stopChildCounter = (parent) => runPromise(parent.stop)
 
 export const runChildCounterBurst = (parent, size) =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       for (let index = 0; index < size; index += 1) {
         const child = yield* parent.child(CounterChild)
@@ -402,7 +408,7 @@ export const runChildCounterBurst = (parent, size) =>
   )
 
 export const runCounterBurst = (ref, size) =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       for (let index = 0; index < size; index += 1) {
         yield* ref.send(incrementEvent)
@@ -412,11 +418,11 @@ export const runCounterBurst = (ref, size) =>
     })
   )
 
-const startHierarchicalCounter = () => Effect.runPromise(Machine.start(hierarchicalCounterMachine))
-const startParallelCounter = () => Effect.runPromise(Machine.start(parallelCounterMachine))
+const startHierarchicalCounter = () => runPromise(Machine.start(hierarchicalCounterMachine))
+const startParallelCounter = () => runPromise(Machine.start(parallelCounterMachine))
 
 const startObservedHierarchicalCounter = () =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       const ref = yield* Machine.start(hierarchicalCounterMachine)
       const observer = yield* ref.changes.pipe(Stream.runDrain, Effect.forkDetach)
@@ -426,7 +432,7 @@ const startObservedHierarchicalCounter = () =>
   )
 
 const runHierarchicalCounterBurst = (ref, size) =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       for (let index = 0; index < size; index += 1) {
         yield* ref.send(hierarchicalIncrementEvent)
@@ -437,7 +443,7 @@ const runHierarchicalCounterBurst = (ref, size) =>
   )
 
 const runParallelCounterBurst = (ref, size) =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       for (let index = 0; index < size; index += 1) {
         yield* ref.send(index % 2 === 0 ? parallelIncrementLeftEvent : parallelIncrementRightEvent)
@@ -448,7 +454,7 @@ const runParallelCounterBurst = (ref, size) =>
   )
 
 const runObservedHierarchicalCounterBurst = ({ ref, observer }, size) =>
-  Effect.runPromise(
+  runPromise(
     Effect.gen(function*() {
       for (let index = 0; index < size; index += 1) {
         yield* ref.send(hierarchicalIncrementEvent)
@@ -461,7 +467,7 @@ const runObservedHierarchicalCounterBurst = ({ ref, observer }, size) =>
   )
 
 export const startCounters = (count) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       Array.from({ length: count }),
       () => Machine.start(counterMachine),
@@ -470,15 +476,15 @@ export const startCounters = (count) =>
   )
 
 export const stopCounters = (refs) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(refs, (ref) => ref.stop, {
       concurrency: "unbounded",
       discard: true
     })
   )
 
 export const startChildCounters = (count) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       Array.from({ length: count }),
       () =>
@@ -522,7 +528,7 @@ const rawCompiledProcessLogic = machineRuntime.compiledProcess === undefined
   }
 
 const startRawProcesses = (count) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       Array.from({ length: count }),
       () => machineRuntime.startProcess(rawProcessLogic),
@@ -531,7 +537,7 @@ const startRawProcesses = (count) =>
   )
 
 const startCompiledRawProcesses = (count) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       Array.from({ length: count }),
       () => machineRuntime.startProcessWithStrategyForTesting(rawCompiledProcessLogic, "compiled"),
@@ -540,7 +546,7 @@ const startCompiledRawProcesses = (count) =>
   )
 
 const startIndependentCounterPairs = (count) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       Array.from({ length: count }),
       () => Effect.all([Machine.start(counterMachine), Machine.start(counterMachine)], { concurrency: 1 }),
@@ -551,7 +557,7 @@ const startIndependentCounterPairs = (count) =>
 const stopIndependentCounterPairs = (pairs) => stopCounters(pairs.flat())
 
 const startObservedChildCounters = (count) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       Array.from({ length: count }),
       () =>
@@ -570,7 +576,7 @@ const startObservedChildCounters = (count) =>
   )
 
 const startSnapshotObservedChildCounters = (count) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       Array.from({ length: count }),
       () =>
@@ -584,7 +590,7 @@ const startSnapshotObservedChildCounters = (count) =>
   )
 
 const stopObservedChildCounters = (units) =>
-  Effect.runPromise(
+  runPromise(
     Effect.forEach(
       units,
       ({ parent, observer }) => parent.stop.pipe(Effect.ensuring(Fiber.interrupt(observer))),

packages/effect-machine/src/Machine.ts

@@ -2011,7 +2011,8 @@ export type RuntimeOutcome<State, Error = never, Output = never> =
  *
  * Subscribe to `emissions` before evaluating `start` when initial-entry
  * emissions must be observed. `start` is one-shot: concurrent and repeated
- * evaluations share the same initialization and running reference.
+ * evaluations share the same initialization and running reference, and the
+ * `Scope` of the first evaluation owns the machine.
  *
  * @category models
  * @since 0.11.0
@@ -2037,7 +2038,7 @@ export interface Prepared<out State, in Event, out Error, out Output, out Emitte
    */
   readonly inspection: Stream.Stream<Inspection.Event>
 
-  /** Initializes this machine once and returns its running reference. */
+  /** Initializes this machine once, owned by the first evaluating `Scope`, and returns its running reference. */
   readonly start: Effect.Effect<MachineRef<State, Event, Error, Output, Emitted>, StartError, StartRequirements>
 }
 
@@ -9278,6 +9279,9 @@ export const waitFor: {
  * `prepared.start`. Ordinary callers can continue to use {@link start}, which
  * starts directly and does not allocate the prepared lifecycle boundary.
  *
+ * `prepared.start` is one-shot and scoped: the `Scope` that first evaluates it
+ * owns the machine, and closing that `Scope` is equivalent to `ref.stop`.
+ *
  * ```ts
  * const prepared = yield* Machine.prepare(machine)
  *
@@ -9346,7 +9350,8 @@ export const prepare: <
     | MachineSchemaDecodeError
     | StartupError
     | StoppedError,
-    ExcludeCompatibleRuntime<
+    | Scope.Scope
+    | ExcludeCompatibleRuntime<
       ExecutionServices<InitialR | R>,
       Machine.EventOf<Events>,
       Machine.EmittedEventOf<Emits>
@@ -9355,7 +9360,7 @@ export const prepare: <
 > = internal.prepare
 
 /**
- * Starts a machine.
+ * Starts a machine owned by the current `Scope`.
  *
  * **When to use**
  *
@@ -9369,6 +9374,13 @@ export const prepare: <
  * publishes the new state, delivers emitted events, and then starts invokes
  * for entered states.
  *
+ * The machine is a scoped resource, like `Effect.forkScoped`. Closing the
+ * `Scope` is equivalent to `ref.stop`: active invocations are interrupted,
+ * owned children stop, the snapshot becomes `stopped`, and later sends fail
+ * with `StoppedError`. An earlier `ref.stop` or terminal completion leaves the
+ * finalizer with nothing to do. Own a `Scope` explicitly, for example with
+ * `Scope.make` and `Scope.provide`, when the machine must outlive the caller.
+ *
  * **Gotchas**
  *
  * The returned handle's `send` operation only enqueues events. Transition
@@ -9395,10 +9407,10 @@ export const prepare: <
  *     Idle: {}
  *   }
  * })
- * const state = Effect.gen(function*() {
+ * const state = Effect.scoped(Effect.gen(function*() {
  *   const ref = yield* Machine.start(machine)
  *   return yield* ref.state
- * })
+ * }))
  * ```
  *
  * @see {@link plan} for inspecting the same transition plan without executing it.
@@ -9461,18 +9473,23 @@ export const start: <
   | MachineSchemaDecodeError
   | StartupError
   | StoppedError,
-  ExcludeCompatibleRuntime<
+  | Scope.Scope
+  | ExcludeCompatibleRuntime<
     ExecutionServices<InitialR | R>,
     Machine.EventOf<Events>,
     Machine.EmittedEventOf<Emits>
   >
 > = internal.start
 
 /**
- * Starts a fresh managed runtime from a decoded logical snapshot.
+ * Starts a fresh managed runtime from a decoded logical snapshot, owned by the
+ * current `Scope`.
  *
  * **Details**
  *
+ * The resumed machine has the same scoped lifetime as {@link start}: closing
+ * the `Scope` is equivalent to `ref.stop`.
+ *
  * `resume` validates and normalizes the supplied snapshot before publishing it
  * as the first state. It does not call the machine's initial function, replay
  * entry or transition actions, re-deliver raised or emitted events, or
@@ -9511,12 +9528,12 @@ export const start: <
  *     Idle: {}
  *   }
  * })
- * const resumed = Effect.gen(function*() {
+ * const resumed = Effect.scoped(Effect.gen(function*() {
  *   const initial = yield* Machine.planInitial(machine)
  *   const encoded = yield* Machine.encodeSnapshot(machine, initial.state)
  *   const snapshot = yield* Machine.decodeSnapshot(machine, encoded)
  *   return yield* Machine.resume(machine, snapshot)
- * })
+ * }))
  * ```
  *
  * @see {@link decodeSnapshot} for the schema and transport boundary.
@@ -9573,7 +9590,8 @@ export const resume: <
     Machine.EmittedEventOf<Emits>
   >,
   MachineSchemaDecodeError,
-  ExcludeCompatibleRuntime<
+  | Scope.Scope
+  | ExcludeCompatibleRuntime<
     ExecutionServices<R>,
     Machine.EventOf<Events>,
     Machine.EmittedEventOf<Emits>