refactor(runtime): Effect-native dispatcher and stream pipeline (wave 3.5 stage 2)

.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.

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
 

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

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

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.

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 |
 | --- | --- | --- |

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",

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<AgentRenderEvent> =>
+  new ReadableStream({
+    start(controller) {
+      controller.error(abortError());
+    },
+  });
+
 const drainCompleteDocument = async (
   events: ReadableStream<AgentRenderEvent>,
   signal: AbortSignal,
@@ -69,43 +77,27 @@ export const createAgentRenderDispatcher = (
   options: AgentRenderDispatcherOptions = {},
 ): AgentRenderDispatcher => {
   const stream = (request: AgentRenderDispatch): ReadableStream<AgentRenderEvent> => {
-    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<ReadableStream<Uint8Array>> } = {};
+    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({

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 = <R, E>(
   });
 };
 
+/**
+ * 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<never> =>
+  Effect.suspend(() => {
+    if (signal.aborted) return interruptAs();
+    return Effect.callback<never>((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,117 @@ export const makeScopedEffectRuntime = <R, E>(
 export const interruptWhenAborted = <A, E, R>(
   effect: Effect.Effect<A, E, R>,
   signal: AbortSignal,
-): Effect.Effect<A, E, R> => {
-  if (signal.aborted) return interruptAs();
-  return Effect.raceFirst(
-    effect,
-    Effect.callback<never>((resume) => {
-      const onAbort = () => {
-        resume(Effect.interrupt);
-      };
-      signal.addEventListener('abort', onAbort, { once: true });
-      return Effect.sync(() => {
-        signal.removeEventListener('abort', onAbort);
+): Effect.Effect<A, E, R> => Effect.raceFirst(effect, abortToInterrupt(signal));
+
+export interface StreamToReadableOptions<A> {
+  readonly closeOn?: (value: A) => boolean;
+  readonly onPull?: () => void;
+  readonly onPullDelivered?: () => void;
+  readonly signal?: AbortSignal;
+  readonly strategy?: QueuingStrategy<A>;
+}
+
+/**
+ * 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 = <A, E>(
+  stream: Stream.Stream<A, E>,
+  options: StreamToReadableOptions<A> = {},
+): ReadableStream<A> => {
+  let currentPull: { readonly resolve: () => void; readonly reject: (error: Error) => void } | undefined;
+  let fiber: Fiber.Fiber<void, E> | 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<A>, error: Error): void => {
+    try {
+      controller.error(error);
+    } catch {
+      // Already closed or errored — pull() still rejects via `terminal`.
+    }
+    finish(error);
+  };
+
+  return new ReadableStream<A>({
+    cancel() {
+      const running = fiber;
+      fiber = undefined;
+      if (running === undefined) return;
+      // 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() {
+      if (terminal !== undefined) {
+        return terminal.error === undefined ? Promise.resolve() : Promise.reject(terminal.error);
+      }
+      options.onPull?.();
+      return new Promise<void>((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);
 };
 
 /**