From 782aedaca2414bc6107dac5453490c7be4803367 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 3 Sep 2026 17:30:17 +0000 Subject: [PATCH 1/9] test(di): pin Effect.promise interruption + closed-scope forkIn semantics for the engine supervisor --- src/node/services/di/appFiberScope.test.ts | 73 ++++++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/src/node/services/di/appFiberScope.test.ts b/src/node/services/di/appFiberScope.test.ts index 94e852c642..ea7ffbecd1 100644 --- a/src/node/services/di/appFiberScope.test.ts +++ b/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((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((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((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(() => 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 { From 0449c8ca5b1f71cb21f5378739ee8471c92fd659 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 3 Sep 2026 17:50:48 +0000 Subject: [PATCH 2/9] feat(streamManager): supervise the stream engine in AppFiberScope; latch cancels; guard completed streams from abort bookkeeping --- src/node/services/di/appFiberScope.ts | 16 +-- src/node/services/di/appRuntime.ts | 25 +++-- src/node/services/di/layers/core.ts | 20 ++-- src/node/services/serviceContainer.ts | 11 +- src/node/services/streamManager.ts | 141 +++++++++++++++++++++++--- 5 files changed, 172 insertions(+), 41 deletions(-) diff --git a/src/node/services/di/appFiberScope.ts b/src/node/services/di/appFiberScope.ts index 2d3302cf51..4d2a9d30ba 100644 --- a/src/node/services/di/appFiberScope.ts +++ b/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"; diff --git a/src/node/services/di/appRuntime.ts b/src/node/services/di/appRuntime.ts index b191c676c0..d3bb44ecda 100644 --- a/src/node/services/di/appRuntime.ts +++ b/src/node/services/di/appRuntime.ts @@ -85,11 +85,15 @@ * - `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 wrapping the already-running processing promise; 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 +105,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] {ms}` debug line (`shutdownStep.ts`). @@ -137,8 +144,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"; diff --git a/src/node/services/di/layers/core.ts b/src/node/services/di/layers/core.ts index a1ba9b894f..22f2699c13 100644 --- a/src/node/services/di/layers/core.ts +++ b/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 /** * 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 ); }) ); diff --git a/src/node/services/serviceContainer.ts b/src/node/services/serviceContainer.ts index c7518ce9e3..1234247a58 100644 --- a/src/node/services/serviceContainer.ts +++ b/src/node/services/serviceContainer.ts @@ -589,10 +589,13 @@ export class ServiceContainer { } }); } - // Interrupt and await the runtime's supervised fibers while every dependency - // they might touch during finalization is still alive. Fixed here (before - // the explicit teardown) so later occupants do not re-derive the position; - // bounded and idempotent, and never rejects (di/appRuntime.ts). + // Interrupt and await the runtime's supervised fibers — the stream engine's + // per-stream supervisors (StreamManager.superviseEngine): every in-flight + // stream is aborted as "system" and its partial committed to chat.jsonl — + // while every dependency they touch during finalization is still alive. + // Fixed here (before the explicit teardown) so clients still receive the + // stream-abort over the bridges; bounded and idempotent, and never rejects + // (di/appRuntime.ts). await closeScopeBounded(this.appFiberScope); // Stop the bridge before closing sessions so desktop clients get a clean disconnect. await shutdownStep("desktopBridgeServer.stop", () => this.desktopBridgeServer.stop()); diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index d12db74e0a..545ac665d1 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -697,6 +697,13 @@ interface WorkspaceStreamInfo { partialWritePromise?: Promise; // Track background processing promise for guaranteed cleanup processingPromise: Promise; + // Latched by the first cancelStreamSafely call (synchronously, before its + // first await) so concurrent cancellers — a user stop racing shutdown's + // engine supervisor — join one cleanup: exactly one stream-abort, one settle. + cancelPromise?: Promise; + // Supervisor fiber in the app's AppFiberScope (superviseEngine); set only when + // the manager was constructed with an engine scope. + engineFiber?: Fiber.Fiber; // Soft-interrupt state: when pending, stream will end at next block boundary softInterrupt: | { pending: false } @@ -787,6 +794,18 @@ export class StreamManager { * and the global runtime wherever nothing is injected (di/effectRunner.ts). */ public readonly effectRunner: EffectRunner; + /** + * Supervised scope for the stream engine (the app runtime's `AppFiberScope`, + * di/appFiberScope.ts). When set, every started stream is wrapped in a + * supervisor fiber forked here (`superviseEngine`), so + * `ServiceContainer.dispose()`/the CLI cleanup lists interrupt **and await** + * in-flight streams — aborted as `"system"`, partial flushed and committed — + * before the bridges and sessions are torn down. `undefined` (direct + * construction in tests, `aiService.ts` compat path) keeps streams + * unsupervised: they die with the process and are recovered from + * partial.json on next load. + */ + private readonly engineScope?: Scope.Closeable; // Token tracker for live streaming statistics private tokenTracker = new StreamingTokenTracker(); // Track OpenAI previousResponseIds that have been invalidated @@ -803,13 +822,15 @@ export class StreamManager { sessionUsageService?: SessionUsageService, getProvidersConfig?: () => ProvidersConfigMap | null, eventSink: TurnEngineEventSink = () => undefined, - runner: EffectRunner = defaultEffectRunner + runner: EffectRunner = defaultEffectRunner, + engineScope?: Scope.Closeable ) { this.historyService = historyService; this.sessionUsageService = sessionUsageService; this.getProvidersConfig = getProvidersConfig ?? (() => null); this.eventSink = eventSink; this.effectRunner = runner; + this.engineScope = engineScope; } setEventSink(eventSink: TurnEngineEventSink): void { @@ -1783,20 +1804,31 @@ export class StreamManager { return; } - try { - streamInfo.state = StreamState.STOPPING; - // Flush any pending partial write immediately (preserves work on interruption) - await this.flushPartialWrite(workspaceId, streamInfo); + // Idempotent for concurrent cancellers (a user stop racing the shutdown + // supervisor's "system" cancel): the latch is checked and assigned here, + // synchronously, with no suspension before it — an await ahead of this + // point would let both callers reach cleanupAbortedStream and emit two + // stream-aborts. Later callers join the first cancel (its abortReason wins). + if (streamInfo.cancelPromise) { + return streamInfo.cancelPromise; + } + streamInfo.cancelPromise = (async () => { + try { + streamInfo.state = StreamState.STOPPING; + // Flush any pending partial write immediately (preserves work on interruption) + await this.flushPartialWrite(workspaceId, streamInfo); - streamInfo.abortController.abort(); + streamInfo.abortController.abort(); - // Unlike checkSoftCancelStream, await cleanup (blocking) - await this.cleanupAbortedStream(workspaceId, streamInfo, abortReason, abandonPartial); - } catch (error) { - log.error("Error during stream cancellation:", error); - // Force cleanup even if cancellation fails - this.workspaceStreams.delete(workspaceId); - } + // Unlike checkSoftCancelStream, await cleanup (blocking) + await this.cleanupAbortedStream(workspaceId, streamInfo, abortReason, abandonPartial); + } catch (error) { + log.error("Error during stream cancellation:", error); + // Force cleanup even if cancellation fails + this.workspaceStreams.delete(workspaceId); + } + })(); + return streamInfo.cancelPromise; } // Checks if a soft interrupt is necessary, and performs one if so @@ -1815,9 +1847,18 @@ export class StreamManager { streamInfo.abortController.abort(); // Return back to the stream loop so we can wait for it to finish before - // sending the stream abort event. + // sending the stream abort event. Not awaited: cleanupAbortedStream waits + // for processingPromise, i.e. for the loop this runs inside of. + // Shares cancelStreamSafely's latch: a hard cancel that lands during the + // flush above already owns the abort bookkeeping (keep its promise), and a + // hard cancel arriving later joins this one — one stream-abort either way. const { abandonPartial, abortReason } = streamInfo.softInterrupt; - void this.cleanupAbortedStream(workspaceId, streamInfo, abortReason, abandonPartial); + streamInfo.cancelPromise ??= this.cleanupAbortedStream( + workspaceId, + streamInfo, + abortReason, + abandonPartial + ); } catch (error) { log.error("Error during stream cancellation:", error); // Force cleanup even if cancellation fails @@ -1836,6 +1877,16 @@ export class StreamManager { // while a new stream starts (e.g., old stream writing to partial.json) await streamInfo.processingPromise; + // The cancel lost the race: the loop had already left the fullStream and + // finished as completed/failed (terminalCompletion set, stream-end/error + // emitted, completion settled by processStreamWithCleanup's finally). + // Re-running the abort bookkeeping here would resurrect partial.json after + // deletePartial and emit a second terminal event (stream-abort after + // stream-end), which the renderer reads as "still partial". + if (streamInfo.terminalCompletion !== undefined) { + return; + } + // For aborts, use our tracked cumulativeUsage directly instead of AI SDK's totalUsage. // cumulativeUsage is updated on each finish-step event (before tool execution), // so it has accurate data even when the stream is interrupted mid-tool-call. @@ -4880,6 +4931,8 @@ export class StreamManager { ).catch((error) => { log.error("Unexpected error in stream processing:", error); }); + // After the assignment: the supervisor wraps the already-started promise. + this.superviseEngine(typedWorkspaceId, streamInfo); return Ok(handle); } finally { @@ -4900,6 +4953,64 @@ export class StreamManager { } } + /** + * Make the engine scope (`AppFiberScope`) supervise this stream: one fiber + * per stream that waits on the already-running `processingPromise` and, when + * interrupted by the scope closing during shutdown, cancels the stream through + * the user-stop path (`cancelStreamSafely`, abort reason `"system"`) and waits + * for the turn to settle — i.e. for the partial to be flushed with usage, + * `stream-abort` delivered (AIService commits the partial into chat.jsonl and + * deletes partial.json) and `completion` resolved. `closeScopeBounded` + * awaits that finalizer, so dispose() proceeds to tear down bridges and + * sessions only once every in-flight stream is durably settled. + * + * The fiber is the ownership unit only; the stream's AbortSignal stays the + * sole cancellation transport (the loop, the AI SDK, soft interrupts and + * `stopStream` all key off it), so interruption meets the stream at exactly + * one point — this finalizer. A stream that completes on its own resolves the + * promise and the fiber exits, which removes its finalizer from the scope + * (no per-stream residue). A stream started after the scope closed is + * interrupted synchronously by `forkIn` and thus aborted right away (pinned in + * appFiberScope.test.ts); the pre-registration window (`pendingStreamStarts`) + * is not supervised — nothing durable exists for it yet and `stopStream` + * already aborts pending controllers. + */ + private superviseEngine(workspaceId: WorkspaceId, streamInfo: WorkspaceStreamInfo): void { + if (this.engineScope === undefined) { + return; + } + assert(streamInfo.engineFiber === undefined, "stream engine already supervised"); + // Zero-arity thunk on purpose: Effect.promise allocates an internal + // AbortController only when the thunk declares a `signal` parameter; the + // stream's own controller must stay the only signal in play. + const supervisor = Effect.promise(() => streamInfo.processingPromise).pipe( + Effect.onInterrupt(() => + // Finalizers already run uninterruptibly; explicit per house doctrine so + // the cancel → settle sequence is visibly atomic under a second interrupt. + Effect.uninterruptible( + Effect.promise(async () => { + await this.cancelStreamSafely(workspaceId, streamInfo, "system"); + // Abort delivery (partial commit, downstream stream-abort listeners) + // settles the completion asynchronously after cancelStreamSafely + // returns; shutdown must not proceed until it has. + await streamInfo.completionController.promise; + }) + ) + ), + Effect.catchDefect((defect) => + Effect.sync(() => { + log.warn("[stream] engine supervisor defect", { workspaceId, error: defect }); + }) + ) + ); + // startImmediately: the fiber reaches its (only) suspension point — the + // promise wait — before forkIn registers it with the scope, so a scope that + // is already closed interrupts it right here, synchronously. + streamInfo.engineFiber = this.effectRunner.runSync( + Effect.forkIn(supervisor, this.engineScope, { startImmediately: true }) + ); + } + /** * Record a previousResponseId as lost if the error indicates OpenAI no longer has it. * StreamManager retries once automatically, and buildProviderOptions filters it for future requests. From 09e258d7085022a5b99f7516ceb9a013c6ed60d4 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 3 Sep 2026 17:54:51 +0000 Subject: [PATCH 3/9] test(streamManager): engine supervision cases; retire partial writes once the final message is committed --- src/node/services/streamManager.test.ts | 257 ++++++++++++++++++++++++ src/node/services/streamManager.ts | 17 +- 2 files changed, 273 insertions(+), 1 deletion(-) diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index ff51f1e4db..c9bf6b2b98 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -45,6 +45,8 @@ import { SessionUsageService } from "./sessionUsageService"; import type { HistoryService } from "./historyService"; import { createTestHistoryService } from "./testHistoryService"; import { makeTestEffectRunner } from "./di/testEffectRunner"; +import { closeScopeBounded } from "./di/appRuntime"; +import { Scope } from "effect"; import { createAnthropic } from "@ai-sdk/anthropic"; import { countTokens } from "@/node/utils/main/tokenizer"; import { shouldRunIntegrationTests, validateApiKeys } from "../../../tests/testUtils"; @@ -1223,6 +1225,261 @@ describe("StreamManager - stream resource scope", () => { }); }); +describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { + type StreamAbortEvent = Extract; + + /** + * A manager whose streams are supervised by `engineScope` (the app runtime's + * AppFiberScope in production). `fullStream` receives the stream's own + * AbortSignal so scenarios can model a provider that stops on abort — or one + * that ignores it. + */ + function createSupervisedStreamManagerForTests( + fullStream: (signal: AbortSignal) => AsyncGenerator, + engineScope: Scope.Closeable | undefined = Scope.makeUnsafe("parallel") + ) { + const events: TurnEngineEvent[] = []; + const streamManager = new StreamManager( + historyService, + undefined, + undefined, + (event) => { + events.push(event); + }, + undefined, + engineScope + ); + Reflect.set(streamManager, "tokenTracker", { + setModel: () => Promise.resolve(undefined), + countTokens: () => Promise.resolve(0), + }); + Reflect.set( + streamManager, + "createStreamResult", + (_request: unknown, abortController: AbortController) => + createStreamResultForTests(fullStream(abortController.signal)) + ); + Reflect.set(streamManager, "createTempDirForStream", () => + Promise.resolve("/tmp/wave4-supervisor-tempdir") + ); + Reflect.set(streamManager, "cleanupStreamTempDir", () => undefined); + return { streamManager, events, engineScope }; + } + + async function startSupervisedStreamForTests( + streamManager: StreamManager, + workspaceId: string, + historySequence = 1 + ) { + const messageId = `${workspaceId}-msg-${historySequence}`; + await appendPartialAssistantForTests(workspaceId, messageId, historySequence); + const result = await streamManager.startStream( + testStartOptions({ + workspaceId, + messageId, + model: createTestLanguageModel(), + tools: {}, + historySequence, + }) + ); + expect(result.success).toBe(true); + if (!result.success) { + throw new Error("Expected stream to start"); + } + return result.data; + } + + /** Yields one delta, then blocks until the stream's AbortSignal fires (a well-behaved provider). */ + function flowingThenBlockedStream(signal: AbortSignal): AsyncGenerator { + return (async function* () { + yield { type: "text-delta", text: "hello" }; + await new Promise((resolve) => { + if (signal.aborted) return resolve(); + signal.addEventListener("abort", () => resolve(), { once: true }); + }); + })(); + } + + async function waitForPartialText(workspaceId: string, text: string): Promise { + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + const partial = await historyService.readPartial(workspaceId); + if ( + partial?.parts.some((part) => part.type === "text" && part.text.includes(text)) === true + ) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error(`partial for ${workspaceId} never contained ${JSON.stringify(text)}`); + } + + function terminalEvents(events: TurnEngineEvent[]): TurnEngineEvent[] { + return events.filter((event) => ["stream-end", "stream-abort", "error"].includes(event.type)); + } + + test("closing the engine scope aborts a flowing stream as 'system', commits its partial and settles it", async () => { + const workspaceId = "supervised-flowing-workspace"; + const { streamManager, events, engineScope } = + createSupervisedStreamManagerForTests(flowingThenBlockedStream); + const writePartialSpy = spyOn(historyService, "writePartial"); + const handle = await startSupervisedStreamForTests(streamManager, workspaceId); + await waitForPartialText(workspaceId, "hello"); + expect(streamManager.isStreaming(workspaceId)).toBe(true); + + // What ServiceContainer.dispose() does at step 2: interrupt + await. + await closeScopeBounded(engineScope!); + + // The finalizer routed through the user-stop path: the streamed text was + // flushed (with usage stamping) before the abort was delivered ... + const lastWrite = writePartialSpy.mock.calls.at(-1)?.[1]; + expect(lastWrite?.parts.some((part) => part.type === "text" && part.text === "hello")).toBe( + true + ); + // ... exactly one terminal event, an involuntary backend abort, no stream-end ... + const terminal = terminalEvents(events); + expect(terminal.map((event) => event.type)).toEqual(["stream-abort"]); + expect((terminal[0] as StreamAbortEvent).abortReason).toBe("system"); + // ... and the turn handle plus the registry were settled before the close resolved. + expect(await handle.completion).toEqual({ status: "aborted", abortReason: "system" }); + expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); + expect(streamManager.isStreaming(workspaceId)).toBe(false); + }); + + test("a wedged provider (never yields, ignores abort) cannot pin the bounded close", async () => { + const workspaceId = "supervised-wedged-workspace"; + const { streamManager, engineScope } = createSupervisedStreamManagerForTests(() => + (async function* () { + await new Promise(() => undefined); + yield { type: "text-delta", text: "never" }; + })() + ); + await startSupervisedStreamForTests(streamManager, workspaceId); + expect(streamManager.isStreaming(workspaceId)).toBe(true); + + // cleanupAbortedStream waits for the loop, which waits on the provider that + // never returns; the close must still resolve at the bound, and never reject. + const startedAt = Date.now(); + await closeScopeBounded(engineScope!, 100); + expect(Date.now() - startedAt).toBeLessThan(2_000); + }); + + test("a stream started after the engine scope closed is aborted immediately (fail-closed during shutdown)", async () => { + const workspaceId = "supervised-late-start-workspace"; + const { streamManager, events, engineScope } = + createSupervisedStreamManagerForTests(flowingThenBlockedStream); + await closeScopeBounded(engineScope!); + + const handle = await startSupervisedStreamForTests(streamManager, workspaceId); + + expect(await handle.completion).toEqual({ status: "aborted", abortReason: "system" }); + expect(terminalEvents(events).map((event) => event.type)).toEqual(["stream-abort"]); + expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); + }); + + test("a cancel landing after the loop finished but before COMPLETED neither resurrects partial.json nor emits a second terminal event", async () => { + // The completion path deletes partial.json, then awaits updateHistory, and + // only then flips state to COMPLETED. A cancel (user stop or the shutdown + // supervisor) landing inside that window previously re-created partial.json + // (pre-abort flush + abort bookkeeping) and emitted stream-abort after + // stream-end. No engine scope here: this is the stopStream path itself. + const workspaceId = "cancel-after-loop-exit-workspace"; + const { streamManager, events } = createSupervisedStreamManagerForTests( + () => + (async function* () { + await Promise.resolve(); + yield { type: "text-delta", text: "final answer" }; + yield { type: "finish", finishReason: "stop" }; + })(), + undefined + ); + let stopPromise: Promise | undefined; + const realUpdateHistory = historyService.updateHistory.bind(historyService); + const updateHistorySpy = spyOn(historyService, "updateHistory").mockImplementation( + async (targetWorkspaceId, message) => { + // partial.json is already gone here; state is still STREAMING. + stopPromise ??= streamManager.stopStream(targetWorkspaceId); + return realUpdateHistory(targetWorkspaceId, message); + } + ); + try { + const handle = await startSupervisedStreamForTests(streamManager, workspaceId); + + expect(await handle.completion).toEqual({ status: "completed" }); + expect(stopPromise).toBeDefined(); + expect(await stopPromise).toEqual(Ok(undefined)); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(terminalEvents(events).map((event) => event.type)).toEqual(["stream-end"]); + expect(await historyService.readPartial(workspaceId)).toBeNull(); + const history = await historyService.getHistoryFromLatestBoundary(workspaceId); + expect(history.success).toBe(true); + if (!history.success) throw new Error(history.error); + const assistantRows = history.data.filter((message) => message.role === "assistant"); + expect(assistantRows).toHaveLength(1); + expect(assistantRows[0].metadata?.partial).not.toBe(true); + expect( + assistantRows[0].parts.some((part) => part.type === "text" && part.text === "final answer") + ).toBe(true); + expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); + } finally { + updateHistorySpy.mockRestore(); + } + }); + + test("stopStream racing the engine-scope close produces exactly one stream-abort and one settle", async () => { + const workspaceId = "supervised-stop-vs-close-workspace"; + const { streamManager, events, engineScope } = + createSupervisedStreamManagerForTests(flowingThenBlockedStream); + const handle = await startSupervisedStreamForTests(streamManager, workspaceId); + await waitForPartialText(workspaceId, "hello"); + let settleCount = 0; + void handle.completion.then(() => { + settleCount += 1; + }); + + // Both cancellers enter cancelStreamSafely in the same tick; the latch is + // taken synchronously, so the second joins the first's cleanup. + const stopPromise = streamManager.stopStream(workspaceId, { abortReason: "user" }); + const closePromise = closeScopeBounded(engineScope!); + await Promise.all([stopPromise, closePromise]); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const aborts = events.filter( + (event): event is StreamAbortEvent => event.type === "stream-abort" + ); + expect(aborts).toHaveLength(1); + // First canceller's reason wins (the user pressed stop before shutdown reached the stream). + expect(aborts[0].abortReason).toBe("user"); + expect(terminalEvents(events)).toHaveLength(1); + expect(settleCount).toBe(1); + expect(await handle.completion).toEqual({ status: "aborted", abortReason: "user" }); + expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); + }); + + test("completed streams leave no supervisor residue: closing the scope after 50 completions aborts nothing", async () => { + const workspaceId = "supervised-residue-workspace"; + const { streamManager, events, engineScope } = createSupervisedStreamManagerForTests(() => + (async function* () { + await Promise.resolve(); + yield { type: "text-delta", text: "done" }; + yield { type: "finish", finishReason: "stop" }; + })() + ); + for (let i = 1; i <= 50; i++) { + const handle = await startSupervisedStreamForTests(streamManager, workspaceId, i); + expect(await handle.completion).toEqual({ status: "completed" }); + } + expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); + expect(events.filter((event) => event.type === "stream-end")).toHaveLength(50); + + await closeScopeBounded(engineScope!); + + expect(events.filter((event) => event.type === "stream-abort")).toHaveLength(0); + expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); + }); +}); + describe("StreamManager - stopWhen configuration", () => { type StopWhenCondition = (options: { steps: unknown[] }) => boolean; type BuildStopWhenCondition = (request: { diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 545ac665d1..fcb8d884f5 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -701,6 +701,12 @@ interface WorkspaceStreamInfo { // first await) so concurrent cancellers — a user stop racing shutdown's // engine supervisor — join one cleanup: exactly one stream-abort, one settle. cancelPromise?: Promise; + // Set by the completion path right before it deletes partial.json and writes + // the final message to chat.jsonl. From then on no partial may be written for + // this stream: a cancel landing between deletePartial and COMPLETED (the + // state flips late, see processStreamWithCleanup) would otherwise resurrect + // partial.json through its pre-abort flush. + partialRetired?: boolean; // Supervisor fiber in the app's AppFiberScope (superviseEngine); set only when // the manager was constructed with an engine scope. engineFiber?: Fiber.Fiber; @@ -1210,6 +1216,12 @@ export class StreamManager { // Cancel any scheduled debounce flush — we're writing now this.interruptPartialWriteFiber(streamInfo); + // The final message owns chat.jsonl now; re-creating partial.json here (a + // cancel racing the completion path) would be committed over it on next load. + if (streamInfo.partialRetired) { + return; + } + // Start new write and track the promise streamInfo.partialWritePromise = (async () => { try { @@ -4028,7 +4040,10 @@ export class StreamManager { }; // CRITICAL: Delete partial.json before updating chat.jsonl - // On successful completion, partial.json becomes stale and must be removed + // On successful completion, partial.json becomes stale and must be removed. + // Retire partial writes first: a cancel (user stop, shutdown) landing + // between here and COMPLETED must not flush partial.json back to disk. + streamInfo.partialRetired = true; const deleteResult = await this.historyService.deletePartial(workspaceId as string); if (!deleteResult.success) { workspaceLog.warn("Failed to delete partial on stream end", { From 7cc88ab989829e24c7c2a2ef0e4b12076cc22c35 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 3 Sep 2026 17:57:31 +0000 Subject: [PATCH 4/9] test(streamManager.chaos): engine-scope close at a random iteration settles every stream exactly once --- src/node/services/streamManager.chaos.test.ts | 121 ++++++++++++++++++ 1 file changed, 121 insertions(+) diff --git a/src/node/services/streamManager.chaos.test.ts b/src/node/services/streamManager.chaos.test.ts index 30bb078bb2..e8184f8d2d 100644 --- a/src/node/services/streamManager.chaos.test.ts +++ b/src/node/services/streamManager.chaos.test.ts @@ -9,7 +9,9 @@ * stream afterwards. */ import { describe, test, expect, afterEach, beforeEach } from "bun:test"; +import { Scope } from "effect"; import { StreamManager, type TurnEngineEvent } from "./streamManager"; +import { closeScopeBounded } from "./di/appRuntime"; import type { HistoryService } from "./historyService"; import { createTestHistoryService } from "./testHistoryService"; import { createRuntime } from "@/node/runtime/runtimeFactory"; @@ -224,6 +226,125 @@ describe("StreamManager chaos", () => { } }, 120_000); + test(`with an engine scope closed at a random iteration, every stream settles exactly once (seed=${SEED})`, async () => { + // Shutdown variant: the manager is constructed with an engine scope (the + // app's AppFiberScope), streams before the close are hostile and settle on + // their own, the stream at the close point flows then blocks until its + // AbortSignal fires (so the close has something to interrupt) and may race a + // user stop, and streams after the close start into a closed scope. + // Invariant under all of it: one terminal event per messageId and every + // completion promise settles. + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => unhandled.push(reason); + process.on("unhandledRejection", onUnhandled); + try { + const rng = mulberry32(SEED); + const closeAt = randInt(rng, ITERATIONS); + const stopRacesClose = rng() < 0.5; + const engineScope = Scope.makeUnsafe("parallel"); + const events: TurnEngineEvent[] = []; + const streamManager = new StreamManager( + historyService, + undefined, + undefined, + (event) => { + events.push(event); + }, + undefined, + engineScope + ); + Reflect.set(streamManager, "tokenTracker", { + setModel: () => Promise.resolve(), + countTokens: () => Promise.resolve(0), + }); + let nextFullStream: (signal: AbortSignal) => AsyncGenerator = () => + randomFullStream(rng); + Reflect.set( + streamManager, + "createStreamResult", + (_request: unknown, abortController: AbortController) => ({ + fullStream: nextFullStream(abortController.signal), + totalUsage: Promise.resolve(rng() < 0.7 ? undefined : randomHostileValue(rng)), + usage: Promise.resolve(undefined), + providerMetadata: Promise.resolve(rng() < 0.8 ? undefined : randomHostileValue(rng)), + steps: Promise.resolve([]), + }) + ); + + const started: Array<{ messageId: string; completion: Promise<{ status: string }> }> = []; + for (let iter = 0; iter < ITERATIONS; iter++) { + const workspaceId = `chaos-scope-ws-${iter}`; + const messageId = `chaos-scope-msg-${iter}`; + const appendResult = await historyService.appendToHistory(workspaceId, { + id: messageId, + role: "assistant", + metadata: { historySequence: 1, partial: true }, + parts: [], + }); + expect(appendResult.success).toBe(true); + + nextFullStream = + iter === closeAt + ? (signal) => + (async function* () { + yield { type: "text-delta", text: randomFragmentString(rng) }; + await new Promise((resolve) => { + if (signal.aborted) return resolve(); + signal.addEventListener("abort", () => resolve(), { once: true }); + }); + })() + : () => randomFullStream(rng); + + const result = await streamManager.startStream({ + workspaceId, + messageId, + model: createTestLanguageModel(), + messages: [{ role: "user", content: "hello" }], + modelString: "openai:gpt-4.1-mini", + historySequence: 1, + system: "system", + runtime: LOCAL_TEST_RUNTIME, + providedRuntimeTempDir: "", + }); + expect(result.success).toBe(true); + if (!result.success) continue; + started.push({ messageId, completion: result.data.completion }); + + if (iter === closeAt) { + await Promise.all([ + stopRacesClose ? streamManager.stopStream(workspaceId) : Promise.resolve(), + closeScopeBounded(engineScope), + ]); + expect(engineScope.state._tag).toBe("Closed"); + } + } + + const completions = await Promise.race([ + Promise.all(started.map((entry) => entry.completion)), + new Promise((_, reject) => + setTimeout(() => reject(new Error("a supervised stream never settled")), 15_000) + ), + ]); + for (const completion of completions) { + expect(["completed", "failed", "aborted"]).toContain(completion.status); + } + // Let the last abort deliveries and finally blocks settle. + await new Promise((resolve) => setTimeout(resolve, 0)); + for (const { messageId } of started) { + const terminal = events.filter( + (event) => + ["stream-end", "error", "stream-abort"].includes(event.type) && + "messageId" in event && + event.messageId === messageId + ); + expect(terminal).toHaveLength(1); + } + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + } + }, 120_000); + test("cyclic error-chunk payloads surface the provider message, not a JSON serialization TypeError", async () => { // Regression: the error-part fallback used bare JSON.stringify, so a // cyclic/BigInt error payload threw "Converting circular structure to From 22c3cd6a35a6b9570c2c4fa0afac72f938b01fe7 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 3 Sep 2026 17:59:55 +0000 Subject: [PATCH 5/9] test(di): dispose()/CLI cleanup abort and await an in-flight stream before the explicit teardown --- src/node/services/coreServicesRoot.test.ts | 91 ++++++++++++++++++ src/node/services/serviceContainer.test.ts | 102 +++++++++++++++++++++ 2 files changed, 193 insertions(+) diff --git a/src/node/services/coreServicesRoot.test.ts b/src/node/services/coreServicesRoot.test.ts index f10bcbd018..f05316d0e6 100644 --- a/src/node/services/coreServicesRoot.test.ts +++ b/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((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 diff --git a/src/node/services/serviceContainer.test.ts b/src/node/services/serviceContainer.test.ts index ccc5604d0d..d9c50b32f4 100644 --- a/src/node/services/serviceContainer.test.ts +++ b/src/node/services/serviceContainer.test.ts @@ -7,6 +7,7 @@ import { TestClock } from "effect/testing"; import { MULTI_PROJECT_CONFIG_KEY } from "@/common/constants/multiProject"; import { createConfigStores, type Config, type ConfigStores } from "@/node/config"; import type { ORPCContext } from "@/node/orpc/context"; +import { createRuntime } from "@/node/runtime/runtimeFactory"; import { isInteractiveHostKeyApprovalAvailable } from "@/node/runtime/sshConnectionPool"; import { AppFiberScopeTag } from "@/node/services/di/appFiberScope"; import { EffectRunnerTag } from "@/node/services/di/effectRunner"; @@ -445,6 +446,107 @@ describe("ServiceContainer", () => { expect(services.appFiberScope.state._tag).toBe("Closed"); }); + it("dispose() aborts and awaits an in-flight stream before desktopBridgeServer.stop()", async () => { + // The AppFiberScope occupant end to end: a real stream on the container's + // StreamManager (wired with the runtime's scope), its provider stubbed to + // flow one delta and then block until the stream's AbortSignal fires. + // dispose() must abort it as "system", let AIService commit the partial + // into chat.jsonl and delete partial.json, and only then stop the bridge. + services = new ServiceContainer(stores); + const workspaceId = "dispose-in-flight-stream-workspace"; + const messageId = "dispose-in-flight-stream-message"; + const steps: string[] = []; + Reflect.set(services.streamManager, "tokenTracker", { + setModel: () => Promise.resolve(undefined), + countTokens: () => Promise.resolve(0), + }); + Reflect.set( + services.streamManager, + "createStreamResult", + (_request: unknown, abortController: AbortController) => ({ + fullStream: (async function* () { + yield { type: "text-delta", text: "hello from a stream shutdown must not lose" }; + await new Promise((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(services.streamManager, "createTempDirForStream", () => + Promise.resolve(path.join(tempDir, "stream-tempdir")) + ); + Reflect.set(services.streamManager, "cleanupStreamTempDir", () => undefined); + services.aiService.on("stream-abort", (event: { abortReason?: string }) => { + steps.push(`stream-abort:${event.abortReason}`); + }); + const bridgeStopSpy = spyOn(services.desktopBridgeServer, "stop").mockImplementation(() => { + steps.push("bridge-stop"); + return Promise.resolve(undefined); + }); + + const appendResult = await services.historyService.appendToHistory(workspaceId, { + id: messageId, + role: "assistant", + metadata: { historySequence: 1, partial: true }, + parts: [], + }); + expect(appendResult.success).toBe(true); + const started = await services.streamManager.startStream({ + workspaceId, + messageId, + model: { + specificationVersion: "v3", + provider: "test", + modelId: "dispose-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"); + // Wait for the delta to reach partial.json so there is something to commit. + const deadline = Date.now() + 5_000; + while ((await services.historyService.readPartial(workspaceId)) === null) { + if (Date.now() > deadline) throw new Error("partial never written"); + await new Promise((resolve) => setTimeout(resolve, 5)); + } + expect(services.streamManager.isStreaming(workspaceId)).toBe(true); + + await services.dispose(); + + expect(bridgeStopSpy).toHaveBeenCalledTimes(1); + expect(steps).toEqual(["stream-abort:system", "bridge-stop"]); + expect(await started.data.completion).toEqual({ status: "aborted", abortReason: "system" }); + expect(services.streamManager.isStreaming(workspaceId)).toBe(false); + // Durable outcome at the moment the bridge stopped: partial.json gone, the + // interrupted assistant message (still flagged partial, as every + // interrupted turn is) committed to chat.jsonl with its streamed text — + // instead of an empty placeholder row plus an orphan partial.json that only + // the next load would reconcile. + expect(await services.historyService.readPartial(workspaceId)).toBeNull(); + const history = await services.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 === "hello from a stream shutdown must not lose" + ) + ).toBe(true); + }); + it("shares one teardown across concurrent dispose() calls", async () => { services = new ServiceContainer(stores); const steps: string[] = []; From 8abed14e889b47b864282fcf4c4783cf9bc5a5aa Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 3 Sep 2026 18:04:03 +0000 Subject: [PATCH 6/9] test: lint/type fixes for the supervision tests --- src/node/services/serviceContainer.test.ts | 11 ++++++----- src/node/services/streamManager.test.ts | 17 +++++++++-------- src/node/services/streamManager.ts | 3 ++- 3 files changed, 17 insertions(+), 14 deletions(-) diff --git a/src/node/services/serviceContainer.test.ts b/src/node/services/serviceContainer.test.ts index d9c50b32f4..19d480f1f9 100644 --- a/src/node/services/serviceContainer.test.ts +++ b/src/node/services/serviceContainer.test.ts @@ -482,14 +482,15 @@ describe("ServiceContainer", () => { ); Reflect.set(services.streamManager, "cleanupStreamTempDir", () => undefined); services.aiService.on("stream-abort", (event: { abortReason?: string }) => { - steps.push(`stream-abort:${event.abortReason}`); + steps.push(`stream-abort:${event.abortReason ?? "none"}`); }); const bridgeStopSpy = spyOn(services.desktopBridgeServer, "stop").mockImplementation(() => { steps.push("bridge-stop"); return Promise.resolve(undefined); }); - const appendResult = await services.historyService.appendToHistory(workspaceId, { + const historyService = services.runtime.get(History); + const appendResult = await historyService.appendToHistory(workspaceId, { id: messageId, role: "assistant", metadata: { historySequence: 1, partial: true }, @@ -518,7 +519,7 @@ describe("ServiceContainer", () => { if (!started.success) throw new Error("expected the stream to start"); // Wait for the delta to reach partial.json so there is something to commit. const deadline = Date.now() + 5_000; - while ((await services.historyService.readPartial(workspaceId)) === null) { + while ((await historyService.readPartial(workspaceId)) === null) { if (Date.now() > deadline) throw new Error("partial never written"); await new Promise((resolve) => setTimeout(resolve, 5)); } @@ -535,8 +536,8 @@ describe("ServiceContainer", () => { // interrupted turn is) committed to chat.jsonl with its streamed text — // instead of an empty placeholder row plus an orphan partial.json that only // the next load would reconcile. - expect(await services.historyService.readPartial(workspaceId)).toBeNull(); - const history = await services.historyService.getHistoryFromLatestBoundary(workspaceId); + expect(await historyService.readPartial(workspaceId)).toBeNull(); + const history = await 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); diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index c9bf6b2b98..0e91ec8ef4 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -1236,8 +1236,9 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { */ function createSupervisedStreamManagerForTests( fullStream: (signal: AbortSignal) => AsyncGenerator, - engineScope: Scope.Closeable | undefined = Scope.makeUnsafe("parallel") + options: { supervised: boolean } = { supervised: true } ) { + const engineScope = Scope.makeUnsafe("parallel"); const events: TurnEngineEvent[] = []; const streamManager = new StreamManager( historyService, @@ -1247,7 +1248,7 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { events.push(event); }, undefined, - engineScope + options.supervised ? engineScope : undefined ); Reflect.set(streamManager, "tokenTracker", { setModel: () => Promise.resolve(undefined), @@ -1328,7 +1329,7 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { expect(streamManager.isStreaming(workspaceId)).toBe(true); // What ServiceContainer.dispose() does at step 2: interrupt + await. - await closeScopeBounded(engineScope!); + await closeScopeBounded(engineScope); // The finalizer routed through the user-stop path: the streamed text was // flushed (with usage stamping) before the abort was delivered ... @@ -1360,7 +1361,7 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { // cleanupAbortedStream waits for the loop, which waits on the provider that // never returns; the close must still resolve at the bound, and never reject. const startedAt = Date.now(); - await closeScopeBounded(engineScope!, 100); + await closeScopeBounded(engineScope, 100); expect(Date.now() - startedAt).toBeLessThan(2_000); }); @@ -1368,7 +1369,7 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { const workspaceId = "supervised-late-start-workspace"; const { streamManager, events, engineScope } = createSupervisedStreamManagerForTests(flowingThenBlockedStream); - await closeScopeBounded(engineScope!); + await closeScopeBounded(engineScope); const handle = await startSupervisedStreamForTests(streamManager, workspaceId); @@ -1391,7 +1392,7 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { yield { type: "text-delta", text: "final answer" }; yield { type: "finish", finishReason: "stop" }; })(), - undefined + { supervised: false } ); let stopPromise: Promise | undefined; const realUpdateHistory = historyService.updateHistory.bind(historyService); @@ -1441,7 +1442,7 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { // Both cancellers enter cancelStreamSafely in the same tick; the latch is // taken synchronously, so the second joins the first's cleanup. const stopPromise = streamManager.stopStream(workspaceId, { abortReason: "user" }); - const closePromise = closeScopeBounded(engineScope!); + const closePromise = closeScopeBounded(engineScope); await Promise.all([stopPromise, closePromise]); await new Promise((resolve) => setTimeout(resolve, 0)); @@ -1473,7 +1474,7 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); expect(events.filter((event) => event.type === "stream-end")).toHaveLength(50); - await closeScopeBounded(engineScope!); + await closeScopeBounded(engineScope); expect(events.filter((event) => event.type === "stream-abort")).toHaveLength(0); expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index fcb8d884f5..455538bd08 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -1820,7 +1820,8 @@ export class StreamManager { // supervisor's "system" cancel): the latch is checked and assigned here, // synchronously, with no suspension before it — an await ahead of this // point would let both callers reach cleanupAbortedStream and emit two - // stream-aborts. Later callers join the first cancel (its abortReason wins). + // stream-aborts. Later callers join the first cancel; its abortReason and + // abandonPartial are the ones delivered. if (streamInfo.cancelPromise) { return streamInfo.cancelPromise; } From b1d2a557672d2555cc369465cf0c33bc67a3db15 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 3 Sep 2026 18:15:17 +0000 Subject: [PATCH 7/9] feat(streamManager): [shutdown] streamManager.abortStream debug line per supervised stream --- src/node/services/streamManager.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 455538bd08..0645b6d589 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -5005,11 +5005,18 @@ export class StreamManager { // the cancel → settle sequence is visibly atomic under a second interrupt. Effect.uninterruptible( Effect.promise(async () => { + const startedAt = performance.now(); await this.cancelStreamSafely(workspaceId, streamInfo, "system"); // Abort delivery (partial commit, downstream stream-abort listeners) // settles the completion asynchronously after cancelStreamSafely // returns; shutdown must not proceed until it has. await streamInfo.completionController.promise; + // Per-stream cost inside the AppFiberScope close (shutdownStep style). + log.debug("[shutdown] streamManager.abortStream", { + workspaceId, + messageId: streamInfo.messageId, + ms: Math.round(performance.now() - startedAt), + }); }) ) ), From 095dc9a1418605b86c538c6e4705b30bb95b0b31 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 4 Sep 2026 08:41:34 +0000 Subject: [PATCH 8/9] fix(streamManager): supervise from registration (STARTING window) and treat post-abort iterator rejections as cancellation --- src/node/services/di/appRuntime.ts | 5 +- src/node/services/streamManager.test.ts | 81 +++++++++++++++++++++++++ src/node/services/streamManager.ts | 50 +++++++++++---- 3 files changed, 122 insertions(+), 14 deletions(-) diff --git a/src/node/services/di/appRuntime.ts b/src/node/services/di/appRuntime.ts index d3bb44ecda..2e8a1a4cd2 100644 --- a/src/node/services/di/appRuntime.ts +++ b/src/node/services/di/appRuntime.ts @@ -87,8 +87,9 @@ * by `closeScopeBounded` early in `dispose()`, while every dependency they * might touch during finalization is still alive. Occupant: the stream * engine (`StreamManager.superviseEngine`, Wave 4 PR 1) — one supervisor - * fiber per stream wrapping the already-running processing promise; the - * stream's AbortSignal stays the cancellation transport and interruption + * 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 diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index 0e91ec8ef4..000488057a 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -1458,6 +1458,87 @@ describe("StreamManager - engine supervision (AppFiberScope occupant)", () => { expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); }); + test("closing the engine scope while the turn-envelope write is pending cancels the STARTING stream inside the close", async () => { + // startStream registers the stream, then awaits onStreamConstructed (the + // durable turn-envelope write) before launching processing. Supervision + // starts at registration, so a shutdown landing inside that await cancels + // the STARTING stream through the same hard-interrupt path a user stop + // takes there — before closeScopeBounded resolves, not after teardown has + // moved on and the envelope write finally returns. + const workspaceId = "supervised-starting-window-workspace"; + const { streamManager, events, engineScope } = + createSupervisedStreamManagerForTests(flowingThenBlockedStream); + const messageId = `${workspaceId}-msg`; + await appendPartialAssistantForTests(workspaceId, messageId, 1); + let releaseEnvelope!: () => void; + const envelopeWritten = new Promise((resolve) => { + releaseEnvelope = resolve; + }); + const startPromise = streamManager.startStream( + testStartOptions({ + workspaceId, + messageId, + model: createTestLanguageModel(), + tools: {}, + onStreamConstructed: () => envelopeWritten, + }) + ); + const workspaceStreams = getWorkspaceStreamsForTests(streamManager); + const deadline = Date.now() + 5_000; + while (!workspaceStreams.has(workspaceId)) { + if (Date.now() > deadline) throw new Error("stream never registered"); + await new Promise((resolve) => setTimeout(resolve, 5)); + } + + await closeScopeBounded(engineScope); + + // Cancelled inside the close: abort delivered, registry cleared, no + // stream-start ever emitted for it. + expect(terminalEvents(events).map((event) => event.type)).toEqual(["stream-abort"]); + expect((terminalEvents(events)[0] as StreamAbortEvent).abortReason).toBe("system"); + expect(workspaceStreams.size).toBe(0); + + releaseEnvelope(); + const result = await startPromise; + expect(result.success).toBe(true); + if (!result.success) throw new Error("expected Ok"); + expect(await result.data.completion).toEqual({ status: "aborted", abortReason: "system" }); + expect(events.filter((event) => event.type === "stream-start")).toHaveLength(0); + expect(terminalEvents(events)).toHaveLength(1); + expect(streamManager.isStreaming(workspaceId)).toBe(false); + }); + + test("a provider whose iterator rejects on abort is still recorded as an abort, not a failure", async () => { + // Some transports surface a cancellation as an iterator rejection rather + // than a clean close. The canceller owns the terminal bookkeeping, so the + // loop must not record that rejection as a provider failure — otherwise the + // lost-race guard would suppress the stream-abort and the partial commit. + const workspaceId = "abort-as-rejection-workspace"; + const { streamManager, events } = createSupervisedStreamManagerForTests( + (signal) => + (async function* () { + yield { type: "text-delta", text: "hello" }; + await new Promise((resolve) => { + if (signal.aborted) return resolve(); + signal.addEventListener("abort", () => resolve(), { once: true }); + }); + throw new Error("connection reset after abort"); + })(), + { supervised: false } + ); + const handle = await startSupervisedStreamForTests(streamManager, workspaceId); + await waitForPartialText(workspaceId, "hello"); + + expect(await streamManager.stopStream(workspaceId, { abortReason: "user" })).toEqual( + Ok(undefined) + ); + + expect(await handle.completion).toEqual({ status: "aborted", abortReason: "user" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(terminalEvents(events).map((event) => event.type)).toEqual(["stream-abort"]); + expect(getWorkspaceStreamsForTests(streamManager).size).toBe(0); + }); + test("completed streams leave no supervisor residue: closing the scope after 50 completions aborts nothing", async () => { const workspaceId = "supervised-residue-workspace"; const { streamManager, events, engineScope } = createSupervisedStreamManagerForTests(() => diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 0645b6d589..1c85ca0a1c 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -4092,6 +4092,21 @@ export class StreamManager { } break; } catch (error) { + // A cancellation may surface as an iterator rejection instead of a + // clean close (provider/transport dependent). The canceller that + // aborted the signal owns the terminal bookkeeping (cleanupAbortedStream + // after processingPromise resolves: stream-abort, partial commit, + // settle), so recording this as a provider failure would turn a user + // stop or a shutdown abort into an error event and — via the + // lost-race guard in cleanupAbortedStream — suppress the abort. Take + // the same exit as the clean-close abort path instead. + if (streamInfo.abortController.signal.aborted) { + workspaceLog.debug("Stream iterator rejected after abort; treating as cancellation", { + error: getErrorMessage(error), + }); + await this.flushPartialWrite(workspaceId, streamInfo); + break; + } let handledError: unknown = error; let retried = false; try { @@ -4915,6 +4930,11 @@ export class StreamManager { streamInfo.unlinkAbortSignal = unlinkAbortSignal; streamRegistered = true; + // Supervise from registration on: a shutdown landing during the envelope + // write below must find this STARTING stream and cancel it inside the + // scope close (the hard-interrupt path documented after the await), + // not after teardown has moved past the bridges. + this.superviseEngine(typedWorkspaceId, streamInfo); // Stream constructed + registered: durable request-describing side // effects (turn envelope) may be recorded now. @@ -4947,8 +4967,6 @@ export class StreamManager { ).catch((error) => { log.error("Unexpected error in stream processing:", error); }); - // After the assignment: the supervisor wraps the already-started promise. - this.superviseEngine(typedWorkspaceId, streamInfo); return Ok(handle); } finally { @@ -4964,6 +4982,10 @@ export class StreamManager { } catch (error) { // Guaranteed cleanup on any failure this.workspaceStreams.delete(typedWorkspaceId); + // No handle is handed out on this path, so the completion is observed only + // by a supervisor forked at registration: settle it so that fiber exits + // instead of cancelling this never-started stream at shutdown. + completionController.settle({ status: "aborted", abortReason: "startup" }); // Convert to strongly-typed error return Err(this.convertToSendMessageError(error)); } @@ -4971,19 +4993,23 @@ export class StreamManager { /** * Make the engine scope (`AppFiberScope`) supervise this stream: one fiber - * per stream that waits on the already-running `processingPromise` and, when - * interrupted by the scope closing during shutdown, cancels the stream through - * the user-stop path (`cancelStreamSafely`, abort reason `"system"`) and waits - * for the turn to settle — i.e. for the partial to be flushed with usage, - * `stream-abort` delivered (AIService commits the partial into chat.jsonl and - * deletes partial.json) and `completion` resolved. `closeScopeBounded` - * awaits that finalizer, so dispose() proceeds to tear down bridges and - * sessions only once every in-flight stream is durably settled. + * per stream that lives exactly as long as the turn is unsettled (it waits on + * `completionController.promise`) and, when interrupted by the scope closing + * during shutdown, cancels the stream through the user-stop path + * (`cancelStreamSafely`, abort reason `"system"`) and waits for the turn to + * settle — i.e. for the partial to be flushed with usage, `stream-abort` + * delivered (AIService commits the partial into chat.jsonl and deletes + * partial.json) and `completion` resolved. `closeScopeBounded` awaits that + * finalizer, so dispose() proceeds to tear down bridges and sessions only once + * every in-flight stream is durably settled. Supervision starts at + * registration (before the awaited turn-envelope write), so a STARTING stream + * is covered too: cancelling it takes the same hard-interrupt path a user stop + * takes during that write. * * The fiber is the ownership unit only; the stream's AbortSignal stays the * sole cancellation transport (the loop, the AI SDK, soft interrupts and * `stopStream` all key off it), so interruption meets the stream at exactly - * one point — this finalizer. A stream that completes on its own resolves the + * one point — this finalizer. A turn that settles on its own resolves the * promise and the fiber exits, which removes its finalizer from the scope * (no per-stream residue). A stream started after the scope closed is * interrupted synchronously by `forkIn` and thus aborted right away (pinned in @@ -4999,7 +5025,7 @@ export class StreamManager { // Zero-arity thunk on purpose: Effect.promise allocates an internal // AbortController only when the thunk declares a `signal` parameter; the // stream's own controller must stay the only signal in play. - const supervisor = Effect.promise(() => streamInfo.processingPromise).pipe( + const supervisor = Effect.promise(() => streamInfo.completionController.promise).pipe( Effect.onInterrupt(() => // Finalizers already run uninterruptibly; explicit per house doctrine so // the cancel → settle sequence is visibly atomic under a second interrupt. From 3d0b390ffab5d68a81488cb0e4917920cd2988f0 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 4 Sep 2026 08:44:54 +0000 Subject: [PATCH 9/9] fix(streamManager): supervisor fiber is Fiber --- src/node/services/streamManager.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 1c85ca0a1c..645bd3e11a 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -5050,7 +5050,8 @@ export class StreamManager { Effect.sync(() => { log.warn("[stream] engine supervisor defect", { workspaceId, error: defect }); }) - ) + ), + Effect.asVoid ); // startImmediately: the fiber reaches its (only) suspension point — the // promise wait — before forkIn registers it with the scope, so a scope that