🤖 refactor: Effect Phase 11 PR 2 — EffectRunner + AppFiberScope runtime seams; TestClock on heartbeat/idleCompaction/retryManager

src/constants/terminationTimeouts.ts

@@ -15,3 +15,10 @@ export const BACKUP_GIT_TIMEOUT_MS = 5 * 60 * 1000;
  * `desktop/main.ts` and `cli/server.ts` race the whole dispose against.
  */
 export const APP_RUNTIME_DISPOSE_TIMEOUT_MS = 2 * 1000;
+
+/**
+ * Bounds the early `AppFiberScope` close in `ServiceContainer.dispose()`
+ * (interrupt + await of supervised fibers). Together with the runtime dispose
+ * bound this fits inside the same 5 s quit budgets.
+ */
+export const APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS = 2 * 1000;

src/node/services/di/appFiberScope.test.ts

@@ -0,0 +1,128 @@
+import { describe, expect, it, spyOn } from "bun:test";
+import { Effect, Exit, Fiber, Layer, Scope } from "effect";
+import { log } from "@/node/services/log";
+import { AppFiberScopeLive, AppFiberScopeTag } from "./appFiberScope";
+import { closeScopeBounded, disposeAppRuntime, makeAppRuntime } from "./appRuntime";
+import { EffectRunnerLive, EffectRunnerTag } from "./effectRunner";
+
+/** An interruptible I/O-style suspension that never resolves; records interruption. */
+function hungIo(onInterrupt: () => void): Effect.Effect<void> {
+  return Effect.callback<void>(() =>
+    Effect.sync(() => {
+      onInterrupt();
+    })
+  );
+}
+
+function buildSeams() {
+  const app = makeAppRuntime(AppFiberScopeLive.pipe(Layer.provideMerge(EffectRunnerLive)));
+  return {
+    app,
+    appFiberScope: app.get(AppFiberScopeTag),
+    runner: app.get(EffectRunnerTag),
+  };
+}
+
+describe("AppFiberScope", () => {
+  it("closeScopeBounded interrupts and awaits an I/O-suspended fiber forked into it", async () => {
+    const { app, appFiberScope } = buildSeams();
+    const steps: string[] = [];
+    const fiber = app.managed.runSync(
+      Effect.forkIn(
+        hungIo(() => steps.push("cancelled")).pipe(
+          Effect.ensuring(Effect.sync(() => steps.push("finalized")))
+        ),
+        appFiberScope
+      )
+    );
+
+    await closeScopeBounded(appFiberScope);
+
+    // Interrupted (the cancel path ran) and awaited (its finalizer ran before
+    // the close resolved), so shutdown can rely on the fiber being gone.
+    expect(steps).toEqual(["cancelled", "finalized"]);
+    expect(fiber.pollUnsafe()).toBeDefined();
+    expect(Exit.isFailure(fiber.pollUnsafe()!)).toBe(true);
+    await disposeAppRuntime(app.managed);
+  });
+
+  it("fibers forked through the EffectRunner are interrupted by neither close (unsupervised)", async () => {
+    const { app, appFiberScope, runner } = buildSeams();
+    let interrupted = false;
+    const fiber = runner.runFork(
+      hungIo(() => {
+        interrupted = true;
+      })
+    );
+
+    await closeScopeBounded(appFiberScope);
+    await disposeAppRuntime(app.managed);
+
+    // The runner is not a supervisor: its fibers belong to whoever forked them
+    // (a worker's own scope with an explicit stop()), so dispose leaves them be.
+    expect(interrupted).toBe(false);
+    expect(fiber.pollUnsafe()).toBeUndefined();
+    await Effect.runPromise(Fiber.interrupt(fiber));
+    expect(interrupted).toBe(true);
+  });
+
+  it("disposeAppRuntime re-closes an already-closed AppFiberScope idempotently", async () => {
+    const { app, appFiberScope } = buildSeams();
+    let finalized = 0;
+    app.managed.runSync(
+      Scope.addFinalizer(
+        appFiberScope,
+        Effect.sync(() => {
+          finalized += 1;
+        })
+      )
+    );
+
+    await closeScopeBounded(appFiberScope);
+    expect(finalized).toBe(1);
+
+    // The runtime's layer scope owns the child; closing the runtime afterwards
+    // must neither throw nor run the child's finalizers a second time.
+    await disposeAppRuntime(app.managed);
+    expect(finalized).toBe(1);
+    expect(app.managed.cachedContext).toBeUndefined();
+  });
+
+  it("runtime dispose alone closes the AppFiberScope (backstop for a missed explicit close)", async () => {
+    const { app, appFiberScope } = buildSeams();
+    let interrupted = false;
+    app.managed.runSync(
+      Effect.forkIn(
+        hungIo(() => {
+          interrupted = true;
+        }),
+        appFiberScope
+      )
+    );
+
+    await disposeAppRuntime(app.managed);
+
+    expect(interrupted).toBe(true);
+    expect(appFiberScope.state._tag).toBe("Closed");
+  });
+
+  it("closeScopeBounded returns at the timeout when a fiber cannot be interrupted, warning instead of rejecting", async () => {
+    const warnSpy = spyOn(log, "warn").mockImplementation(() => undefined);
+    try {
+      const { app, appFiberScope } = buildSeams();
+      // Uninterruptible and never-resolving: the scope close can never finish.
+      app.managed.runSync(Effect.forkIn(Effect.uninterruptible(Effect.never), appFiberScope));
+
+      const startedAt = Date.now();
+      await closeScopeBounded(appFiberScope, 50);
+
+      expect(Date.now() - startedAt).toBeLessThan(2_000);
+      expect(warnSpy).toHaveBeenCalledTimes(1);
+      expect(String(warnSpy.mock.calls[0]?.[0])).toContain("timed out");
+      // The runtime dispose then hits the same hung finalizer; keep it bounded too.
+      await disposeAppRuntime(app.managed, 50);
+    } finally {
+      warnSpy.mockRestore();
+    }
+  });
+});

