Complete supervised recovery and restart shutdown

.changeset/plain-supervisor-stop.md

@@ -0,0 +1,7 @@
+---
+"effect-machine": patch
+---
+
+Allow an actor to stop itself from protected supervised recovery. Wait for the supervisor to finish cleanup and report its cleanup defects to every stop caller.
+
+Apply the restart policy when a restarted generation fails during startup. Continue within the retry budget and complete the actor exit when that budget ends.

AGENTS.md

@@ -281,6 +281,8 @@ const count = yield* actor.ask(Event.GetCount);  // number
 
 - `actor.stop` cancels pending startup and waits for recovery cleanup. Stop callers can cancel their own wait without cancelling shutdown. A stop from recovery marks that startup interrupted. Shutdown waits for its protected regions and finalizers to finish. Recovery cleanup defects reach every stop caller.
 - Protect both shutdown owner creation and its cache publication from interruption. Protecting only the cached body can cache cancellation when its protected region ends.
+- Recovery self-stop must also identify the supervisor fiber. A protected supervisor must not join an owner that waits for that same supervisor. Preserve supervised recovery cleanup defects on stop.
+- A supervised generation can fail during `runtime.start`. Let the loop read its recorded exit and apply the restart policy. Do not let that start failure end the supervisor before it completes the actor exit.
 - The runtime event loop must be ready before startup completes. Host client sends must still commit synchronous transitions before returning.
 - Publish Active only while the same generation is still Starting. A terminal lifecycle must never return to Active.
 - `actor.start` shares one startup result across concurrent and repeated calls, including failure or interruption. Stop the actor and spawn a new actor to retry initialization. It cannot restart a terminal actor.

src/actor.ts

@@ -777,7 +777,8 @@ const runSupervisionLoop = <
 
       const newRuntime = yield* options.spawnGeneration(cell.machine);
       cell.runtimeRef.current = newRuntime;
-      yield* newRuntime.start;
+      // The runtime records startup failure. The next iteration applies the restart policy.
+      yield* newRuntime.start.pipe(Effect.ignoreCause);
       const restartExit = yield* Deferred.poll(newRuntime.exitDeferred);
       if (Option.isNone(restartExit)) {
         yield* activateGeneration(cell.lifecycleRef, nextGeneration);
@@ -1072,6 +1073,11 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* <
     }
     if (supervisorFiberRef.current !== undefined) {
       yield* Fiber.interrupt(supervisorFiberRef.current);
+      const supervisorExit = yield* Fiber.await(supervisorFiberRef.current);
+      if (Exit.isFailure(supervisorExit) && !Cause.hasInterruptsOnly(supervisorExit.cause)) {
+        if (cleanupCause === undefined) cleanupCause = supervisorExit.cause;
+        else cleanupCause = Cause.combine(cleanupCause)(supervisorExit.cause);
+      }
     }
     const currentRuntime = runtimeRef.current;
     let runtimeExit: RuntimeExit<S> = { _tag: "Stopped" };
