Merge pull request #65 from cevr/fix/machine-ref-client

feat(machine): expose synchronous self client

Author: cevr

.changeset/quiet-callbacks-observe.md

@@ -0,0 +1,5 @@
+---
+"effect-machine": minor
+---
+
+Expose a narrow synchronous client to machine-owned Effects. Bind each background resource to its actor generation.

README.md

@@ -96,8 +96,9 @@ Effect Machine does not add an action queue or a second context system.
 Effect requirements remain in `R`. A machine cannot start until the application provides every required service. Effectful transition handlers must have `never` in their error channel. Convert expected failures to states or events.
 
 Machine-lifetime backgrounds can read `self.state` and `self.latestTransition`. These are the
-actor-owned subscription refs. They stay stable across supervision generations. Treat them as
-read-only and use `SubscriptionRef.get` or `SubscriptionRef.changes` to observe them.
+actor-owned subscription refs. They stay stable across supervision generations. Use
+`SubscriptionRef.get` or `SubscriptionRef.changes` for Effect workflows. Use `self.client` for
+synchronous host callbacks that need `send`, `getSnapshot`, or `subscribe`.
 
 Read [the Effect model](./docs/effect-model.md) and [async work ownership](./docs/async-work.md).
 

SKILL.md

@@ -52,6 +52,10 @@ const machine = Machine.make({
 | `.background(handler)`                 | Machine-lifetime effect                 |
 | `.final(state)`                        | Mark final state                        |
 
+Use `self.client` in a host callback that must send or observe state synchronously. Use
+`self.state` and `self.latestTransition` in Effect workflows. A background resource receives the
+actor generation Scope. The runtime closes that Scope when the actor generation stops.
+
 ## State.with()
 
 Construct state from existing source:

src/actor.ts

@@ -43,6 +43,7 @@ import type { NoReplyError } from "./errors.js";
 import { DuplicateActorError, ActorStoppedError } from "./errors.js";
 import {
   createRuntime,
+  notifyStateListeners,
   type RuntimeLifecycleHooks,
   type RuntimeQueuedEvent,
   type RuntimeHandle,
@@ -408,19 +409,6 @@ export class ActorScope extends Context.Service<ActorScope, Scope.Scope>()(
 /** Listener set for sync subscriptions */
 type Listeners<S> = Set<(state: S) => void>;
 
-/**
- * Notify all listeners of state change.
- */
-const notifyListeners = <S>(listeners: Listeners<S>, state: S): void => {
-  for (const listener of listeners) {
-    try {
-      listener(state);
-    } catch {
-      // Ignore listener failures to avoid crashing the actor loop
-    }
-  }
-};
-
 const toActorExit = <S, O>(
   machine: { readonly _output: (state: S) => O },
   exit: RuntimeExit<S>,
@@ -788,7 +776,7 @@ const runSupervisionLoop = <
         yield* options.onRestart(nextGeneration, generationExit);
       }
 
-      notifyListeners(cell.listeners, restartState);
+      notifyStateListeners(cell.listeners, restartState);
     }
   });
 
@@ -913,7 +901,6 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* <
       onEvent,
       onStateChange: (result, event) =>
         Effect.gen(function* () {
-          notifyListeners(listeners, result.newState);
           const durability = lifecycle?.durability;
           if (durability === undefined || !result.transitioned) return;
           const shouldPersist =
@@ -974,6 +961,7 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* <
               latestTransitionRef,
               stoppedRef,
               eventQueue: currentQueue,
+              listeners,
             },
             lifecycle: buildRuntimeLifecycle(runtimeGeneration),
             onChildSpawned: (childId, child) =>

src/cluster/entity-machine.ts

@@ -188,6 +188,7 @@ export const EntityMachine = {
       );
       const stoppedRef = yield* Ref.make(false);
       const eventQueue = yield* Queue.unbounded<RuntimeQueuedEvent<S, E>>();
+      const listeners = new Set<(state: S) => void>();
       let hooks: ReturnType<typeof makeInspectionHooks<S, E>> | undefined = undefined;
       if (inspector !== undefined) {
         hooks = makeInspectionHooks(entityId, inspector);
@@ -199,7 +200,7 @@ export const EntityMachine = {
           actorId: entityId,
           hooks,
           childIdPrefix: `${entityId}/`,
-          cellResources: { stateRef, latestTransitionRef, stoppedRef, eventQueue },
+          cellResources: { stateRef, latestTransitionRef, stoppedRef, eventQueue, listeners },
         }),
       );
       yield* withActorInspection(runtime.start);

src/internal/runtime.ts

@@ -29,6 +29,7 @@ import {
   Scope,
   SubscriptionRef,
 } from "effect";
+import { dual } from "effect/Function";
 
 import type { Machine, MachineRef } from "../machine.js";
 import type { ActorRef, ActorSystemService, TransitionInfo } from "../actor.js";
@@ -81,8 +82,25 @@ export interface RuntimeCellResources<S, E> {
   readonly latestTransitionRef: SubscriptionRef.SubscriptionRef<TransitionInfo<S, E> | undefined>;
   readonly eventQueue: Queue.Queue<RuntimeQueuedEvent<S, E>>;
   readonly stoppedRef: Ref.Ref<boolean>;
+  readonly listeners: Set<(state: S) => void>;
 }
 
+/** Notify synchronous state listeners without letting a listener defect fail the runtime. */
+export const notifyStateListeners: {
+  <S>(state: S): (listeners: Set<(state: S) => void>) => void;
+  <S>(listeners: Set<(state: S) => void>, state: S): void;
+} = dual(2, <S>(listeners: Set<(state: S) => void>, state: S): void => {
+  for (const listener of listeners) {
+    // Host callbacks can throw. Each listener must stay isolated from the actor runtime.
+    // eslint-disable-next-line effect/noTryCatch
+    try {
+      listener(state);
+    } catch {
+      // Listener defects must not fail the runtime.
+    }
+  }
+});
+
 // ============================================================================
 // Runtime interface
 // ============================================================================
@@ -224,9 +242,17 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
   const services = yield* Effect.context<R>();
   const fork = Effect.runForkWith(services);
 
-  const { stateRef, latestTransitionRef, stoppedRef, eventQueue } = config.cellResources;
+  const { stateRef, latestTransitionRef, stoppedRef, eventQueue, listeners } = config.cellResources;
   const pendingRequests = new Set<(error: ActorStoppedError) => Effect.Effect<void>>();
 
+  const loopFiberRef: { current: Fiber.Fiber<void> | undefined } = { current: undefined };
+  const sendSync = (event: E): void => {
+    if (Ref.getUnsafe(stoppedRef)) return;
+    Queue.offerUnsafe(eventQueue, { _tag: "send", event });
+    eventQueue.dispatcher.flush();
+    loopFiberRef.current?.currentDispatcher?.flush();
+  };
+
   // Exit deferred — set exactly once with the exit reason
   const exitDeferred = yield* Deferred.make<RuntimeExit<S>>();
 
@@ -265,6 +291,14 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
     state: stateRef,
     latestTransition: latestTransitionRef,
     send: selfSend,
+    client: {
+      send: sendSync,
+      getSnapshot: () => SubscriptionRef.getUnsafe(stateRef),
+      subscribe: (listener) => {
+        listeners.add(listener);
+        return () => listeners.delete(listener);
+      },
+    },
     spawn,
     reply: (value: unknown) =>
       Effect.sync(() => {
@@ -285,9 +319,6 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
 
   // Shared mutable refs used by both start() and stop()
   const initEvent = { _tag: INTERNAL_INIT_EVENT } as E;
-  // Mutable holder for the loop fiber — needed by stop() and spawn defect signals
-  const loopFiberRef: { current: Fiber.Fiber<void> | undefined } = { current: undefined };
-
   /** Set the exit deferred exactly once. */
   const setExit = (exit: RuntimeExit<S>) =>
     Deferred.succeed(exitDeferred, exit).pipe(Effect.asVoid);
@@ -351,6 +382,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
           event: latest.event,
         });
       }
