fix(runtime): start actor effects eagerly

.changeset/eager-actor-effects.md

@@ -0,0 +1,5 @@
+---
+"effect-machine": patch
+---
+
+Start state and background Effects eagerly so their synchronous setup completes before actor state becomes visible. A task that completes in its first Effect slice can now enqueue its result before a caller sends its next event. A suspending Inspector can delay this setup because inspection remains ordered and awaited.

src/actor.ts

@@ -1094,7 +1094,12 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* <
     // Delegate to runtime.start (forks event loop, background, spawn effects)
     const currentRuntime = runtimeRef.current;
     if (currentRuntime !== undefined) {
-      yield* currentRuntime.start;
+      yield* currentRuntime.start.pipe(
+        Effect.catchCause((cause) => {
+          if (supervision === undefined) return Effect.failCause(cause);
+          return Effect.void;
+        }),
+      );
       const currentExit = yield* Deferred.poll(currentRuntime.exitDeferred);
       if (Option.isNone(currentExit)) {
         yield* SubscriptionRef.set(lifecycleRef, {

src/internal/runtime.ts

@@ -381,7 +381,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
           self,
           system,
         })
-        .pipe(Effect.forkIn(actorScope));
+        .pipe(Effect.forkIn(actorScope, { startImmediately: true }));
       backgroundFibers.push(fiber);
     }
 
@@ -438,10 +438,9 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
         Deferred.succeed(exitDeferred, RuntimeExit.Defect(cause, "spawn")).pipe(
           Effect.andThen(Ref.set(stoppedRef, true)),
           Effect.andThen(
-            Effect.suspend(() => {
+            Effect.sync(() => {
               const loopFiber = loopFiberRef.current;
-              if (loopFiber !== undefined) return Fiber.interrupt(loopFiber);
-              return Effect.void;
+              if (loopFiber !== undefined) fork(Fiber.interrupt(loopFiber));
             }),
           ),
           Effect.asVoid,
@@ -725,6 +724,10 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function*
     Effect.gen(function* () {
       const event = queued.event;
 
+      if (queued._tag === "ask" && deferredReplyRef !== undefined) {
+        deferredReplyRef.current = queued.reply;
+      }
+
       // Lifecycle: onEvent (actor emits @machine.event)
       if (lifecycle?.onEvent !== undefined) yield* lifecycle.onEvent(currentState, event);
 
@@ -776,6 +779,9 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function*
           break;
         case "ask":
           if (result.hasReply) {
+            if (deferredReplyRef?.current === queued.reply) {
+              deferredReplyRef.current = undefined;
+            }
             const replySchema = machine._replySchema(event._tag);
             if (replySchema !== undefined) {
               const decoded = yield* Schema.decodeUnknownEffect(replySchema)(result.reply).pipe(
@@ -790,10 +796,10 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function*
             } else {
               yield* Deferred.succeed(queued.reply, result.reply);
             }
-          } else if (result.deferReply && deferredReplyRef !== undefined) {
-            // Handler returned Machine.deferReply() — spawn handler will call self.reply()
-            deferredReplyRef.current = queued.reply;
-          } else {
+          } else if (!result.deferReply) {
+            if (deferredReplyRef?.current === queued.reply) {
+              deferredReplyRef.current = undefined;
+            }
             yield* Deferred.fail(
               queued.reply,
               NoReplyError.make({ actorId, eventTag: event._tag }),

src/internal/transition.ts

@@ -688,7 +688,13 @@ export const runSpawnEffects = Effect.fn("effect-machine.runSpawnEffects")(funct
         }),
       );
 
-    yield* Effect.forkScoped(effect).pipe(Effect.provideService(Scope.Scope, stateScope));
+    const fiber = yield* Effect.forkScoped(effect, { startImmediately: true }).pipe(
+      Effect.provideService(Scope.Scope, stateScope),
+    );
+    const exit = fiber.pollUnsafe();
+    if (exit?._tag === "Failure" && !Cause.hasInterruptsOnly(exit.cause)) {
+      return yield* Effect.failCause(exit.cause);
+    }
   }
 });
 

test/actor.test.ts

@@ -808,9 +808,7 @@ describe("ActorRef", () => {
           })
           .final(TS.Done);
 
-        // Mirrors the gent AgentLoop pattern: send → yieldNow → waitFor(Running)
-        // The task fails immediately so Running→Idle can happen before
-        // waitFor subscribes. With the old get-then-subscribe waitFor, this hangs.
+        // The wait must own its subscription before the task can leave Running.
         interface LoopService {
           readonly run: () => Effect.Effect<string>;
         }
@@ -841,9 +839,11 @@ describe("ActorRef", () => {
               run: () =>
                 Effect.gen(function* () {
                   const actor = yield* getActor;
+                  const runningFiber = yield* Effect.forkDetach(actor.waitFor(TS.Running), {
+                    startImmediately: true,
+                  });
                   yield* actor.send(TE.Start);
-                  yield* Effect.yieldNow;
-                  yield* actor.waitFor(TS.Running);
+                  yield* Fiber.join(runningFiber);
                   yield* actor.waitFor((s) => s._tag !== "Running");
                   const final = yield* actor.snapshot;
                   return final._tag;

test/eager-effect-start.test.ts

@@ -0,0 +1,232 @@
+// @effect-diagnostics strictEffectProvide:off - tests are entry points
+import { Cause, Effect, Schema } from "effect";
+import * as AtomRegistry from "effect/unstable/reactivity/AtomRegistry";
+import { describe, expect, it } from "effect-bun-test";
+
+import * as ActorAtom from "../src/atom.js";
+import { Event, Machine, State } from "../src/index.js";
+
+const LifecycleState = State({ Idle: {}, Listening: {}, Done: {} });
+const LifecycleEvent = Event({ Start: {}, Stop: {} });
+
+describe("actor Effect eager start", () => {
+  it.scopedLive("starts state Effect setup before state subscribers run", () =>
+    Effect.gen(function* () {
+      const records: Array<string> = [];
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Idle,
+      })
+        .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening)
+        .spawn(LifecycleState.Listening, () =>
+          Effect.sync(() => records.push("setup")).pipe(Effect.andThen(Effect.never)),
+        );
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+      const unsubscribe = actor.subscribe((state) => {
+        if (LifecycleState.$is("Listening")(state)) records.push("visible");
+      });
+
+      yield* actor.call(LifecycleEvent.Start);
+
+      expect(records).toEqual(["setup", "visible"]);
+      unsubscribe();
+      yield* actor.stop;
+    }),
+  );
+
+  it.scopedLive("starts state Effects in registration order", () =>
+    Effect.gen(function* () {
+      const records: Array<string> = [];
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Idle,
+      })
+        .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening)
+        .spawn(LifecycleState.Listening, () =>
+          Effect.sync(() => records.push("first")).pipe(Effect.andThen(Effect.never)),
+        )
+        .spawn(LifecycleState.Listening, () =>
+          Effect.sync(() => records.push("second")).pipe(Effect.andThen(Effect.never)),
+        );
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+      const unsubscribe = actor.subscribe((state) => {
+        if (LifecycleState.$is("Listening")(state)) records.push("visible");
+      });
+
+      yield* actor.call(LifecycleEvent.Start);
+
+      expect(records).toEqual(["first", "second", "visible"]);
+      unsubscribe();
+      yield* actor.stop;
+    }),
+  );
+
+  it.scopedLive("starts task setup before state subscribers run", () =>
+    Effect.gen(function* () {
+      const records: Array<string> = [];
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Idle,
+      })
+        .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening)
+        .task(
+          LifecycleState.Listening,
+          () => Effect.sync(() => records.push("task")).pipe(Effect.andThen(Effect.never)),
+          { onSuccess: () => LifecycleEvent.Stop },
+        );
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+      const unsubscribe = actor.subscribe((state) => {
+        if (LifecycleState.$is("Listening")(state)) records.push("visible");
+      });
+
+      yield* actor.call(LifecycleEvent.Start);
+
+      expect(records).toEqual(["task", "visible"]);
+      unsubscribe();
+      yield* actor.stop;
+    }),
+  );
+
+  it.scopedLive("starts background Effect setup before actor start completes", () =>
+    Effect.gen(function* () {
+      const records: Array<string> = [];
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Idle,
+      }).background(() =>
+        Effect.sync(() => records.push("background")).pipe(Effect.andThen(Effect.never)),
+      );
+      const actor = yield* Machine.spawn(machine);
+
+      yield* actor.start;
+
+      expect(records).toEqual(["background"]);
+      yield* actor.stop;
+    }),
+  );
+
+  it.scopedLive("interrupts an eagerly started state Effect on state exit", () =>
+    Effect.gen(function* () {
+      const records: Array<string> = [];
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Idle,
+      })
+        .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening)
+        .on(LifecycleState.Listening, LifecycleEvent.Stop, () => LifecycleState.Done)
+        .spawn(LifecycleState.Listening, () =>
+          Effect.gen(function* () {
+            yield* Effect.addFinalizer(() => Effect.sync(() => records.push("cleanup")));
+            records.push("setup");
+            return yield* Effect.never;
+          }),
+        );
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+
+      yield* actor.call(LifecycleEvent.Start);
+      yield* actor.call(LifecycleEvent.Stop);
+
+      expect(records).toEqual(["setup", "cleanup"]);
+      yield* actor.stop;
+    }),
+  );
+
+  it.scopedLive("reports a synchronous state Effect defect as spawn", () =>
+    Effect.gen(function* () {
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Idle,
+      })
+        .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening)
+        .spawn(LifecycleState.Listening, () => Effect.die("spawn boom"));
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+
+      yield* actor.send(LifecycleEvent.Start);
+      const exit = yield* actor.awaitExit;
+
+      expect(exit._tag).toBe("Defect");
+      if (exit._tag === "Defect") {
+        expect(exit.phase).toBe("spawn");
+        expect(Cause.hasInterruptsOnly(exit.cause)).toBe(false);
+        expect(Cause.pretty(exit.cause)).toContain("spawn boom");
+      }
+    }),
+  );
+
+  it.scopedLive("settles a deferred ask from synchronous state Effect setup", () =>
+    Effect.gen(function* () {
+      const ReplyState = State({ Idle: {}, Replying: {} });
+      const ReplyEvent = Event({ Request: Event.reply({}, Schema.String) });
+      const replyResults: Array<boolean> = [];
+      const machine = Machine.make({
+        state: ReplyState,
+        event: ReplyEvent,
+        initial: ReplyState.Idle,
+      })
+        .on(ReplyState.Idle, ReplyEvent.Request, () => Machine.deferReply(ReplyState.Replying))
+        .spawn(ReplyState.Replying, ({ self }) =>
+          self.reply("ready").pipe(
+            Effect.tap((didReply) => Effect.sync(() => replyResults.push(didReply))),
+            Effect.andThen(Effect.never),
+          ),
+        );
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+
+      const reply = yield* Effect.race(
+        actor.ask(ReplyEvent.Request),
+        Effect.sleep("100 millis").pipe(Effect.as("timeout")),
+      );
+      yield* Effect.yieldNow;
+
+      expect(reply).toBe("ready");
+      expect(replyResults).toEqual([true]);
+      yield* actor.stop;
+    }),
+  );
+
+  it.scopedLive("starts state Effect setup before Actor Atom subscribers run", () =>
+    Effect.gen(function* () {
+      const records: Array<string> = [];
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Idle,
+      })
+        .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening)
+        .spawn(LifecycleState.Listening, () =>
+          Effect.sync(() => records.push("setup")).pipe(Effect.andThen(Effect.never)),
+        );
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+      const registry = AtomRegistry.make();
+      const stateAtom = ActorAtom.make(actor);
+      const unsubscribe = registry.subscribe(
+        stateAtom,
+        (state) => {
+          if (LifecycleState.$is("Listening")(state)) records.push("visible");
+        },
+        { immediate: true },
+      );
+
+      yield* actor.call(LifecycleEvent.Start);
+      yield* Effect.yieldNow;
+
+      expect(records).toEqual(["setup", "visible"]);
+      unsubscribe();
+      registry.dispose();
+      yield* actor.stop;
+    }),
+  );
+});

