🤖 feat: stream engine becomes the AppFiberScope occupant — dispose() aborts and awaits in-flight streams (Wave 4 PR 1)

src/node/services/coreServicesRoot.test.ts

@@ -4,6 +4,7 @@ import * as path from "path";
 import { afterEach, beforeEach, describe, expect, it, spyOn } from "bun:test";
 import type { Context } from "effect";
 import { createConfigStores, type ConfigStores } from "@/node/config";
+import { createRuntime } from "@/node/runtime/runtimeFactory";
 import * as agentPluginsMcpConfig from "@/node/services/agentPlugins/mcpConfig";
 import type { CoreServices } from "@/node/services/coreServices";
 import { AppFiberScopeTag } from "@/node/services/di/appFiberScope";
@@ -157,6 +158,96 @@ describe("createCoreServices", () => {
     // The afterEach pair then exercises the idempotent second close/dispose.
   });
 
+  it("the CLI cleanup list's appFiberScope.close aborts and awaits an in-flight stream", async () => {
+    // `xum run`/`xum workflow` mirror ServiceContainer.dispose(): the
+    // appFiberScope.close step runs before session.dispose. With the stream
+    // engine as the scope's occupant, that step must abort a live stream as
+    // "system", commit its partial into chat.jsonl and remove partial.json.
+    root = createCoreServices({
+      ...stores,
+      extensionMetadataPath: path.join(tempDir, "extensionMetadata.json"),
+    });
+    const workspaceId = "cli-cleanup-in-flight-stream-workspace";
+    const messageId = "cli-cleanup-in-flight-stream-message";
+    const abortReasons: string[] = [];
+    Reflect.set(root.streamManager, "tokenTracker", {
+      setModel: () => Promise.resolve(undefined),
+      countTokens: () => Promise.resolve(0),
+    });
+    Reflect.set(
+      root.streamManager,
+      "createStreamResult",
+      (_request: unknown, abortController: AbortController) => ({
+        fullStream: (async function* () {
+          yield { type: "text-delta", text: "cli stream text" };
+          await new Promise<void>((resolve) => {
+            if (abortController.signal.aborted) return resolve();
+            abortController.signal.addEventListener("abort", () => resolve(), { once: true });
+          });
+        })(),
+        totalUsage: Promise.resolve(undefined),
+        usage: Promise.resolve(undefined),
+        providerMetadata: Promise.resolve(undefined),
+        steps: Promise.resolve([]),
+      })
+    );
+    Reflect.set(root.streamManager, "createTempDirForStream", () =>
+      Promise.resolve(path.join(tempDir, "stream-tempdir"))
+    );
+    Reflect.set(root.streamManager, "cleanupStreamTempDir", () => undefined);
+    root.aiService.on("stream-abort", (event: { abortReason?: string }) => {
+      abortReasons.push(event.abortReason ?? "");
+    });
+    const appendResult = await root.historyService.appendToHistory(workspaceId, {
+      id: messageId,
+      role: "assistant",
+      metadata: { historySequence: 1, partial: true },
+      parts: [],
+    });
+    expect(appendResult.success).toBe(true);
+    const started = await root.streamManager.startStream({
+      workspaceId,
+      messageId,
+      model: {
+        specificationVersion: "v3",
+        provider: "test",
+        modelId: "cli-cleanup-model",
+        supportedUrls: {},
+        doGenerate: () => Promise.reject(new Error("unused")),
+        doStream: () => Promise.reject(new Error("unused")),
+      },
+      messages: [{ role: "user", content: "hello" }],
+      modelString: "openai:gpt-4.1-mini",
+      historySequence: 1,
+      system: "system",
+      runtime: createRuntime({ type: "local", srcBaseDir: tempDir }),
+      providedRuntimeTempDir: "",
+    });
+    expect(started.success).toBe(true);
+    if (!started.success) throw new Error("expected the stream to start");
+    const deadline = Date.now() + 5_000;
+    while ((await root.historyService.readPartial(workspaceId)) === null) {
+      if (Date.now() > deadline) throw new Error("partial never written");
+      await new Promise((resolve) => setTimeout(resolve, 5));
+    }
+
+    await closeScopeBounded(root.appFiberScope);
+
+    expect(abortReasons).toEqual(["system"]);
+    expect(await started.data.completion).toEqual({ status: "aborted", abortReason: "system" });
+    expect(root.streamManager.isStreaming(workspaceId)).toBe(false);
+    expect(await root.historyService.readPartial(workspaceId)).toBeNull();
+    const history = await root.historyService.getHistoryFromLatestBoundary(workspaceId);
+    expect(history.success).toBe(true);
+    if (!history.success) throw new Error(history.error);
+    const committed = history.data.find((message) => message.id === messageId);
+    expect(
+      committed?.parts.some((part) => part.type === "text" && part.text === "cli stream text")
+    ).toBe(true);
+    // Dependencies are still alive for the remaining CLI cleanup steps.
+    expect(root.runtime.managed.cachedContext).toBeDefined();
+  });
+
   it("surfaces a throwing layer body as a synchronous throw", () => {
     // A throw deep inside a nested stage (MCPConfigLive, S4) must propagate
     // through the staged composition as the same synchronous throw a service

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

@@ -106,6 +106,79 @@ describe("AppFiberScope", () => {
     expect(appFiberScope.state._tag).toBe("Closed");
   });
 
+  it("interrupts a fiber suspended on Effect.promise and awaits its onInterrupt finalizer before the close resolves", async () => {
+    // The stream engine supervisor's shape (streamManager.ts superviseEngine):
+    // a fiber that suspends on a plain Promise and finalizes through another
+    // async Promise. The close must (1) interrupt the suspended wait without
+    // settling the wrapped promise, (2) run the finalizer to completion, and
+    // (3) resolve only afterwards.
+    const { app, appFiberScope } = buildSeams();
+    const steps: string[] = [];
+    let resolveWork!: () => void;
+    const work = new Promise<void>((resolve) => {
+      resolveWork = resolve;
+    });
+    app.managed.runSync(
+      Effect.forkIn(
+        Effect.promise(() => work).pipe(
+          Effect.onInterrupt(() =>
+            Effect.uninterruptible(
+              Effect.promise(async () => {
+                steps.push("finalizer-start");
+                await new Promise<void>((resolve) => setTimeout(resolve, 10));
+                steps.push("finalizer-end");
+              })
+            )
+          )
+        ),
+        appFiberScope,
+        { startImmediately: true }
+      )
+    );
+
+    await closeScopeBounded(appFiberScope);
+
+    expect(steps).toEqual(["finalizer-start", "finalizer-end"]);
+    // The wrapped promise itself is untouched by the interruption (the
+    // occupant's own cancellation transport decides when it settles).
+    let workSettled = false;
+    void work.then(() => {
+      workSettled = true;
+    });
+    await new Promise<void>((resolve) => setTimeout(resolve, 0));
+    expect(workSettled).toBe(false);
+    resolveWork();
+    await disposeAppRuntime(app.managed);
+  });
+
+  it("a fiber forked with startImmediately into an already-closed scope still runs its onInterrupt finalizer", async () => {
+    // Pins the rc.112 forkIn semantics the supervisor relies on for streams
+    // that start mid-shutdown: the body runs synchronously up to its first
+    // async boundary, the closed scope interrupts it right there, and the
+    // interruption unwinds through onInterrupt — so such a stream is aborted
+    // rather than left running unsupervised.
+    const { app, appFiberScope } = buildSeams();
+    await closeScopeBounded(appFiberScope);
+    expect(appFiberScope.state._tag).toBe("Closed");
+
+    const steps: string[] = [];
+    const fiber = app.managed.runSync(
+      Effect.forkIn(
+        Effect.promise(() => {
+          steps.push("body-started");
+          return new Promise<void>(() => undefined);
+        }).pipe(Effect.onInterrupt(() => Effect.sync(() => steps.push("interrupted")))),
+        appFiberScope,
+        { startImmediately: true }
+      )
+    );
+
+    expect(steps).toEqual(["body-started", "interrupted"]);
+    expect(fiber.pollUnsafe()).toBeDefined();
+    expect(Exit.isFailure(fiber.pollUnsafe()!)).toBe(true);
+    await disposeAppRuntime(app.managed);
+  });
+
   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 {

src/node/services/di/appFiberScope.ts

@@ -11,13 +11,15 @@
  * 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.
+ * for. Its occupant is the stream engine: `StreamManager.superviseEngine`
+ * forks one supervisor fiber per stream into it, whose interruption cancels the
+ * stream (`"system"` abort) and awaits the turn's settlement, so dispose()
+ * commits the partial into chat.jsonl before the bridges stop. 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`). The contract is pinned by tests.
  */
 import { Context, Effect, Layer, Scope } from "effect";
 

src/node/services/di/appRuntime.ts

@@ -85,11 +85,16 @@
  * - `AppFiberScope` is **supervised**: a child of the runtime's layer scope;
  *   fibers forked into it with `Effect.forkIn` are interrupted *and awaited*
  *   by `closeScopeBounded` early in `dispose()`, while every dependency they
- *   might touch during finalization is still alive. No production occupant in
- *   Phase 11; the first candidate is the streamManager engine core. **Rule for
- *   occupants:** tolerate interruption at any suspension point, do not depend
- *   on resources torn down before step 2 below, and never fork long-lived I/O
- *   work through `EffectRunner` expecting shutdown to await it.
+ *   might touch during finalization is still alive. Occupant: the stream
+ *   engine (`StreamManager.superviseEngine`, Wave 4 PR 1) — one supervisor
+ *   fiber per stream, forked at registration and living until the turn's
+ *   completion settles; the stream's AbortSignal stays the cancellation
+ *   transport and interruption
+ *   routes through the user-stop path (`"system"` abort, partial committed,
+ *   `completion` settled) before the close resolves. **Rule for occupants:**
+ *   tolerate interruption at any suspension point, do not depend on resources
+ *   torn down before step 2 below, and never fork long-lived I/O work through
+ *   `EffectRunner` expecting shutdown to await it.
  *
  * ## Shutdown order (`ServiceContainer.dispose()`, one shared teardown behind
  * a latch so concurrent/repeated calls — the desktop's two `before-quit`
@@ -101,7 +106,10 @@
  *    chat session against further dispatch (`workspaceService.beginShutdown()`,
  *    which also disposes the transient chat-recovery sessions housekeeping
  *    scheduled), and only then bounded-join the housekeeping.
- * 2. `closeScopeBounded(appFiberScope, APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS)`.
+ * 2. `closeScopeBounded(appFiberScope, APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS)` —
+ *    aborts (`"system"`) and awaits every in-flight stream; a flowing stream
+ *    settles within one chunk, a wedged provider (no chunks, ignores abort)
+ *    hits the bound, warns, and the process still exits.
  * 3. The explicit sequence verbatim (`desktopBridgeServer.stop()` …
  *    `terminateAll()` … `timelineService.flush()` last), each step timed as a
  *    `[shutdown] <step> {ms}` debug line (`shutdownStep.ts`).
@@ -137,8 +145,10 @@
  * `initialize()` as a Layer/startup effect (would break I1's failure
  * semantics), layer finalizers for the existing `dispose()` steps (I5),
  * `streamBridge` on the runtime, per-service optional tags (optional
- * cross-cutting services stay optional via `CoreOptionsTag`), the streamManager
- * engine core as the first `AppFiberScope` occupant.
+ * cross-cutting services stay optional via `CoreOptionsTag`). The streamManager
+ * engine core became the `AppFiberScope` occupant in Wave 4 PR 1; the
+ * pre-registration stream-start window (`pendingStreamStarts`) stays
+ * unsupervised (nothing durable exists for it yet).
  */
 import assert from "@/common/utils/assert";
 import { Context, Duration, Effect, Exit, Fiber, ManagedRuntime, Scope } from "effect";

src/node/services/di/layers/core.ts

@@ -12,7 +12,7 @@ import {
 import { AIService } from "@/node/services/aiService";
 import { BackgroundProcessManager } from "@/node/services/backgroundProcessManager";
 import type { CoreOptions, CoreServices, CoreServicesOptions } from "@/node/services/coreServices";
-import { AppFiberScopeLive } from "@/node/services/di/appFiberScope";
+import { AppFiberScopeLive, AppFiberScopeTag } from "@/node/services/di/appFiberScope";
 import { EffectRunnerLive, EffectRunnerTag } from "@/node/services/di/effectRunner";
 import {
   AI,
@@ -90,16 +90,18 @@ export class CoreOptionsTag extends Context.Service<CoreOptionsTag, CoreOptions>
 
 /**
  * What the roots must provide beneath `CoreLive`: the stores, the options,
- * the runtime's `EffectRunner` (the base seam in both roots; StreamManager's
- * clock-driven fibers run through it), and the two always-present
- * collaborators the desktop builds elsewhere (`MemoryMetaLive`;
- * `WorkspaceMcpOverrides` from `CrossCuttingLive`). CLI roots supply the
- * defaults (`MemoryMetaLive`, `WorkspaceMcpOverridesDefaultLive`).
+ * the runtime seams (the base of both roots: `EffectRunner`, through which
+ * StreamManager's clock-driven fibers run, and `AppFiberScope`, which
+ * supervises its stream engine), and the two always-present collaborators the
+ * desktop builds elsewhere (`MemoryMetaLive`; `WorkspaceMcpOverrides` from
+ * `CrossCuttingLive`). CLI roots supply the defaults (`MemoryMetaLive`,
+ * `WorkspaceMcpOverridesDefaultLive`).
  */
 export type CoreInputTags =
   | StoreTags
   | CoreOptionsTag
   | EffectRunnerTag
+  | AppFiberScopeTag
   | MemoryMeta
   | WorkspaceMcpOverrides;
 
@@ -233,7 +235,11 @@ export const StreamManagerLive = Layer.effect(
       () => providerService.getConfig(),
       // Default event sink: AIService installs itself as the sink (S3).
       undefined,
-      yield* EffectRunnerTag
+      yield* EffectRunnerTag,
+      // The stream engine is the AppFiberScope's occupant: dispose() closes the
+      // scope before the explicit teardown steps, which aborts and awaits every
+      // in-flight stream (StreamManager.superviseEngine).
+      yield* AppFiberScopeTag
     );
   })
 );