+      notifyStateListeners(listeners, initialResult.newState);
       if (lifecycle?.onStateChange !== undefined) {
         const stateChange = lifecycle.onStateChange(initialResult, initEvent);
         if (isEffect(stateChange)) yield* stateChange;
@@ -375,7 +407,10 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
           self,
           system,
         })
-        .pipe(Effect.forkIn(actorScope, { startImmediately: true }));
+        .pipe(
+          Effect.provideService(Scope.Scope, actorScope),
+          Effect.forkIn(actorScope, { startImmediately: true }),
+        );
       backgroundFibers.push(fiber);
     }
 
@@ -443,6 +478,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
       eventQueue,
       pendingRequests,
       stoppedRef,
+      listeners,
       self,
       stateScopeRef,
       actorId,
@@ -537,7 +573,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
       eventQueue,
       pendingRequests,
       exitDeferred,
-      () => loopFiberRef.current?.currentDispatcher?.flush(),
+      sendSync,
     ),
     stop: stop.pipe(Effect.provide(services)),
     start: start.pipe(Effect.provide(services)),
@@ -556,7 +592,7 @@ const makeHandle = <S extends { readonly _tag: string }, E extends { readonly _t
   eventQueue: Queue.Queue<RuntimeQueuedEvent<S, E>>,
   pendingRequests: Set<(error: ActorStoppedError) => Effect.Effect<void>>,
   exitDeferred: Deferred.Deferred<RuntimeExit<S>>,
-  flushLoop: () => void,
+  sendSync: (event: E) => void,
 ): RuntimeHandle<S, E> => {
   const track = <A, RequestError>(
     deferred: Deferred.Deferred<A, RequestError>,
@@ -612,13 +648,7 @@ const makeHandle = <S extends { readonly _tag: string }, E extends { readonly _t
       yield* Queue.offer(eventQueue, { _tag: "drain", done });
       yield* Deferred.await(done);
     }).pipe(Effect.asVoid),
-    sendSync: (event: E) => {
-      const stopped = Effect.runSync(Ref.get(stoppedRef));
-      if (stopped) return;
-      Effect.runSync(Queue.offer(eventQueue, { _tag: "send", event }));
-      eventQueue.dispatcher.flush();
-      flushLoop();
-    },
+    sendSync,
     getState: SubscriptionRef.get(stateRef),
     stateRef,
     latestTransitionRef,
@@ -655,6 +685,7 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function*
   eventQueue: Queue.Queue<RuntimeQueuedEvent<S, E>>,
   pendingRequests: Set<(error: ActorStoppedError) => Effect.Effect<void>>,
   stoppedRef: Ref.Ref<boolean>,
+  listeners: Set<(state: S) => void>,
   self: MachineRef<E, S>,
   stateScopeRef: { current: Scope.Closeable },
   actorId: string,
@@ -761,6 +792,7 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function*
             generation,
           );
         }