src/node/services/di/appFiberScope.ts

@@ -0,0 +1,36 @@
+/**
+ * `AppFiberScope` — the runtime-owned, *supervised* fiber scope (Effect
+ * migration Phase 11, PR 2).
+ *
+ * A child of the app runtime's layer scope. Fibers forked into it with
+ * `Effect.forkIn(effect, appFiberScope)` are interrupted **and awaited** when
+ * it closes, which `ServiceContainer.dispose()` does explicitly and early
+ * (right after `backgroundProcessManager.beginShutdown()`, before the
+ * hand-ordered teardown steps) via `closeScopeBounded`, so interrupted fibers
+ * can still use their dependencies while they finalize. `runtime.dispose()`
+ * later re-closes it idempotently as a backstop.
+ *
+ * This is the seam for I/O-suspended, long-lived work that shutdown must wait
+ * for (the streamManager engine core, in a later phase). It is the counterpart
+ * of `EffectRunner` (`./effectRunner.ts`), which is unsupervised: a fiber forked
+ * through the runner is interrupted by neither close. Anything forked here must
+ * tolerate interruption at any suspension point and must not depend on
+ * resources torn down before the close (see the dispose order in
+ * `ServiceContainer`). No production occupant yet; the contract is pinned by
+ * tests.
+ */
+import { Context, Effect, Layer, Scope } from "effect";
+
+export class AppFiberScopeTag extends Context.Service<AppFiberScopeTag, Scope.Closeable>()(
+  "xum/AppFiberScope"
+) {}
+
+/**
+ * Synchronous body (DI contract: layers build without suspending). Parallel
+ * finalizer strategy: forked fibers are independent, so they are interrupted
+ * concurrently rather than one after another.
+ */
+export const AppFiberScopeLive: Layer.Layer<AppFiberScopeTag> = Layer.effect(
+  AppFiberScopeTag,
+  Effect.flatMap(Effect.scope, (parent) => Scope.fork(parent, "parallel"))
+);

src/node/services/di/appRuntime.ts

@@ -29,11 +29,20 @@
  *   `disposeAppRuntime` (the last dispose step) reorders nothing. It is wired
  *   now so that later phases (the streamManager engine core) have a fixed,
  *   bounded slot for scope-owned resources.