@@ -1114,7 +1120,7 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* <
       // Cache publication must finish before a caller can cancel its wait.
       Effect.uninterruptible,
       Effect.flatMap((owner) => {
-        if (startupFiber?.id === caller.id) {
+        if (startupFiber?.id === caller.id || supervisorFiberRef.current?.id === caller.id) {
           // Mark the fiber itself. A cause-level fallback must not swallow self-stop.
           // Do not join self: recovery may be inside an uninterruptible region.
           return Effect.sync(() => caller.interruptUnsafe(caller.id));

test/stop-completion.test.ts

@@ -3,12 +3,113 @@
 import { Cause, Deferred, Effect, Exit, Fiber, Option, Schema, SubscriptionRef } from "effect";
 import { describe, expect, it, yieldFibers } from "effect-bun-test";
 
-import { Event, Machine, State } from "../src/index.js";
+import { Event, Machine, State, Supervision } from "../src/index.js";
 
 const LifecycleState = State({ Active: {} });
 const LifecycleEvent = Event({ Ping: {} });
 
 describe("actor stop completion", () => {
+  it.scopedLive("waits for supervised self-stop cleanup and preserves its defects", () =>
+    Effect.gen(function* () {
+      const stop = yield* Deferred.make<Effect.Effect<void>>();
+      const cleaning = yield* Deferred.make<void>();
+      const release = yield* Deferred.make<void>();
+      let starts = 0;
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Active,
+      }).spawn(LifecycleState.Active, () =>
+        Effect.suspend(() => {
+          starts++;
+          return Effect.die("restart for cleanup");
+        }),
+      );
+      const actor = yield* Machine.spawn(machine, {
+        supervision: Supervision.restart({ maxRestarts: 1 }),
+        lifecycle: {
+          recovery: {
+            resolve: ({ generation }) => {
+              if (generation === 0) return Effect.succeedNone;
+              return Deferred.await(stop).pipe(
+                Effect.flatMap((stopActor) => stopActor),
+                Effect.as(Option.none()),
+                Effect.catchCause(() => Effect.succeedNone),
+                Effect.ensuring(
+                  Deferred.succeed(cleaning, undefined).pipe(
+                    Effect.andThen(Deferred.await(release)),
+                    Effect.andThen(Effect.die("supervised recovery cleanup")),
+                  ),
+                ),
+              );
+            },
+          },
+        },
+      });
+      yield* Deferred.succeed(stop, actor.stop);
+      yield* actor.start;
+      yield* Deferred.await(cleaning);
+      const stopping = yield* actor.stop.pipe(Effect.forkScoped({ startImmediately: true }));
+      yield* yieldFibers;
+      const earlyExit = stopping.pollUnsafe();
+      yield* Deferred.succeed(release, undefined);
+      const stopped = yield* Fiber.await(stopping);
+      const repeated = yield* actor.stop.pipe(Effect.exit);
+      expect(earlyExit).toBeUndefined();
+      for (const exit of [stopped, repeated]) {
+        expect(exit._tag).toBe("Failure");
+        if (Exit.isFailure(exit)) {
+          expect(Cause.pretty(exit.cause)).toContain("supervised recovery cleanup");
+        }
+      }
+      expect(starts).toBe(1);
+      const terminal = yield* actor.awaitExit;
+      expect(terminal._tag).toBe("Defect");
+      if (terminal._tag === "Defect") expect(terminal.phase).toBe("cleanup");
+    }).pipe(Effect.timeout("2 seconds")),
+  );
+
+  it.scopedLive("can stop itself during protected supervised recovery", () =>
+    Effect.gen(function* () {
+      const stop = yield* Deferred.make<Effect.Effect<void>>();
+      const recovering = yield* Deferred.make<void>();
+      let starts = 0;
+      const machine = Machine.make({
+        state: LifecycleState,
+        event: LifecycleEvent,
+        initial: LifecycleState.Active,
+      }).spawn(LifecycleState.Active, () =>
+        Effect.suspend(() => {
+          starts++;
+          return Effect.die("restart for recovery");
+        }),
+      );
+      const actor = yield* Machine.spawn(machine, {
+        supervision: Supervision.restart({ maxRestarts: 1 }),
+        lifecycle: {
+          recovery: {
+            resolve: ({ generation }) => {
+              if (generation === 0) return Effect.succeedNone;
+              return Deferred.succeed(recovering, undefined).pipe(
+                Effect.andThen(Deferred.await(stop)),
+                Effect.flatMap((stopActor) => stopActor),
+                Effect.as(Option.none()),
+                Effect.uninterruptible,
+              );
+            },
+          },
+        },
+      });
+      yield* Deferred.succeed(stop, actor.stop);
+      yield* actor.start;
+      yield* Deferred.await(recovering);
+      const stopped = yield* actor.stop.pipe(Effect.timeout("1 second"), Effect.exit);
+      expect(stopped._tag).toBe("Success");
+      expect(starts).toBe(1);
+      expect((yield* actor.awaitExit)._tag).toBe("Defect");
+    }).pipe(Effect.timeout("2 seconds")),
+  );
+
   it.scopedLive("can stop after cancellation interrupts recovery self-stop", () =>
     Effect.gen(function* () {
       const stop = yield* Deferred.make<Effect.Effect<void>>();

test/supervision.test.ts

@@ -1,5 +1,5 @@
 // @effect-diagnostics strictEffectProvide:off - tests are entry points
-import { Deferred, Duration, Effect, Schema } from "effect";
+import { Cause, Deferred, Duration, Effect, Schema } from "effect";
 
 import {
   ActorSystemDefault,
@@ -34,6 +34,62 @@ const machine = Machine.make({ state: S, event: E, initial: S.Idle })
 // ============================================================================
 
 describe("supervision: restart on defect", () => {
+  it.scopedLive("exhausts the restart budget after repeated initial spawn defects", () =>
+    Effect.gen(function* () {
+      let attempts = 0;
+      const failingMachine = Machine.make({ state: S, event: E, initial: S.Idle }).spawn(
+        S.Idle,
+        () =>
+          Effect.suspend(() => {
+            attempts++;
+            return Effect.die("every initial spawn fails");
+          }),
+      );
+      const actor = yield* Machine.spawn(failingMachine, {
+        supervision: Supervision.restart({ maxRestarts: 2 }),
+      });
+      yield* actor.start;
+      const exit = yield* actor.awaitExit.pipe(Effect.timeout("1 second"));
+      expect(attempts).toBe(3);
+      expect(exit._tag).toBe("Defect");
+      if (exit._tag === "Defect") {
+        expect(exit.phase).toBe("initial-spawn");
+        expect(Cause.pretty(exit.cause)).toContain("every initial spawn fails");
+      }
+      expect(actor.client.getLifecycle()).toBe(exit);
+      yield* actor.stop;
+      yield* actor.stop;
+    }),
+  );
+
+  it.scopedLive("can recover after an initial spawn fails in a restarted generation", () =>
+    Effect.gen(function* () {
+      let attempts = 0;
+      const ready = yield* Deferred.make<void>();
+      const recoveringMachine = Machine.make({ state: S, event: E, initial: S.Idle })
+        .spawn(S.Idle, () =>
+          Effect.suspend(() => {
+            attempts++;
+            if (attempts < 3) return Effect.die("retry initial spawn");
+            return Deferred.succeed(ready, undefined);
+          }),
+        )
+        .on(S.Idle, E.Finish, () => S.Done)
+        .final(S.Done);
+      const actor = yield* Machine.spawn(recoveringMachine, {
+        supervision: Supervision.restart({ maxRestarts: 2 }),
+      });
+      yield* actor.start;
+      yield* Deferred.await(ready).pipe(Effect.timeout("1 second"));
+      yield* actor.call(E.Finish);
+      const exit = yield* actor.awaitExit;
+      expect(attempts).toBe(3);
+      expect(exit._tag).toBe("Final");
+      expect(actor.client.getLifecycle()).toBe(exit);
+      yield* actor.stop;
+    }),
+  );
+
   it.scopedLive("restarts after synchronous initial state Effect defect", () =>
     Effect.gen(function* () {
       const restarted = yield* Deferred.make<void>();