From bdbee09f6e989f30042cb7cce0247544bdb15dd2 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 20:40:22 +0000 Subject: [PATCH] fix(reconciler): finalize progress and surface cancel failures Classify setup and stream interruption as aborts while preserving handoff after normal completion, and propagate standalone Flight reader cancellation failures before complete. --- .../reconciler-interrupt-finalization.md | 7 ++ packages/rsc-runtime/src/reconciler.ts | 46 ++++++++---- packages/rsc-runtime/tests/dispatcher.test.ts | 70 +++++++++++++++++++ 3 files changed, 111 insertions(+), 12 deletions(-) create mode 100644 .changeset/reconciler-interrupt-finalization.md diff --git a/.changeset/reconciler-interrupt-finalization.md b/.changeset/reconciler-interrupt-finalization.md new file mode 100644 index 000000000..188a375c4 --- /dev/null +++ b/.changeset/reconciler-interrupt-finalization.md @@ -0,0 +1,7 @@ +--- +"@agent-bundle/runtime": patch +--- + +Finalize progress reporters on setup and stream interruption with an abort +outcome, and surface Flight reader cancellation failures that do not mask an +earlier stream failure. diff --git a/packages/rsc-runtime/src/reconciler.ts b/packages/rsc-runtime/src/reconciler.ts index 416aebe7a..150a5ce8e 100644 --- a/packages/rsc-runtime/src/reconciler.ts +++ b/packages/rsc-runtime/src/reconciler.ts @@ -1,4 +1,4 @@ -import { Duration, Effect, Option, Queue, Stream, type Scope } from 'effect'; +import { Deferred, Duration, Effect, Exit, Option, Queue, Stream, type Scope } from 'effect'; import { createElement, isValidElement, type ReactElement, type ReactNode } from 'react'; import { createFromReadableStream } from 'react-server-dom-rspack/client.node'; @@ -18,6 +18,7 @@ import { decodeAgentDocument } from './decode-document.js'; import { abortToInterrupt, isAbortError, + mapCause, runPromise, scopedAbortSignal, streamToReadableStream, @@ -353,6 +354,7 @@ const reconcileLoopStream = ( ids: Map, initial: TreeSnapshot, limits: Partial | undefined, + flightDone: Deferred.Deferred, progressInputs: Queue.Queue, sequence: AgentRenderEventSequence, ): Stream.Stream => @@ -363,7 +365,8 @@ const reconcileLoopStream = ( Error > => { if (snapshot.pending.length === 0) { - return Queue.clear(progressInputs).pipe( + return Deferred.await(flightDone).pipe( + Effect.andThen(Queue.clear(progressInputs)), Effect.flatMap((queued) => Effect.try({ catch: (error) => toRuntimeError(error), @@ -412,14 +415,28 @@ const reconcileLoopStream = ( const gatedFlightStream = ( flight: ReadableStream, demand: FlightDemand, + flightDone: Deferred.Deferred, ): Stream.Stream => Stream.unwrap( Effect.gen(function*() { const reader = yield* Effect.acquireRelease( Effect.sync(() => flight.getReader()), - // cancel() rejects when the source already errored; a defect inside - // this closing scope must not replace the stream's own failure. - (handle) => Effect.promise(() => handle.cancel().then(() => undefined, () => undefined)), + (handle, exit) => Effect.gen(function*() { + const cancelExit = yield* Effect.exit(Effect.tryPromise({ + catch: (error) => toRuntimeError(error), + try: () => handle.cancel(), + })); + if (Exit.isFailure(exit)) { + yield* Deferred.fail(flightDone, mapCause(exit.cause)); + return; + } + if (Exit.isFailure(cancelExit)) { + const error = mapCause(cancelExit.cause); + yield* Deferred.fail(flightDone, error); + return yield* Effect.die(error); + } + yield* Deferred.succeed(flightDone, undefined); + }), ); return Stream.unfold(undefined, () => demand.wait.pipe( @@ -439,6 +456,7 @@ const decodeFlightRoot = ( flight: ReadableStream, demand: FlightDemand, signal: AbortSignal, + flightDone: Deferred.Deferred, ): Effect.Effect => Effect.gen(function*() { ensureAgentFlightManifest(); @@ -449,7 +467,7 @@ const decodeFlightRoot = ( // (the maxEvents hang). The scoped signal interrupts the source stream // without touching the locked ReadableStream. const flightAbort = yield* scopedAbortSignal; - const readable = streamToReadableStream(gatedFlightStream(flight, demand), { + const readable = streamToReadableStream(gatedFlightStream(flight, demand, flightDone), { signal: flightAbort, strategy: { highWaterMark: 1 }, }); @@ -521,6 +539,7 @@ export const createAgentRenderEventSession = ( const events = Stream.unwrap( Effect.gen(function*() { const progressInputs = yield* Queue.bounded(0); + const flightDone = yield* Deferred.make(); const bindProgress = (): void => { offerProgress = (input) => Queue.offer(progressInputs, input).pipe( @@ -541,7 +560,7 @@ export const createAgentRenderEventSession = ( try: () => options.flight, }); if (options.signal.aborted) return yield* Effect.fail(abortError()); - const root = yield* decodeFlightRoot(flight, options.demand, options.signal); + const root = yield* decodeFlightRoot(flight, options.demand, options.signal, flightDone); if (options.signal.aborted) return yield* Effect.fail(abortError()); const prepared = yield* Effect.try({ catch: (error) => toRuntimeError(error), @@ -566,21 +585,24 @@ export const createAgentRenderEventSession = ( prepared.ids, prepared.initial, options.limits, + flightDone, progressInputs, sequence, ), ).pipe( Stream.mapEffect((input) => emitBoundRenderEvent(sequence, input)), Stream.tap((event) => (event.type === 'shell' ? options.demand.markShell : Effect.void)), - Stream.tapError((error) => Effect.sync(() => { - progressFailure = error; - })), Stream.takeUntil((event) => event.type === 'complete'), - Stream.ensuring(Effect.suspend(() => finalizeProgress(progressFailure ?? handoffRequired()))), + Stream.onExit((exit) => { + if (Exit.isSuccess(exit)) return finalizeProgress(handoffRequired()); + const error = mapCause(exit.cause); + return finalizeProgress(sequence.completed && isAbortError(error) ? handoffRequired() : error); + }), ); }); return yield* setup.pipe( - Effect.catch((error) => finalizeProgress(error).pipe(Effect.andThen(Effect.fail(error)))), + Effect.onExit((exit) => + Exit.isFailure(exit) ? finalizeProgress(mapCause(exit.cause)) : Effect.void), ); }), ); diff --git a/packages/rsc-runtime/tests/dispatcher.test.ts b/packages/rsc-runtime/tests/dispatcher.test.ts index 8c9d65ba0..4c96b2c8b 100644 --- a/packages/rsc-runtime/tests/dispatcher.test.ts +++ b/packages/rsc-runtime/tests/dispatcher.test.ts @@ -402,6 +402,50 @@ describe('AgentRenderDispatcher streaming', () => { await expect(reader.read()).rejects.toMatchObject({ name: 'AbortError' }); }); + it('finalizes progress with an abort when Flight setup is interrupted', async () => { + let progress: AgentProgressReporter | undefined; + const host: AgentFlightExecutionHost = { + execute: (request) => { + progress = request.progress; + return new Promise>(() => undefined); + }, + }; + const dispatcher = createAgentRenderDispatcher(host); + const controller = new AbortController(); + const reader = dispatcher.stream({ invocation, signal: controller.signal }).getReader(); + const pending = reader.read(); + controller.abort(); + + await expect(pending).rejects.toMatchObject({ name: 'AbortError' }); + if (progress === undefined) throw new Error('expected the dispatcher to install a progress reporter'); + await expect(progress.report({ completed: 1, message: 'after-abort' })).rejects.toMatchObject({ + name: 'AbortError', + }); + }); + + it('finalizes progress with an abort when the event stream is interrupted', { retry: 2 }, async () => { + let progress: AgentProgressReporter | undefined; + const inner = createWorkerHost('single'); + const host: AgentFlightExecutionHost = { + execute: async (request) => { + progress = request.progress; + return inner.execute(request); + }, + }; + const dispatcher = createAgentRenderDispatcher(host); + const controller = new AbortController(); + const reader = dispatcher.stream({ invocation, signal: controller.signal }).getReader(); + const shell = await reader.read(); + if (shell.value?.type !== 'shell') throw new Error('expected a shell event'); + controller.abort(); + + await expect(reader.read()).rejects.toMatchObject({ name: 'AbortError' }); + if (progress === undefined) throw new Error('expected the dispatcher to install a progress reporter'); + await expect(progress.report({ completed: 1, message: 'after-abort' })).rejects.toMatchObject({ + name: 'AbortError', + }); + }); + it('rejects a post-completion progress producer with a typed handoff', { retry: 2 }, async () => { let progress: AgentProgressReporter | undefined; const inner = createWorkerHost('ready'); @@ -588,4 +632,30 @@ describe('AgentRenderDispatcher streaming', () => { if (progress === undefined) throw new Error('expected the dispatcher to install a progress reporter'); await expect(progress.report({ completed: 1, message: 'after-fail' })).rejects.toThrow('flight setup failed'); }); + + it('surfaces a Flight reader cancel rejection without a prior stream failure', { retry: 2 }, async () => { + const inner = createWorkerHost('ready'); + let cancelCalls = 0; + const host: AgentFlightExecutionHost = { + execute: async (request) => { + const flight = await inner.execute(request); + const reader = flight.getReader(); + return { + getReader: () => ({ + cancel: async () => { + cancelCalls += 1; + throw new Error('flight cancel failed'); + }, + read: () => reader.read(), + }), + } as ReadableStream; + }, + }; + const dispatcher = createAgentRenderDispatcher(host); + + await expect(collectEvents( + dispatcher.stream({ invocation, signal: new AbortController().signal }), + )).rejects.toThrow('flight cancel failed'); + expect(cancelCalls).toBe(1); + }); });