+ * - **Two runtime seams, not one handle.** Workers hold an `EffectRunner`
+ *   (`./effectRunner.ts`: context-bound, unsupervised, `R = never`) so they can
+ *   run on the runtime's `Clock` without ever holding the runtime; work that
+ *   shutdown must await forks into `AppFiberScope` (`./appFiberScope.ts`),
+ *   the one supervised resource, closed explicitly early in `dispose()` via
+ *   `closeScopeBounded` and re-closed idempotently by `disposeAppRuntime`.
  */
 import assert from "@/common/utils/assert";
-import { Context, Duration, Effect, Fiber, ManagedRuntime } from "effect";
+import { Context, Duration, Effect, Exit, Fiber, ManagedRuntime, Scope } from "effect";
 import type { Layer } from "effect";
-import { APP_RUNTIME_DISPOSE_TIMEOUT_MS } from "@/constants/terminationTimeouts";
+import {
+  APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS,
+  APP_RUNTIME_DISPOSE_TIMEOUT_MS,
+} from "@/constants/terminationTimeouts";
 import { log } from "@/node/services/log";
 
 export interface AppRuntime<R> {
@@ -79,37 +88,63 @@ export function disposeAppRuntime(
   runtime: ManagedRuntime.ManagedRuntime<never, never>,
   timeoutMs: number = APP_RUNTIME_DISPOSE_TIMEOUT_MS
 ): Promise<void> {
-  return Effect.runPromise(disposeAppRuntimeEffect(runtime, timeoutMs));
+  return Effect.runPromise(
+    boundedTeardown("AppRuntime", "disposed", runtime.disposeEffect, timeoutMs)
+  );
 }
 
-function disposeAppRuntimeEffect(
-  runtime: ManagedRuntime.ManagedRuntime<never, never>,
+/**
+ * Close a runtime-owned scope (`AppFiberScope`), interrupting and awaiting the
+ * fibers forked into it, bounded by `timeoutMs`. Same contract as
+ * `disposeAppRuntime`: never rejects, idempotent (`Scope.close` on a closed
+ * scope is a no-op), warns and returns at the bound if a fiber's finalization
+ * hangs.
+ */
+export function closeScopeBounded(
+  scope: Scope.Closeable,
+  timeoutMs: number = APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS
+): Promise<void> {
+  return Effect.runPromise(
+    boundedTeardown("AppFiberScope", "closed", Scope.close(scope, Exit.void), timeoutMs)
+  );
+}
+
+/**
+ * Uninterruptible teardown shell around a bounded wait on `target`. The target
+ * runs in a detached fiber and the caller waits on its join, so a hung
+ * finalizer cannot pin shutdown past the bound — only the wait is
+ * interruptible (so the timeout can win the race; house shape from #4038),
+ * the target keeps running best-effort, and a warning is logged. Defects are
+ * folded into a warning so the returned effect never fails.
+ */
+function boundedTeardown(
+  subject: string,
+  doneVerb: string,
+  target: Effect.Effect<void>,
   timeoutMs: number
 ): Effect.Effect<void> {
   const startedAt = performance.now();
-  // Uninterruptible teardown shell; only the bounded wait inside is interruptible
-  // so the timeout can win the race (house shape from #4038).
   return Effect.uninterruptible(
     Effect.gen(function* () {
-      const closing = yield* Effect.forkDetach(runtime.disposeEffect);
+      const closing = yield* Effect.forkDetach(target);
       yield* Effect.interruptible(
         Fiber.join(closing).pipe(Effect.timeout(Duration.millis(timeoutMs)))
       ).pipe(
         Effect.catchTag("TimeoutError", () =>
           Effect.sync(() => {
-            log.warn("[shutdown] AppRuntime dispose timed out; finalizers continue best-effort", {
+            log.warn(`[shutdown] ${subject} teardown timed out; finalizers continue best-effort`, {
               timeoutMs,
             });
           })
         )
       );
-      log.debug("[shutdown] AppRuntime disposed", {
+      log.debug(`[shutdown] ${subject} ${doneVerb}`, {
         ms: Math.round(performance.now() - startedAt),
       });
     }).pipe(
       Effect.catchDefect((defect) =>
         Effect.sync(() => {
-          log.warn("[shutdown] AppRuntime dispose failed", { error: defect });
+          log.warn(`[shutdown] ${subject} teardown failed`, { error: defect });
         })
       )
     )