test/inspection.test.ts

@@ -247,9 +247,9 @@ describe("Inspection", () => {
       }).spawn(TestState.Idle, () => Effect.die("boom"));
 
       const system = yield* ActorSystemService;
-      yield* system.spawn("test", machine);
-      yield* yieldFibers;
+      const spawnExit = yield* Effect.exit(system.spawn("test", machine));
 
+      expect(spawnExit._tag).toBe("Failure");
       const errorEvent = events.find((e) => e.type === "@machine.error");
       expect(errorEvent).toBeDefined();
       if (errorEvent?.type === "@machine.error") {

test/supervision.test.ts

@@ -34,6 +34,31 @@ const machine = Machine.make({ state: S, event: E, initial: S.Idle })
 // ============================================================================
 
 describe("supervision: restart on defect", () => {
+  it.scopedLive("restarts after synchronous initial state Effect defect", () =>
+    Effect.gen(function* () {
+      const restarted = yield* Deferred.make<void>();
+      let attempts = 0;
+      const initialEffectMachine = Machine.make({ state: S, event: E, initial: S.Idle }).spawn(
+        S.Idle,
+        () =>
+          Effect.suspend(() => {
+            attempts += 1;
+            if (attempts === 1) return Effect.die("initial boom");
+            return Deferred.succeed(restarted, void 0);
+          }),
+      );
+      const actor = yield* Machine.spawn(initialEffectMachine, {
+        supervision: Supervision.restart({ maxRestarts: 1 }),
+      });
+
+      yield* actor.start;
+      yield* Deferred.await(restarted);
+
+      expect(attempts).toBe(2);
+      yield* actor.stop;
+    }),
+  );
+
   it.scopedLive("restarts actor from initial state after transition defect", () =>
     Effect.gen(function* () {
       const actor = yield* Machine.spawn(machine, {