+        notifyStateListeners(listeners, result.newState);
       }
 
       // Lifecycle: onStateChange (actor notifies listeners and saves durability)

src/machine.ts

@@ -56,7 +56,7 @@ import {
   type MachineInitialization as MachineInitializationType,
 } from "./internal/machine-initialization.js";
 import type { BackgroundEffect, SpawnEffect, Transition } from "./internal/machine-definition.js";
-import type { ActorRef, ActorSystemService, TransitionInfo } from "./actor.js";
+import type { ActorClient, ActorRef, ActorSystemService, TransitionInfo } from "./actor.js";
 import type { InspectorService } from "./inspection.js";
 
 // ============================================================================
@@ -74,6 +74,11 @@ export interface MachineRef<Event, State = never> {
     TransitionInfo<State, Event> | undefined
   >;
   readonly send: (event: Event) => Effect.Effect<void>;
+  /** Synchronous client for host callback boundaries. */
+  readonly client: Pick<
+    ActorClient<State & { readonly _tag: string }, Event>,
+    "send" | "getSnapshot" | "subscribe"
+  >;
   readonly spawn: <S2 extends { readonly _tag: string }, E2 extends { readonly _tag: string }, R2>(
     id: string,
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -1091,6 +1096,7 @@ export class Machine<
 
   /**
    * Machine-lifetime effect that is forked on actor spawn and runs until the actor stops.
+   * Each handler receives the generation Scope. It closes when the generation stops.
    *
    * @example
    * ```ts

test/machine-self-observation.test.ts

@@ -31,6 +31,64 @@ interface SelfSource {
 }
 
 describe("machine self observation", () => {
+  it.scopedLive("provides an inline host client to machine-owned effects", () =>
+    Effect.gen(function* () {
+      const observedStates: Array<TestState["_tag"]> = [];
+      let defectiveListenerCalls = 0;
+      const released = yield* Deferred.make<void>();
+      const machine = Machine.make({
+        state: TestState,
+        event: TestEvent,
+        initial: TestState.Idle,
+      })
+        .on(TestState.Idle, TestEvent.Start, ({ event }) =>
+          TestState.Active({ count: event.count }),
+        )
+        .on(TestState.Active, TestEvent.Increment, ({ state }) =>
+          TestState.Active({ count: state.count + 1 }),
+        )
+        .background(({ self }) =>
+          Effect.acquireRelease(
+            Effect.sync(() =>
+              self.client.subscribe((state) => {
+                const exactState: TestState = state;
+                observedStates.push(exactState._tag);
+                if (TestState.$is("Active")(exactState) && exactState.count === 1) {
+                  self.client.send(TestEvent.Increment);
+                }
+              }),
+            ),
+            (unsubscribe) =>
+              Effect.sync(unsubscribe).pipe(Effect.andThen(Deferred.succeed(released, undefined))),
+          ).pipe(Effect.andThen(Effect.never)),
+        )
+        .background(({ self }) =>
+          Effect.acquireRelease(
+            Effect.sync(() =>
+              self.client.subscribe(() => {
+                defectiveListenerCalls += 1;
+                // This host callback must throw to test defect isolation.
+                // eslint-disable-next-line effect/noThrowStatement, effect/noNewError
+                throw new Error("listener defect");
+              }),
+            ),
+            (unsubscribe) => Effect.sync(unsubscribe),
+          ).pipe(Effect.andThen(Effect.never)),
+        );
+
+      const actor = yield* Machine.spawn(machine);
+      yield* actor.start;
+
+      actor.client.send(TestEvent.Start({ count: 1 }));
+
+      expect(actor.client.getSnapshot()).toEqual(TestState.Active({ count: 2 }));
+      expect(observedStates).toEqual(["Active", "Active"]);
+      expect(defectiveListenerCalls).toBe(2);
+      yield* actor.stop;
+      yield* Deferred.await(released);
+    }),
+  );
+
   it.scopedLive("shares typed actor refs for accepted transitions and stop cleanup", () =>
     Effect.gen(function* () {
       const sourceReady = yield* Deferred.make<SelfSource>();