Unify invoked child ownership

Author: SandroMaglione

.changeset/compiled-machine-startup.md

@@ -2,4 +2,4 @@
 "@typeonce/effect-machine": patch
 ---
 
-Compile eligible machine initial-state normalization and reuse the validated startup configuration when initializing invoked children.
+Compile eligible machine initial-state normalization, reuse the validated startup configuration, and unify invoked-child ownership with the runtime child registry.

src/internal/machineProcess.ts

@@ -251,7 +251,6 @@ const makeChildlessCompiledDrain = (
 }
 
 class InvokeExecutionKernel {
-  readonly sessions: Map<string, InvokeSession>
   initialized = false
   initial:
     | {
@@ -266,7 +265,6 @@ class InvokeExecutionKernel {
     readonly activeConfiguration: Model.ActiveConfiguration
     readonly entryPaths: ReadonlyArray<string>
   }) {
-    this.sessions = new Map()
     this.initial = initial
   }
 
@@ -278,84 +276,46 @@ class InvokeExecutionKernel {
     return `Machine.invoke:${this.makeSessionKey(path, id)}`
   }
 
-  private stopSession(
-    scope: internalRuntime.ProcessScope<any>,
-    session: InvokeSession
-  ): Effect.Effect<void> {
-    return scope.stopChild(session.childId)
-  }
-
-  private remove(
-    scope: internalRuntime.ProcessScope<any>,
-    key: string,
-    token: symbol | undefined
-  ): Effect.Effect<void> {
-    return Effect.suspend(() => {
-      const current = this.sessions.get(key)
-      if (current === undefined || (token !== undefined && current.token !== token)) {
-        return Effect.void
-      }
-      this.sessions.delete(key)
-      return this.stopSession(scope, current)
-    })
-  }
-
-  stop(scope: internalRuntime.ProcessScope<any>, key: string): Effect.Effect<void> {
-    return this.remove(scope, key, undefined)
-  }
-
-  stopAll(scope: internalRuntime.ProcessScope<any>): Effect.Effect<void> {
-    return Effect.suspend(() => {
-      const effects = Array.from(this.sessions.values(), (session) => this.stopSession(scope, session))
-      this.sessions.clear()
-      return runParallelDiscard(effects)
-    })
-  }
-
   start(
     context: internalRuntime.CompiledProcessContext<any, any>,
     path: string,
     config: AnyInvokeConfig
   ): Effect.Effect<void, any, any> {
     return Effect.suspend(() => {
-      const token = Symbol()
       const invokeId = String(config.id)
       const key = this.makeSessionKey(path, invokeId)
       const childId = config.address === undefined ? this.makeChildId(path, invokeId) : String(config.address)
-      if (this.sessions.has(key)) {
-        return Effect.fail(new ChildAlreadyExistsError({ id: invokeId }))
-      }
-      this.sessions.set(key, { token, childId, path })
       const logic = config.src() as internalRuntime.ProcessLogic<any, any, any, any, any, any>
       const scope = context.scope
-      const sendParent = makeInvokeSendParent(this.sessions, scope.self, key, token)
-      return scope.spawn(
+      return context.ownedChildren.spawn(
         logic,
         {
+          key,
+          path,
           id: childId,
+          duplicateId: invokeId,
           ...(config.descriptor === undefined ? undefined : { descriptor: config.descriptor }),
-          [internalRuntime.sendParentOverride]: sendParent,
-          onOutcome: makeInvokeOutcomeHandler(
-            this.sessions,
-            scope.self,
-            scope.failCause,
-            config,
-            key,
-            token
-          ),
+          sendParent: (isCurrent, event) => isCurrent() ? scope.self.send(event) : Effect.void,
+          onOutcome: (isCurrent, outcome) => {
+            if (outcome._tag === "Stopped" || !isCurrent()) return Effect.void
+            if (outcome._tag !== "Done") return scope.failCause(outcome.cause)
+            const mappedEvent = config.onDone === undefined
+              ? outcome.output
+              : config.onDone({ id: config.id, output: outcome.output })
+            return mappedEvent === undefined
+              ? Effect.void
+              : scope.self.send(mappedEvent).pipe(Effect.catchTag("StoppedError", () => Effect.void))
+          },
           ...(config.snapshot === undefined ? undefined : {
-            [internalRuntime.activeSnapshotObserver]: makeInvokeSnapshotHandler(
-              this.sessions,
-              scope.self,
-              config,
-              key,
-              token
-            )
+            onSnapshot: (isCurrent: () => boolean, snapshot: any) => {
+              if (!isCurrent()) return Effect.void
+              const mappedEvent = config.snapshot!({ id: config.id, snapshot })
+              return mappedEvent === undefined
+                ? Effect.void
+                : scope.self.send(mappedEvent).pipe(Effect.catchTag("StoppedError", () => Effect.void))
+            }
           })
         }
-      ).pipe(
-        Effect.onExit((exit) => Exit.isFailure(exit) ? this.remove(scope, key, token) : Effect.void),
-        Effect.asVoid
       )
     })
   }
@@ -379,18 +339,6 @@ class InvokeExecutionKernel {
       )
     return effects.length === 0 ? undefined : runSequentialDiscard(effects)
   }
-
-  stopPaths(
-    machine: Machine.Any,
-    scope: internalRuntime.ProcessScope<any>,
-    paths: ReadonlyArray<string>
-  ): Effect.Effect<void> | undefined {
-    const keys = new Set(internalPlanner.sortExitPaths(machine, paths))
-    const effects = Array.from(this.sessions.entries())
-      .filter(([, session]) => keys.has(session.path))
-      .map(([key]) => this.stop(scope, key))
-    return effects.length === 0 ? undefined : runParallelDiscard(effects)
-  }
 }
 
 const makeInvokingCompiledDrain = (
@@ -464,7 +412,7 @@ const makeInvokingCompiledDrain = (
         beforeCommit.push(internalPlanner.runCommands(planned.commands, scope))
       }
       if (changed) {
-        const stopping = execution.stopPaths(machine, scope, exitPaths)
+        const stopping = context.ownedChildren.stopPaths(exitPaths)
         if (stopping !== undefined) beforeCommit.push(stopping)
       }
       const afterCommit: Array<Effect.Effect<void, any, any>> = []
@@ -477,7 +425,7 @@ const makeInvokingCompiledDrain = (
         )
       }
       if (planned.done) {
-        afterCommit.push(execution.stopAll(scope))
+        afterCommit.push(context.ownedChildren.stopAll())
       } else if (changed) {
         for (const [path, entryEvent] of entryEvents) {
           const starting = execution.startAll(machine, context, activeConfiguration, [path], entryEvent)