From 337d3c799c854f1b1d2ea2f0cc375cb45a112e70 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 16:30:24 +0000 Subject: [PATCH 1/3] refactor(runtime): Effect-native dispatcher and stream pipeline (wave 3.5 stage 2) Rewrite the #145 Flight decode and bound-event pipeline on Effect Streams and fibers while keeping dispatch() and stream() byte-compatible. Refs #152. --- .changeset/effect-dispatcher-stage2.md | 9 + agent-patterns/effect-concurrency.md | 4 +- agent-patterns/effect-errors.md | 2 + agent-patterns/effect-scope.md | 7 + agent-patterns/effect-stream.md | 38 +- docs/effect-conventions.md | 18 +- packages/rsc-runtime/src/dispatcher.ts | 66 ++- packages/rsc-runtime/src/effect/boundary.ts | 148 +++++- .../rsc-runtime/src/effect/render-stream.ts | 69 +++ packages/rsc-runtime/src/reconciler.ts | 453 +++++++++--------- .../rsc-runtime/tests/agent-document.test.ts | 25 + .../rsc-runtime/tests/effect-boundary.test.ts | 60 ++- .../rsc-runtime/tests/state-packaging.test.ts | 9 +- 13 files changed, 623 insertions(+), 285 deletions(-) create mode 100644 .changeset/effect-dispatcher-stage2.md create mode 100644 packages/rsc-runtime/src/effect/render-stream.ts diff --git a/.changeset/effect-dispatcher-stage2.md b/.changeset/effect-dispatcher-stage2.md new file mode 100644 index 000000000..d8325645b --- /dev/null +++ b/.changeset/effect-dispatcher-stage2.md @@ -0,0 +1,9 @@ +--- +"@agent-bundle/runtime": patch +--- + +Rewrite the runtime dispatcher internals on Effect v4 behind the unchanged +`dispatch()` / `stream()` Promise and ReadableStream edges. Flight decode is +an Effect Stream with native pull backpressure; invocation-local boundary +reconciliation and contract bounds are stream stages; host AbortSignal is +honored at the public edge via the boundary interruption bridges. diff --git a/agent-patterns/effect-concurrency.md b/agent-patterns/effect-concurrency.md index 3a9bd41d5..85b362e40 100644 --- a/agent-patterns/effect-concurrency.md +++ b/agent-patterns/effect-concurrency.md @@ -18,7 +18,9 @@ lifecycle). Build those Effect-native from day one behind Promise edges. Host `AbortSignal` still exists at the edges (`dispatch()` / `stream()`). Do not thread extra internal signals once the program is an Effect; interrupt -the fiber. +the fiber (`Stream.interruptWhen` + `abortToInterrupt`). `Latch.makeUnsafe` +is the legal sync bridge when a web `ReadableStream.pull` must open demand +for an Effect stream — do not invent a second AbortSignal for that. ## Bounded work diff --git a/agent-patterns/effect-errors.md b/agent-patterns/effect-errors.md index 40451a22e..13ad7a30b 100644 --- a/agent-patterns/effect-errors.md +++ b/agent-patterns/effect-errors.md @@ -52,6 +52,8 @@ Callers of `runPromise` see the same types they see today. - Putting `unknown` or global `Error` in the fail channel (`unknownInEffectCatch`, `globalErrorInEffectFailure`). - Swallowing interruption as a typed success. Cancellation is `AbortError`. + `Stream.toReadableStream`'s `Cause.squash` is not that mapping — use the + boundary helper. - `Effect.runPromise` in a test to "see the error" when `runPromiseExit` + `Cause` is the assertion you want — still only through the boundary. - New public error codes without updating the authoring docs and the mapping diff --git a/agent-patterns/effect-scope.md b/agent-patterns/effect-scope.md index 522666329..d6d3f6bdd 100644 --- a/agent-patterns/effect-scope.md +++ b/agent-patterns/effect-scope.md @@ -28,6 +28,13 @@ const connection = Effect.acquireRelease( Finalizers run in reverse acquire order. Interruption still runs them. +Do not `ReadableStream.cancel()` a stream React still holds a reader on — +that throws `ReadableStream is locked` (sync or as a rejected promise) and +can defect the finalizer. Interrupt the Effect producer (`AbortSignal` + +`Stream.interruptWhen`, or `runFork(Fiber.interrupt)`) instead of canceling +the locked web stream. Never `runPromise` from a finalizer that another +fiber is already tearing down. + ## Transactions (stage-1 kernel idiom) `Effect.acquireUseRelease` when begin/commit/rollback are one unit and the diff --git a/agent-patterns/effect-stream.md b/agent-patterns/effect-stream.md index 16f331717..1421dbb8e 100644 --- a/agent-patterns/effect-stream.md +++ b/agent-patterns/effect-stream.md @@ -46,14 +46,46 @@ Streams are pull-based. Downstream `run*` pulls chunks; producers that honor the pull (readable streams, queues) automatically apply backpressure. Do not add a second gate (`TransformStream` + manual pause) around an Effect stream. +## Stage 2 dispatcher lessons + +- `Stream.toReadableStream` calls `runFork` and maps failures with + `Cause.squash`. Wrap it in the package boundary (`streamToReadableStream`) + and use `mapCause` so interrupt-only causes stay `AbortError`. `tapError` + must error the web controller *before* scope finalizers run — a hanging + Flight cancel otherwise hides bound-violation and abort failures. +- Never `runPromise(Fiber.interrupt)` from `ReadableStream.cancel`. That + cancel is invoked from `acquireRelease` on the parent event fiber; + blocking on interrupt deadlocks it. Use `runFork` and do not await it. +- React's `createFromReadableStream` still owns a web `ReadableStream`. + Wait for event demand *then* `reader.read()` (`Stream.unfold` + Latch). + Wait-after-`fromReadableStream` either over-pulls (fails backpressure) + or never pulls (deadlock). Do not cancel the Flight byte stream when the + shell root arrives — later boundaries still need those bytes. React may + still hold the reader at scope close; `stream.cancel()` then throws + "locked" and must be swallowed. +- `host.execute({ progress })` must run in the same turn as `stream()`. + Lazy `Stream.unwrap` raced `resolve('a')` ("Flight worker is not running"). +- `Stream.callback` is the wrong event fan-in: a failed producer does not + fail the stream unless you `Queue.fail`. Use `Stream.merge` + + `takeUntil(complete)` + `Queue.shutdown` on `ensuring`. +- `Stream.paginate` is the pending-boundary loop (shell → replace/error* → + complete). +- `progress.report()` after complete must reject `handoff-required` on the + reporter, not only on the stream — share `createAgentRenderEventSequence`. +- After a producer fail, a later `pull()` with HWM 0 must *reject*, not + resolve. `controller.error` alone can lose the error if no read is pending. +- Flight is not Ndjson. Do not adopt `effect/unstable/encoding` for this + pipeline. + ## What to avoid - `for await` over a stream you already have as `Stream` — use `mapEffect` / `runForEach`. - Encoding/decoding JSON by hand when `Stream.pipeThroughChannel` + `effect/unstable/encoding` (Ndjson / SchemaBinary) would do. Unstable - encoding is **not** adopted in Stage 0; list it in - `docs/effect-conventions.md` before first use. -- Constructing `new ReadableStream` to paper over missing backpressure. + encoding is **not** adopted; list it in `docs/effect-conventions.md` + before first use. +- Constructing `new ReadableStream` to paper over missing backpressure + (the boundary `streamToReadableStream` is the legal web-stream edge). - Calling `Effect.runPromise` on each chunk. Consume with `Stream.run*` inside Effect, one `runPromise` at the boundary. diff --git a/docs/effect-conventions.md b/docs/effect-conventions.md index 03c26ff74..c32440ed9 100644 --- a/docs/effect-conventions.md +++ b/docs/effect-conventions.md @@ -84,9 +84,15 @@ on the Promise edge. Do not widen public error types to satisfy Effect. ## Streams and concurrency -Stage 2+ use Effect `Stream` (native backpressure replaces the pull-gated -`TransformStream`), fibers instead of internal signal threading, and -`PubSub` / `Semaphore` / `Latch` for the dev seam. Pattern files: +Stage 2 uses Effect `Stream` for the #145 dispatcher: Flight bytes via +`Stream.unfold` that waits for event-stream demand *before* `reader.read()`, +pending boundaries via `Stream.paginate`, contract bounds as the emit stage +(`boundRenderEventStream` / `createAgentRenderEventSequence`), and progress +via `Stream.merge` + `takeUntil(complete)` (not `Stream.callback` — a failed +callback producer does not fail the stream). A `Latch` opened from the +public event-stream pull gates Flight bytes after the shell. Host +`AbortSignal` becomes `Stream.interruptWhen` + `abortToInterrupt` at the +public edge. Pattern files: - [agent-patterns/effect-stream.md](../agent-patterns/effect-stream.md) - [agent-patterns/effect-scope.md](../agent-patterns/effect-scope.md) @@ -102,11 +108,13 @@ Stage 2+ use Effect `Stream` (native backpressure replaces the pull-gated - `@effect/vitest` — this repo uses rstest. - `NodeRuntime.runMain` / `BunRuntime` as a substitute for the boundary. - Ad-hoc `ManagedRuntime` outside a boundary module. -- `effect/unstable/*` until listed below. +- `effect/unstable/*` until listed below (Stage 2 listed none). ## Unstable-module adoptions -Re-pin chores re-verify every row. Stage 0 adopts none. +Re-pin chores re-verify every row. Stage 2 adopts none: Flight is a React +binary stream, not Ndjson/SchemaBinary, and no other `effect/unstable/*` +module fits the dispatcher rewrite. | Module | Adopted in | Re-verify | | --- | --- | --- | diff --git a/packages/rsc-runtime/src/dispatcher.ts b/packages/rsc-runtime/src/dispatcher.ts index e0438707e..747f6e5f4 100644 --- a/packages/rsc-runtime/src/dispatcher.ts +++ b/packages/rsc-runtime/src/dispatcher.ts @@ -4,8 +4,9 @@ import { type AgentRenderEvent, type AgentRenderLimits, } from './agent-document.js'; -import type { AgentProgressReporter, AgentProgressUpdate, AgentRenderInvocation } from './agent-request.js'; -import { createAgentFlightEventSession, decodeAgentFlightStream } from './reconciler.js'; +import type { AgentProgressReporter, AgentRenderInvocation } from './agent-request.js'; +import { createFlightDemand } from './effect/render-stream.js'; +import { createAgentRenderEventSession, toPublicEventStream } from './reconciler.js'; export { decodeAgentDocument } from './decode-document.js'; @@ -30,6 +31,13 @@ export interface AgentRenderDispatcherOptions { const abortError = (): DOMException => new DOMException('Agent render was aborted', 'AbortError'); +const abortedStream = (): ReadableStream => + new ReadableStream({ + start(controller) { + controller.error(abortError()); + }, + }); + const drainCompleteDocument = async ( events: ReadableStream, signal: AbortSignal, @@ -69,43 +77,27 @@ export const createAgentRenderDispatcher = ( options: AgentRenderDispatcherOptions = {}, ): AgentRenderDispatcher => { const stream = (request: AgentRenderDispatch): ReadableStream => { - if (request.signal.aborted) { - return new ReadableStream({ - start(controller) { - controller.error(abortError()); - }, - }); - } - const session = createAgentFlightEventSession({ limits: options.limits, signal: request.signal }); - const progress: AgentProgressReporter = Object.freeze({ - report: async (update: AgentProgressUpdate) => { - await session.live.emit(session.sequence.emit({ - completed: update.completed ?? 0, - ...(update.message === undefined ? {} : { message: update.message }), - ...(update.total === undefined ? {} : { total: update.total }), - type: 'progress', - })); + if (request.signal.aborted) return abortedStream(); + const demand = createFlightDemand(); + const pendingFlight: { current?: Promise> } = {}; + const session = createAgentRenderEventSession({ + demand, + get flight() { + const current = pendingFlight.current; + if (current === undefined) { + return Promise.reject(new AgentContractError('invalid-document', 'Flight worker is not running')); + } + return current; }, + limits: options.limits, + signal: request.signal, }); - void (async () => { - try { - if (request.signal.aborted) throw abortError(); - const flight = await host.execute({ - invocation: request.invocation, - progress, - signal: request.signal, - }); - if (request.signal.aborted) throw abortError(); - decodeAgentFlightStream(flight, { - limits: options.limits, - session, - signal: request.signal, - }); - } catch (error) { - session.live.fail(request.signal.aborted ? abortError() : error); - } - })(); - return session.readable; + pendingFlight.current = host.execute({ + invocation: request.invocation, + progress: session.progress, + signal: request.signal, + }); + return toPublicEventStream(session.events, demand, request.signal); }; return Object.freeze({ diff --git a/packages/rsc-runtime/src/effect/boundary.ts b/packages/rsc-runtime/src/effect/boundary.ts index ae5cefe27..dc76aa41a 100644 --- a/packages/rsc-runtime/src/effect/boundary.ts +++ b/packages/rsc-runtime/src/effect/boundary.ts @@ -2,6 +2,9 @@ import { Cause, Effect, Exit, + Fiber, + Latch, + Stream, type Layer, ManagedRuntime, type Scope, @@ -126,6 +129,29 @@ export const makeScopedEffectRuntime = ( }); }; +/** + * Host AbortSignal → Effect interruption. Re-checks `signal.aborted` when + * the effect starts (not only when this helper is constructed) so a signal + * that aborts between construction and run still interrupts. + */ +export const abortToInterrupt = (signal: AbortSignal): Effect.Effect => + Effect.suspend(() => { + if (signal.aborted) return interruptAs(); + return Effect.callback((resume) => { + if (signal.aborted) { + resume(Effect.interrupt); + return undefined; + } + const onAbort = (): void => { + resume(Effect.interrupt); + }; + signal.addEventListener('abort', onAbort, { once: true }); + return Effect.sync(() => { + signal.removeEventListener('abort', onAbort); + }); + }); + }); + /** * AbortSignal → Effect interruption, for programs that still run inside * Effect and receive a host signal. The Promise edge also accepts `signal` @@ -134,20 +160,116 @@ export const makeScopedEffectRuntime = ( export const interruptWhenAborted = ( effect: Effect.Effect, signal: AbortSignal, -): Effect.Effect => { - if (signal.aborted) return interruptAs(); - return Effect.raceFirst( - effect, - Effect.callback((resume) => { - const onAbort = () => { - resume(Effect.interrupt); - }; - signal.addEventListener('abort', onAbort, { once: true }); - return Effect.sync(() => { - signal.removeEventListener('abort', onAbort); +): Effect.Effect => Effect.raceFirst(effect, abortToInterrupt(signal)); + +export interface StreamToReadableOptions { + readonly closeOn?: (value: A) => boolean; + readonly onPull?: () => void; + readonly onPullDelivered?: () => void; + readonly signal?: AbortSignal; + readonly strategy?: QueuingStrategy; +} + +/** + * Stream → web ReadableStream. Owns the `runFork` / cancel `runPromise` pair + * that Effect's `Stream.toReadableStream` would otherwise call. Failures map + * through {@link mapCause} so interrupt-only causes stay `AbortError` + * (Effect's helper uses `Cause.squash`, which is the wrong public contract). + */ +export const streamToReadableStream = ( + stream: Stream.Stream, + options: StreamToReadableOptions = {}, +): ReadableStream => { + let currentPull: { readonly resolve: () => void; readonly reject: (error: Error) => void } | undefined; + let fiber: Fiber.Fiber | undefined; + let terminal: { readonly error?: Error } | undefined; + const latch = Latch.makeUnsafe(false); + const source = options.signal === undefined + ? stream + : Stream.interruptWhen(stream, abortToInterrupt(options.signal)); + const settlePull = (error?: Error): void => { + const waiter = currentPull; + currentPull = undefined; + if (waiter === undefined) return; + if (error === undefined) waiter.resolve(); + else waiter.reject(error); + }; + const finish = (error?: Error): void => { + if (terminal !== undefined) return; + terminal = { error }; + settlePull(error); + }; + + const failController = (controller: ReadableStreamDefaultController, error: Error): void => { + try { + controller.error(error); + } catch { + // Already closed or errored — pull() still rejects via `terminal`. + } + finish(error); + }; + + return new ReadableStream({ + cancel() { + const running = fiber; + fiber = undefined; + if (running === undefined) return; + // Never `runPromise` here: this cancel is invoked from an Effect + // acquireRelease finalizer (Flight bytes). Blocking on interrupt + // deadlocks the parent fiber and hides contract / abort errors. + void Effect.runFork(Effect.asVoid(Fiber.interrupt(running))); + }, + pull() { + if (terminal !== undefined) { + return terminal.error === undefined ? Promise.resolve() : Promise.reject(terminal.error); + } + options.onPull?.(); + return new Promise((resolve, reject) => { + currentPull = { reject, resolve }; + latch.openUnsafe(); }); - }), - ); + }, + start(controller) { + const watched = Stream.tapError(source, (error) => + Effect.sync(() => { + failController(controller, toRuntimeError(error)); + }), + ); + fiber = Effect.runFork(Stream.runForEachArray(watched, (chunk) => + latch.whenOpen(Effect.sync(() => { + if (terminal !== undefined) return; + latch.closeUnsafe(); + for (const item of chunk) { + controller.enqueue(item); + if (options.closeOn?.(item) === true) { + controller.close(); + options.onPullDelivered?.(); + finish(); + if (fiber !== undefined) { + void Effect.runFork(Fiber.interrupt(fiber)); + } + return; + } + } + options.onPullDelivered?.(); + settlePull(); + })), + )); + fiber.addObserver((exit) => { + if (terminal !== undefined) return; + if (Exit.isFailure(exit)) { + failController(controller, mapCause(exit.cause)); + return; + } + try { + controller.close(); + } catch { + // Already closed by closeOn. + } + finish(); + }); + }, + }, options.strategy); }; /** diff --git a/packages/rsc-runtime/src/effect/render-stream.ts b/packages/rsc-runtime/src/effect/render-stream.ts new file mode 100644 index 000000000..ba012415c --- /dev/null +++ b/packages/rsc-runtime/src/effect/render-stream.ts @@ -0,0 +1,69 @@ +import { Effect, Latch, Stream } from 'effect'; + +import { + createAgentRenderEventSequence, + type AgentRenderEvent, + type AgentRenderEventInput, + type AgentRenderLimits, +} from '../agent-document.js'; +import { toRuntimeError } from './boundary.js'; + +/** + * Flight-byte demand latch. After the shell event is produced, further + * Flight chunks wait until the public event stream is being pulled. The + * pull hooks are sync because they fire from a web ReadableStream. + */ +export interface FlightDemand { + readonly markShell: Effect.Effect; + readonly notePull: () => void; + readonly notePullEnd: () => void; + readonly wait: Effect.Effect; +} + +export const createFlightDemand = (): FlightDemand => { + const pulling = Latch.makeUnsafe(false); + let shellEmitted = false; + return { + markShell: Effect.sync(() => { + shellEmitted = true; + }), + notePull() { + pulling.openUnsafe(); + }, + notePullEnd() { + pulling.closeUnsafe(); + }, + wait: Effect.suspend(() => (shellEmitted ? pulling.await : Effect.void)), + }; +}; + +/** + * Contract bounds as a stream stage: sequence numbers, elapsed / rate / + * count / event-bytes, document snapshot bounds (depth / nodes / bytes), + * and handoff-required after complete. Uses the same stepper as + * `createAgentRenderEventSequence` so the #140 tests stay the source of + * truth. + */ +export const boundRenderEventStream = ( + limits?: Partial, +): ( + stream: Stream.Stream, +) => Stream.Stream => { + const sequence = createAgentRenderEventSequence(limits); + return (stream: Stream.Stream) => + Stream.mapEffect(stream, (input) => + Effect.try({ + catch: (error) => toRuntimeError(error), + try: () => sequence.emit(input), + }), + ); +}; + +export const emitBoundRenderEvent = ( + sequence: ReturnType, + input: AgentRenderEventInput, +): Effect.Effect => + Effect.try({ + catch: (error) => toRuntimeError(error), + try: () => sequence.emit(input), + }); diff --git a/packages/rsc-runtime/src/reconciler.ts b/packages/rsc-runtime/src/reconciler.ts index 83da7fb33..94ac8b2b2 100644 --- a/packages/rsc-runtime/src/reconciler.ts +++ b/packages/rsc-runtime/src/reconciler.ts @@ -1,3 +1,4 @@ +import { Effect, 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'; @@ -6,10 +7,24 @@ import { createAgentRenderEventSequence, type AgentRenderError, type AgentRenderEvent, - type AgentRenderEventSequence, + type AgentRenderEventInput, type AgentRenderLimits, } from './agent-document.js'; +import type { AgentProgressReporter, AgentProgressUpdate } from './agent-request.js'; import { decodeAgentDocument } from './decode-document.js'; +import { + abortToInterrupt, + isAbortError, + runPromise, + scopedAbortSignal, + streamToReadableStream, + toRuntimeError, +} from './effect/boundary.js'; +import { + createFlightDemand, + emitBoundRenderEvent, + type FlightDemand, +} from './effect/render-stream.js'; import { ensureAgentFlightManifest } from './flight-manifest.js'; const REACT_FRAGMENT = Symbol.for('react.fragment'); @@ -246,257 +261,247 @@ const materializeNode = (node: ReactNode, path: string, ctx: MaterializeContext) } }; -const snapshotTree = (root: ReactNode, ids: Map) => { +interface TreeSnapshot { + readonly pending: readonly PendingBoundary[]; + readonly rejected: MaterializeContext['rejected']; + readonly tree: ReactNode; +} + +const snapshotTree = (root: ReactNode, ids: Map): TreeSnapshot => { const ctx: MaterializeContext = { ids, pending: [], rejected: [] }; return { pending: ctx.pending, rejected: ctx.rejected, tree: materializeNode(root, '', ctx) }; }; -interface DemandGate { - readonly consume: () => void; - readonly notify: () => void; - readonly wait: () => Promise; -} +const hostError = (signal: AbortSignal, error: unknown): Error => + signal.aborted || isAbortError(error) ? abortError() : toRuntimeError(error); -const createDemandGate = (): DemandGate => { - let notifyWaiter: (() => void) | undefined; - let signaled = false; - return { - consume() { - signaled = false; - }, - notify() { - signaled = true; - const waiter = notifyWaiter; - notifyWaiter = undefined; - waiter?.(); - }, - wait() { - if (signaled) { - signaled = false; - return Promise.resolve(); - } - return new Promise((resolve) => { - notifyWaiter = () => { - signaled = false; - resolve(); - }; - }); - }, - }; -}; +const waitSettledBoundary = ( + pending: readonly PendingBoundary[], +): Effect.Effect<{ readonly boundary: PendingBoundary; readonly error?: unknown; readonly ok: boolean }, Error> => + Effect.raceAll( + pending.map((boundary) => + Effect.tryPromise({ + catch: (error) => error, + try: () => Promise.resolve(boundary.thenable), + }).pipe( + Effect.map(() => ({ boundary, ok: true as const })), + Effect.catch((error) => Effect.succeed({ boundary, error, ok: false as const })), + ), + ), + ); -export interface LiveEventStream { - readonly emit: (event: AgentRenderEvent) => Promise; - readonly fail: (error: unknown) => void; - readonly holdFlight: () => boolean; - readonly readable: ReadableStream; - readonly waitForFlightDemand: () => Promise; -} +type ReconcileState = + | { readonly kind: 'shell'; readonly snapshot: TreeSnapshot } + | { readonly kind: 'loop'; readonly snapshot: TreeSnapshot }; -const createLiveEventStream = (signal: AbortSignal): LiveEventStream => { - const buffer: AgentRenderEvent[] = []; - const space = createDemandGate(); - const data = createDemandGate(); - const flight = createDemandGate(); - let waitingPulls = 0; - let shellEmitted = false; - let failed: unknown; - let closed = false; - - const readable = new ReadableStream({ - cancel() { - closed = true; - space.notify(); - data.notify(); - flight.notify(); - }, - pull(controller) { - if (failed !== undefined) { - controller.error(failed); - return; - } - const deliver = (): void => { - if (failed !== undefined) { - controller.error(failed); - return; +const reconcileInputStream = (root: ReactNode): Stream.Stream => { + const ids = new Map(); + const initial = snapshotTree(root, ids); + return Stream.paginate( + { kind: 'shell', snapshot: initial } satisfies ReconcileState, + (state): Effect.Effect< + readonly [readonly AgentRenderEventInput[], Option.Option], + Error + > => { + switch (state.kind) { + case 'shell': + return Effect.try({ + catch: (error) => toRuntimeError(error), + try: () => + [ + [{ document: decodeAgentDocument(state.snapshot.tree), type: 'shell' as const }], + Option.some({ kind: 'loop' as const, snapshot: state.snapshot }), + ] as const, + }); + case 'loop': { + if (state.snapshot.pending.length === 0) { + return Effect.try({ + catch: (error) => toRuntimeError(error), + try: () => + [ + [{ document: decodeAgentDocument(state.snapshot.tree), type: 'complete' as const }], + Option.none(), + ] as const, + }); + } + return waitSettledBoundary(state.snapshot.pending).pipe( + Effect.flatMap((settled) => + Effect.try({ + catch: (error) => toRuntimeError(error), + try: () => { + const snapshot = snapshotTree(root, ids); + const input: AgentRenderEventInput = settled.ok + ? { + boundaryId: settled.boundary.id, + document: decodeAgentDocument(snapshot.tree), + type: 'replace', + } + : { + boundaryId: settled.boundary.id, + error: renderErrorFrom(settled.error), + type: 'error', + }; + return [[input], Option.some({ kind: 'loop' as const, snapshot })] as const; + }, + }), + ), + ); } - const event = buffer.shift(); - if (event === undefined) { - controller.close(); - return; + default: { + const exhaustive: never = state; + return exhaustive; } - if (event.type === 'shell') shellEmitted = true; - controller.enqueue(event); - space.notify(); - if (event.type === 'complete') controller.close(); - }; - if (buffer.length > 0) { - data.consume(); - deliver(); - return; - } - waitingPulls += 1; - flight.notify(); - return data.wait().then(() => { - waitingPulls = Math.max(0, waitingPulls - 1); - deliver(); - }); - }, - }, { highWaterMark: 0 }); - - return { - async emit(event) { - if (failed !== undefined) throw failed; - if (closed && event.type !== 'complete') { - throw new AgentContractError('handoff-required', 'The render is complete; later work requires a new invocation handoff'); - } - while (buffer.length >= 1) { - if (signal.aborted) throw abortError(); - await space.wait(); } - buffer.push(event); - if (event.type === 'shell') shellEmitted = true; - if (event.type === 'complete') closed = true; - data.notify(); - }, - fail(error) { - if (failed !== undefined || (closed && buffer.length === 0)) return; - failed = error; - closed = true; - data.notify(); - space.notify(); - flight.notify(); - }, - holdFlight() { - return shellEmitted && waitingPulls === 0; }, - readable, - async waitForFlightDemand() { - while (shellEmitted && waitingPulls === 0 && failed === undefined) { - await flight.wait(); - } - }, - }; + ); }; -const gateFlight = ( +const gatedFlightStream = ( flight: ReadableStream, - live: LiveEventStream, - signal: AbortSignal, -): ReadableStream => - flight.pipeThrough(new TransformStream({ - async transform(chunk, controller) { - await live.waitForFlightDemand(); - if (signal.aborted) throw abortError(); - controller.enqueue(chunk); - }, - }, { highWaterMark: 1 }, { highWaterMark: 0 }), { signal }); - -const nextBoundary = async ( - pending: readonly PendingBoundary[], - signal: AbortSignal, -): Promise<{ readonly boundary: PendingBoundary; readonly error?: unknown; readonly ok: boolean }> => { - if (signal.aborted) throw abortError(); - return Promise.race([ - new Promise((_, reject) => { - const onAbort = (): void => reject(abortError()); - if (signal.aborted) { - onAbort(); - return; - } - signal.addEventListener('abort', onAbort, { once: true }); - }), - ...pending.map(async (boundary) => { - try { - await boundary.thenable; - return { boundary, ok: true as const }; - } catch (error) { - return { boundary, error, ok: false as const }; - } + demand: FlightDemand, +): Stream.Stream => + Stream.unwrap( + Effect.gen(function*() { + const reader = yield* Effect.acquireRelease( + Effect.sync(() => flight.getReader()), + (handle) => Effect.promise(() => handle.cancel().then(() => undefined)), + ); + return Stream.unfold(undefined, () => + demand.wait.pipe( + Effect.flatMap(() => + Effect.tryPromise({ + catch: (error) => toRuntimeError(error), + try: () => reader.read(), + }), + ), + Effect.map((next) => (next.done ? undefined : [next.value, undefined] as const)), + ), + ); }), - ]); -}; + ); -const reconcile = async ( - root: ReactNode, - sequence: AgentRenderEventSequence, - live: LiveEventStream, +const decodeFlightRoot = ( + flight: ReadableStream, + demand: FlightDemand, signal: AbortSignal, -): Promise => { - const ids = new Map(); - let snapshot = snapshotTree(root, ids); - await live.emit(sequence.emit({ document: decodeAgentDocument(snapshot.tree), type: 'shell' })); - - while (snapshot.pending.length > 0) { - if (signal.aborted) throw abortError(); - const settled = await nextBoundary(snapshot.pending, signal); - if (!settled.ok) { - const error = renderErrorFrom(settled.error); - await live.emit(sequence.emit({ - boundaryId: settled.boundary.id, - error, - type: 'error', - })); - } - snapshot = snapshotTree(root, ids); - if (settled.ok) { - await live.emit(sequence.emit({ - boundaryId: settled.boundary.id, - document: decodeAgentDocument(snapshot.tree), - type: 'replace', - })); - } - } +): Effect.Effect => + Effect.gen(function*() { + ensureAgentFlightManifest(); + const flightAbort = yield* scopedAbortSignal; + const readable = streamToReadableStream(gatedFlightStream(flight, demand), { + signal: flightAbort, + strategy: { highWaterMark: 1 }, + }); + return yield* Effect.tryPromise({ + catch: (error) => hostError(signal, error), + try: () => + createFromReadableStream(readable, { unstable_allowPartialStream: true }), + }); + }); - if (signal.aborted) throw abortError(); - await live.emit(sequence.emit({ document: decodeAgentDocument(snapshot.tree), type: 'complete' })); -}; +const progressInput = (update: AgentProgressUpdate): AgentRenderEventInput => ({ + completed: update.completed ?? 0, + ...(update.message === undefined ? {} : { message: update.message }), + ...(update.total === undefined ? {} : { total: update.total }), + type: 'progress', +}); -export interface AgentFlightEventSession { - readonly live: LiveEventStream; - readonly readable: ReadableStream; - readonly sequence: AgentRenderEventSequence; +export interface AgentRenderEventStreamOptions { + readonly demand: FlightDemand; + readonly flight: Promise>; + readonly limits?: Partial; + readonly signal: AbortSignal; } -export const createAgentFlightEventSession = ( - options: AgentFlightDecodeOptions = {}, -): AgentFlightEventSession => { - const signal = options.signal ?? new AbortController().signal; - const live = createLiveEventStream(signal); - return Object.freeze({ - live, - readable: live.readable, - sequence: createAgentRenderEventSequence(options.limits), +export interface AgentRenderEventSession { + readonly events: Stream.Stream; + readonly progress: AgentProgressReporter; +} + +/** + * Invocation-local render pipeline: Flight bytes as a pull-gated Stream, + * pending boundaries as `Stream.paginate`, contract bounds as the emit + * stage. `progress` is created synchronously so the host can execute in the + * same turn as `stream()`. + */ +export const createAgentRenderEventSession = ( + options: AgentRenderEventStreamOptions, +): AgentRenderEventSession => { + const sequence = createAgentRenderEventSequence(options.limits); + let offerProgress: ((input: AgentRenderEventInput) => Effect.Effect) | undefined; + const bufferedProgress: AgentRenderEventInput[] = []; + const progress: AgentProgressReporter = Object.freeze({ + report: async (update: AgentProgressUpdate) => { + if (sequence.completed) { + throw new AgentContractError( + 'handoff-required', + 'The render is complete; later work requires a new invocation handoff', + ); + } + const input = progressInput(update); + if (offerProgress === undefined) { + bufferedProgress.push(input); + return; + } + await runPromise(offerProgress(input)); + }, }); + const events = Stream.unwrap( + Effect.gen(function*() { + const progressInputs = yield* Queue.unbounded(); + offerProgress = (input) => Queue.offer(progressInputs, input).pipe(Effect.asVoid); + for (const input of bufferedProgress) { + yield* Queue.offer(progressInputs, input); + } + const flight = yield* Effect.tryPromise({ + catch: (error) => hostError(options.signal, error), + try: () => options.flight, + }); + if (options.signal.aborted) return yield* Effect.fail(abortError()); + const root = yield* decodeFlightRoot(flight, options.demand, options.signal); + if (options.signal.aborted) return yield* Effect.fail(abortError()); + return Stream.merge( + reconcileInputStream(root), + Stream.fromQueue(progressInputs), + { haltStrategy: 'left' }, + ).pipe( + Stream.mapEffect((input) => emitBoundRenderEvent(sequence, input)), + Stream.tap((event) => (event.type === 'shell' ? options.demand.markShell : Effect.void)), + Stream.takeUntil((event) => event.type === 'complete'), + Stream.ensuring(Queue.shutdown(progressInputs)), + ); + }), + ); + return { events, progress }; }; -const decodeIntoSession = ( - flight: ReadableStream, - session: AgentFlightEventSession, + +export const toPublicEventStream = ( + events: Stream.Stream, + demand: FlightDemand, signal: AbortSignal, -): void => { - ensureAgentFlightManifest(); - void (async () => { - try { - if (signal.aborted) throw abortError(); - const node = await createFromReadableStream( - gateFlight(flight, session.live, signal), - { unstable_allowPartialStream: true }, - ); - if (signal.aborted) throw abortError(); - await reconcile(node, session.sequence, session.live, signal); - } catch (error) { - session.live.fail(signal.aborted ? abortError() : error); - } - })(); -}; +): ReadableStream => + streamToReadableStream(Stream.interruptWhen(events, abortToInterrupt(signal)), { + closeOn: (event) => event.type === 'complete', + onPull: demand.notePull, + onPullDelivered: demand.notePullEnd, + strategy: { highWaterMark: 0 }, + }); export const decodeAgentFlightStream = ( flight: ReadableStream, - options: AgentFlightDecodeOptions & { readonly session?: AgentFlightEventSession } = {}, + options: AgentFlightDecodeOptions = {}, ): ReadableStream => { const signal = options.signal ?? new AbortController().signal; - const session = options.session ?? createAgentFlightEventSession({ limits: options.limits, signal }); - decodeIntoSession(flight, session, signal); - return session.readable; + const demand = createFlightDemand(); + return toPublicEventStream( + createAgentRenderEventSession({ + demand, + flight: Promise.resolve(flight), + limits: options.limits, + signal, + }).events, + demand, + signal, + ); }; diff --git a/packages/rsc-runtime/tests/agent-document.test.ts b/packages/rsc-runtime/tests/agent-document.test.ts index d4f5f9dc3..8bb30c3da 100644 --- a/packages/rsc-runtime/tests/agent-document.test.ts +++ b/packages/rsc-runtime/tests/agent-document.test.ts @@ -1,3 +1,4 @@ +import { Stream } from 'effect'; import { describe, expect, it } from '@rstest/core'; import { @@ -8,6 +9,8 @@ import { type AgentDocumentNode, type AgentRenderInvocation, } from '../src/index.js'; +import { runPromise } from '../src/effect/boundary.js'; +import { boundRenderEventStream } from '../src/effect/render-stream.js'; const root = (): AgentDocumentNode => ({ children: [ @@ -201,6 +204,28 @@ describe('Agent render events', () => { }); }); +describe('boundRenderEventStream', () => { + it('assigns sequence numbers and fails closed after complete', async () => { + const events = await runPromise(Stream.runCollect( + Stream.make( + { completed: 0, type: 'progress' as const }, + { completed: 1, type: 'progress' as const }, + ).pipe(boundRenderEventStream()), + )); + expect(events.map((event) => event.sequence)).toEqual([0, 1]); + + await expect(runPromise(Stream.runCollect( + Stream.make( + { + document: { root: root(), status: 'success' as const, version: 1 as const }, + type: 'complete' as const, + }, + { completed: 2, type: 'progress' as const }, + ).pipe(boundRenderEventStream()), + ))).rejects.toMatchObject({ code: 'handoff-required' }); + }); +}); + describe('AgentRenderInvocation', () => { it('discriminates typed props for every invocation kind', () => { expect([ diff --git a/packages/rsc-runtime/tests/effect-boundary.test.ts b/packages/rsc-runtime/tests/effect-boundary.test.ts index ea88f4005..0b1d95ff0 100644 --- a/packages/rsc-runtime/tests/effect-boundary.test.ts +++ b/packages/rsc-runtime/tests/effect-boundary.test.ts @@ -1,10 +1,12 @@ -import { Cause, Effect, Exit } from 'effect'; +import { Cause, Effect, Exit, Stream } from 'effect'; import { describe, expect, it } from '@rstest/core'; import { AgentContractError } from '../src/agent-document.js'; import { AgentRequestError } from '../src/agent-request.js'; + import { abortError, + abortToInterrupt, interruptWhenAborted, isAbortError, isTypedRuntimeError, @@ -12,6 +14,7 @@ import { runPromise, runPromiseExit, runSync, + streamToReadableStream, toRuntimeError, } from '../src/effect/boundary.js'; import * as runtime from '../src/index.js'; @@ -73,4 +76,59 @@ describe('effect boundary', () => { expect(toRuntimeError('plain')).toEqual(new Error('plain')); expect(abortError().name).toBe('AbortError'); }); + + it('interrupts when the host signal aborts between construction and run', async () => { + const controller = new AbortController(); + const program = interruptWhenAborted(Effect.never, controller.signal); + controller.abort(); + await expect(runPromise(program)).rejects.toSatisfy(isAbortError); + await expect(runPromise(abortToInterrupt(controller.signal))).rejects.toSatisfy(isAbortError); + }); + + it('fails the readable when the stream fails', async () => { + const readable = streamToReadableStream(Stream.fail(new AgentContractError('event-count-exceeded', 'too many'))); + const reader = readable.getReader(); + await expect(reader.read()).rejects.toMatchObject({ code: 'event-count-exceeded' }); + }); + + it('fails the readable after a successful event when the stream fails', async () => { + const readable = streamToReadableStream( + Stream.make({ type: 'shell' as const }).pipe( + Stream.concat(Stream.fail(new AgentContractError('event-count-exceeded', 'too many'))), + ), + { strategy: { highWaterMark: 0 } }, + ); + const reader = readable.getReader(); + const first = await reader.read(); + expect(first.value).toEqual({ type: 'shell' }); + await expect(reader.read()).rejects.toMatchObject({ code: 'event-count-exceeded' }); + }); + + it('maps stream interruption to AbortError without a complete event', async () => { + const controller = new AbortController(); + const readable = streamToReadableStream(Stream.never, { signal: controller.signal }); + const reader = readable.getReader(); + const pending = reader.read(); + controller.abort(); + await expect(pending).rejects.toSatisfy(isAbortError); + }); + + it('rejects a later read after the producer has already failed', async () => { + const stream = Stream.succeed('shell').pipe( + Stream.concat(Stream.fromEffect(Effect.sleep('20 millis').pipe( + Effect.andThen(Effect.fail(new AgentContractError('event-count-exceeded', 'too many'))), + ))), + ); + const readable = streamToReadableStream(stream, { strategy: { highWaterMark: 0 } }); + const reader = readable.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + expect(first.value).toBe('shell'); + await new Promise((resolve) => { setTimeout(resolve, 50); }); + await expect(reader.read()).rejects.toMatchObject({ + name: 'AgentContractError', + code: 'event-count-exceeded', + }); + }); + }); diff --git a/packages/rsc-runtime/tests/state-packaging.test.ts b/packages/rsc-runtime/tests/state-packaging.test.ts index 0391ed1aa..68b42cba4 100644 --- a/packages/rsc-runtime/tests/state-packaging.test.ts +++ b/packages/rsc-runtime/tests/state-packaging.test.ts @@ -20,12 +20,19 @@ const distFile = async (...segments: string[]): Promise => describe.sequential('state kernel packaging boundaries', () => { it('keeps every kernel and storage identifier out of the root and plugin entries', async () => { + const kernel = ['node:sqlite', 'defineState', 'AgentStateError', 'DatabaseSync', 'agent_state_journal'] as const; for (const entry of ['index.js', 'plugin.js']) { const source = await distFile(entry); - for (const identifier of ['node:sqlite', 'defineState', 'AgentStateError', 'DatabaseSync', 'agent_state_journal', 'from "effect"', 'Effect.runPromise']) { + for (const identifier of kernel) { expect(source, `${entry} must not contain ${identifier}`).not.toContain(identifier); } } + // Stage 2 puts Effect on the dispatcher, which is part of the root graph. + // The plugin entry stays Effect-free so hook-only artifacts still skip it. + const plugin = await distFile('plugin.js'); + for (const identifier of ['from "effect"', 'Effect.runPromise']) { + expect(plugin, `plugin.js must not contain ${identifier}`).not.toContain(identifier); + } }); it('keeps node:sqlite out of the volatile state entry', async () => { From cae0ed4562b534d27d5d30da290d5a3eedaecc0f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 16:51:19 +0000 Subject: [PATCH 2/3] fix(build): emit runtime types before agent-bundle declaration emit agent-bundle's test helpers import @agent-bundle/runtime; CI was typechecking that package before runtime dist existed, which is why main's post-#154 Verify jobs fail in ~30s. --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 8e2df5df5..bf2ae7410 100644 --- a/package.json +++ b/package.json @@ -8,7 +8,7 @@ }, "packageManager": "pnpm@11.23.0", "scripts": { - "build": "pnpm --filter agent-bundle build && pnpm --filter @agent-bundle/runtime build && pnpm --filter create-agent-bundle build", + "build": "pnpm --filter @agent-bundle/runtime build && pnpm --filter agent-bundle build && pnpm --filter create-agent-bundle build", "lint:package": "publint packages/agent-bundle && publint packages/rsc-runtime && publint packages/create-agent-bundle", "test": "pnpm test:unit && pnpm test:route-unit && pnpm test:integration", "test:unit": "rstest --config rstest.unit.config.ts", From 28044bb7e511fd8e59b31b0d3c9a56fef7e3cfed Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 17:07:58 +0000 Subject: [PATCH 3/3] docs(runtime): record the verified locked-stream teardown mechanism in comments The maxEvents hang was not a blocking-interrupt deadlock: the acquireRelease finalizer called cancel() on the Flight readable while the Flight client held its reader, the locked-stream rejection became an Effect.promise defect during scope close, and the event stream's exit never surfaced. Document that at the scopedAbortSignal fix site and correct the boundary cancel() comment. --- packages/rsc-runtime/src/effect/boundary.ts | 7 ++++--- packages/rsc-runtime/src/reconciler.ts | 6 ++++++ 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/rsc-runtime/src/effect/boundary.ts b/packages/rsc-runtime/src/effect/boundary.ts index dc76aa41a..9fd4bbb10 100644 --- a/packages/rsc-runtime/src/effect/boundary.ts +++ b/packages/rsc-runtime/src/effect/boundary.ts @@ -214,9 +214,10 @@ export const streamToReadableStream = ( const running = fiber; fiber = undefined; if (running === undefined) return; - // Never `runPromise` here: this cancel is invoked from an Effect - // acquireRelease finalizer (Flight bytes). Blocking on interrupt - // deadlocks the parent fiber and hides contract / abort errors. + // Fork, never `runPromise`: cancel may run inside Effect teardown + // (or a consumer's cancel path), and blocking a finalizer on the + // producer fiber's exit risks deadlock and hides contract / abort + // errors. Interrupt in the background and return immediately. void Effect.runFork(Effect.asVoid(Fiber.interrupt(running))); }, pull() { diff --git a/packages/rsc-runtime/src/reconciler.ts b/packages/rsc-runtime/src/reconciler.ts index 94ac8b2b2..6b9f0c40e 100644 --- a/packages/rsc-runtime/src/reconciler.ts +++ b/packages/rsc-runtime/src/reconciler.ts @@ -387,6 +387,12 @@ const decodeFlightRoot = ( ): Effect.Effect => Effect.gen(function*() { ensureAgentFlightManifest(); + // Teardown must interrupt the Flight source, never `readable.cancel()`: + // the Flight client below holds this readable's reader, so cancel() on + // the locked stream rejects (ERR_INVALID_STATE), and that rejection + // would defect the closing scope and wedge the event stream's exit + // (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), { signal: flightAbort,