Preserve stream actor context and integrate reviewed CI fixes

.changeset/stream-callback-actor-gates.md

@@ -0,0 +1,8 @@
+---
+"@mcp-b/do-runtime": patch
+---
+
+Preserve the creating actor's input gate and async context in ReadableStream
+and TransformStream callbacks. Delayed input and stream demand previously entered
+provider callbacks or tool execution without their creating scope, causing
+valid model turns, title actions, and routine actions to fail at storage access.

.github/workflows/ci.yml

@@ -28,7 +28,7 @@ jobs:
           persist-credentials: false
 
       - name: Set up pnpm
-        uses: pnpm/action-setup@0977fd99725f1db4007ccb2928dbb4e90d06cc86 # v6.0.10
+        uses: pnpm/action-setup@ea17c68df8912ef543352723c149a84f56e3d413 # v6.1.0
 
       - name: Set up Node.js
         uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0

.github/workflows/release.yml

@@ -32,7 +32,7 @@ jobs:
           fetch-depth: 0
 
       - name: Set up pnpm
-        uses: pnpm/action-setup@0977fd99725f1db4007ccb2928dbb4e90d06cc86 # v6.0.10
+        uses: pnpm/action-setup@ea17c68df8912ef543352723c149a84f56e3d413 # v6.1.0
 
       - name: Set up Node.js
         uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0

docs/browser-async-context.md

@@ -13,6 +13,13 @@ its caller's scope in `finally`; it never leaves one mutable store active while
 a promise is pending. Actor entry, reentry, critical sections and timer callbacks
 capture that scope before the runtime waits for its input lock.
 
+`installActorScope` also binds native `ReadableStream` and `TransformStream`
+callbacks. Sources and transformers created inside an actor retain its actor
+and async stores when later demand or network chunks arrive. Pull, transform,
+flush and cancel re-enter through the existing input gate; start remains synchronous.
+Streams constructed outside an actor retain native behavior. This covers the
+AI SDK's streamed tool execution, whose callbacks bypass transformed awaits.
+
 This requires compilation. Native `await` bypasses `Promise.prototype.then`, as
 the [TC39 async-context proposal](https://github.com/tc39/proposal-async-context)
 explains. The opt-in Vite transform first gates awaits, then uses Vite's installed

docs/gating-coverage.md

@@ -32,6 +32,7 @@ enumerates it. Every row is one of:
 | `body.values()` / async iteration | iterator reads through the gated reader; early return preserves native cancel and lock-release semantics | `api/http.ts` |
 | Reader/stream lifecycle (`reader.closed`, both `cancel()` methods) | settlement uses `awaitIo`; `closed` is gated and registered once | `api/http.ts` |
 | `body.tee()` | both halves re-gated | `api/http.ts` |
+| Actor-created `ReadableStream` / `TransformStream` callbacks | constructor captures the current actor and async stores; `pull`, `transform`, `flush`, and `cancel` use `makeReentryCallback`; synchronous `start` retains its native timing and receiver | `api/global-scope.ts`; delayed input and external consumer regressions in `global-scope.test.ts` |
 | `body.pipeThrough()` / `pipeTo()` | returned readable re-gated (recurses through chains); settlement `awaitIo`d — native pipe machinery bypasses the `getReader` override and would launder the stream | `api/http.ts`, 0.2.2 |
 | `setTimeout` / `setInterval` | arming captures the critical section; firing re-enters via `ctx.run` | `api/global-scope.ts` |
 | `scheduler.wait()` / `scheduler.yield()` | scoped `Scheduler` over the same timer path | `api/global-scope.ts` |

examples/vibe-platform/scripts/e2e.mjs

@@ -131,10 +131,23 @@ page.on("console", (message) => {
 
 const preview = () => page.frameLocator("#preview");
 const file = (path) => page.locator(`#files button[data-path="${path}"]`);
-const builtOk = () =>
-  page.waitForFunction(() => /^built in \d+ms$/.test(document.querySelector("#status").textContent), {
-    timeout: TIMEOUT,
-  });
+// Register before the action: the build status updates before srcdoc navigation commits.
+const builtOk = async (run) => {
+  const [frame] = await Promise.all([
+    page.waitForEvent("framenavigated", {
+      predicate: (frame) => frame.parentFrame() === page.mainFrame() && frame.url() === "about:srcdoc",
+      timeout: TIMEOUT,
+    }),
+    run(),
+  ]);
+  // The bridge is installed before #root is parsed; no CDN resources are needed.
+  await frame.locator("#root").waitFor({ state: "attached", timeout: TIMEOUT });
+  await page.waitForFunction(
+    () => /^built in \d+ms$/.test(document.querySelector("#status").textContent),
+    undefined,
+    { timeout: TIMEOUT },
+  );
+};
 
 const previewApi = (method = "GET") =>
   preview()
@@ -156,7 +169,7 @@ let exportDirectory;
 try {
   // 1 -----------------------------------------------------------------------
   await step("page loads and the actor seeded its starter files", async () => {
-    await page.goto(url, { waitUntil: "domcontentloaded" });
+    await builtOk(() => page.goto(url, { waitUntil: "domcontentloaded" }));
     for (const seeded of [
       "/server/agent.ts",
       "/src/main.tsx",
@@ -200,8 +213,7 @@ try {
 
   // 3 -----------------------------------------------------------------------
   await step("build & run bundles the workspace out of the Durable Object", async () => {
-    await page.locator("#build").click();
-    await builtOk();
+    await builtOk(() => page.locator("#build").click());
   });
 
   await step("the seeded front-end reaches the user Agent through /api/*", async () => {
@@ -248,8 +260,7 @@ try {
     agentSourceForExport = source.replace("a quiet hello", "a cheerful hello");
     if (agentSourceForExport === source) throw new Error("agent edit marker was missing");
     await page.locator("#editor").fill(agentSourceForExport);
-    await page.locator("#save").click();
-    await builtOk();
+    await builtOk(() => page.locator("#save").click());
     await page.waitForFunction(
       () => document.querySelector("#log").textContent.includes("agent restarted; storage intact"),
       undefined,
@@ -277,8 +288,7 @@ try {
     }
 
     await page.locator("#editor").fill(agentSourceForExport);
-    await page.locator("#save").click();
-    await builtOk();
+    await builtOk(() => page.locator("#save").click());
     const state = await previewApi();
     if (state.visits !== 1) throw new Error(`visit count after recovery is ${String(state.visits)}`);
   });
@@ -302,8 +312,7 @@ try {
     );
 
     await page.locator("#editor").fill(agentSourceForExport);
-    await page.locator("#save").click();
-    await builtOk();
+    await builtOk(() => page.locator("#save").click());
     const state = await previewApi();
     if (state.visits !== 1) {
       throw new Error(`visit count after constructor recovery is ${String(state.visits)}`);
@@ -321,8 +330,7 @@ try {
     );
     const source = await editor.inputValue();
     await editor.fill(source.replace(TITLE_BEFORE, TITLE_AFTER));
-    await page.locator("#save").click();
-    await builtOk();
+    await builtOk(() => page.locator("#save").click());
   });
 
   if (!online) {
@@ -338,8 +346,7 @@ try {
 
   // 6 -----------------------------------------------------------------------
   await step("the edit survives a page reload (OPFS persistence)", async () => {
-    await page.reload({ waitUntil: "domcontentloaded" });
-    await builtOk();
+    await builtOk(() => page.reload({ waitUntil: "domcontentloaded" }));
     await file("/src/App.tsx").waitFor({ timeout: TIMEOUT });
     await file("/src/App.tsx").click();
     await page.waitForFunction(

src/api/global-scope.test.ts

@@ -25,6 +25,73 @@ import {
   NO_GLOBAL_OUTBOUND_MESSAGE,
 } from "./global-scope";
 import { HibernatableWebSocketRegistry } from "./web-socket";
+import { AsyncLocalStorage } from "../browser/async-hooks";
+
+test("readable stream callbacks re-enter their creator when consumed outside its actor", async () => {
+  const { ctx, scope } = newScope();
+  const target = { ReadableStream: globalThis.ReadableStream };
+  installActorScope(target, () => scope);
+  const store = new AsyncLocalStorage<string>();
+  const seen: unknown[] = [];
+  const source: UnderlyingDefaultSource<string> = {
+    start() {
+      seen.push(["start", ctx.hasCurrent(), store.getStore(), this === source]);
+    },
+    pull(controller) {
+      seen.push(["pull", ctx.hasCurrent(), store.getStore(), this === source]);
+      controller.enqueue("chunk");
+    },
+    cancel() {
+      seen.push(["cancel", ctx.hasCurrent(), store.getStore(), this === source]);
+    },
+  };
+  const stream = await ctx.run(() =>
+    store.run("creator", () => new target.ReadableStream(Object.freeze(source), { highWaterMark: 0 })),
+  );
+  await new Promise((resolve) => setTimeout(resolve, 0));
+  expect(ctx.hasCurrent()).toBe(false);
+  const reader = stream.getReader();
+  expect(await store.run("unrelated", () => reader.read())).toEqual({ value: "chunk", done: false });
+  await reader.cancel();
+  expect(seen).toEqual([
+    ["start", true, "creator", true],
+    ["pull", true, "creator", true],
+    ["cancel", true, "creator", true],
+  ]);
+  expect(store.getStore()).toBeUndefined();
+});
+
+test("stream callbacks retain their creator's actor and async scope after delayed input", async () => {
+  const { ctx, scope } = newScope();
+  const target = { TransformStream: globalThis.TransformStream };
+  installActorScope(target, () => scope);
+  const store = new AsyncLocalStorage<string>();
+  const seen: unknown[] = [];
+  const transformer: Transformer<string, string> = {
+    transform(value, controller) {
+      seen.push(["transform", ctx.hasCurrent(), store.getStore(), this === transformer]);
+      controller.enqueue(value);
+    },
+    flush() {
+      seen.push(["flush", ctx.hasCurrent(), store.getStore(), this === transformer]);
+    },
+  };
+  const stream = await ctx.run(() =>
+    store.run("creator", () => new target.TransformStream(Object.freeze(transformer))),
+  );
+  await new Promise((resolve) => setTimeout(resolve, 0));
+  const reader = stream.readable.getReader();
+  const writer = stream.writable.getWriter();
+  const read = reader.read();
+  await store.run("unrelated", () => writer.write("tool call"));
+  expect(await read).toEqual({ value: "tool call", done: false });
+  await writer.close();
+  expect(seen).toEqual([
+    ["transform", true, "creator", true],
+    ["flush", true, "creator", true],
+  ]);
+  expect(store.getStore()).toBeUndefined();
+});
 
 describe("AlarmInvocationInfo", () => {
   test("carries the scheduled time and the retry count", () => {
@@ -465,6 +532,8 @@ describe("installActorScope", () => {
     installActorScope(target, () => scope);
 
     expect(Object.keys(target).sort()).toEqual([
+      "ReadableStream",
+      "TransformStream",
       "WebSocket",
       "WebSocketPair",
       "WebSocketRequestResponsePair",

src/api/global-scope.ts

@@ -47,6 +47,7 @@ import {
   EXCEPTION_DURABLE_OBJECT_ABORT_NO_RETRY,
   hasUserErrorDetail,
   isExceptionFromInputGateBroken,
+  tryCurrentIoContext,
   tryCurrentSlice,
   type IoContext,
 } from "../io/io-context";
@@ -531,6 +532,8 @@ export type ActorScopeBindings = {
   readonly WebSocket: typeof globalThis.WebSocket;
   readonly WebSocketPair: WebSocketPairConstructor;
   readonly WebSocketRequestResponsePair: typeof WebSocketRequestResponsePair;
+  readonly ReadableStream: typeof globalThis.ReadableStream;
+  readonly TransformStream: typeof globalThis.TransformStream;
   readonly currentExternalEntry?: object | undefined;
 };
 
@@ -572,6 +575,8 @@ export function actorScopeBindings(resolve: () => ActorGlobalScope): ActorScopeB
     }),
     WebSocketPair: BoundWebSocketPair,
     WebSocketRequestResponsePair,
+    ReadableStream: ActorReadableStream,
+    TransformStream: ActorTransformStream,
     get currentExternalEntry(): object | undefined {
       return resolve().currentExternalEntry;
     },
@@ -613,6 +618,45 @@ function scopeCrypto(resolve: () => ActorGlobalScope): Crypto {
 /** Captured at import, before any host installs a scope over it. */
 const platformCrypto = globalThis.crypto;
 
+// Native stream callbacks bypass transformed awaits. Capture their creator,
+// including async stores, before later input or demand arrives from another actor.
+// These constructors are captured before a host installs its actor scope.
+const ActorReadableStream = withActorStreamCallbacks(globalThis.ReadableStream);
+const ActorTransformStream = withActorStreamCallbacks(globalThis.TransformStream);
+
+function withActorStreamCallbacks<T extends typeof ReadableStream | typeof TransformStream>(
+  Stream: T,
+): T {
+  return new Proxy(Stream, {
+    construct(target, args, newTarget) {
+      const context = tryCurrentIoContext();
+      const [source, ...strategies] = args;
+      if (
+        !context || source == null ||
+        (typeof source !== "object" && typeof source !== "function")
+      ) {
+        return Reflect.construct(target, args, newTarget);
+      }
+      // A separate target preserves frozen sources and inherited getters.
+      const callbacks = new Proxy({}, {
+        get(_target, name) {
+          const callback = Reflect.get(source, name, source);
+          if (typeof callback !== "function") return callback;
+          // start runs synchronously during construction, already inside the actor.
+          if (name === "start") return callback.bind(source);
+          if (name === "pull" || name === "transform" || name === "flush" || name === "cancel") {
+            return context.makeReentryCallback((_lock, ...values: unknown[]) =>
+              Reflect.apply(callback, source, values),
+            );
+          }
+          return callback;
+        },
+      });
+      return Reflect.construct(target, [callbacks, ...strategies], newTarget);
+    },
+  });
+}
+
 /** ← every `SubtleCrypto` member that returns a promise, as a value the binding can iterate. */
 const ASYNC_SUBTLE_METHODS = [
   "decrypt",

src/api/http.ts

@@ -246,8 +246,9 @@ export function gateReadableStream<T>(ctx: IoAwaiter, stream: ReadableStream<T>)
     // internal spec operations (not the `getReader` property above), and `pipeThrough` hands
     // back the transform's readable — a brand-new stream with none of this instrumentation.
     // `res.body.pipeThrough(new TextDecoderStream()).getReader().read()` would resume foreign
-    // on every chunk. The pipe's own internals never surface a user continuation, so the two
-    // seams that do are the ones gated: the returned readable, and `pipeTo`'s settlement.
+    // on every chunk. Gate the returned readable and `pipeTo`'s settlement here;
+    // actorScopeBindings.TransformStream separately gates user transformer callbacks
+    // invoked by the native pipe machinery.
     pipeThrough: {
       configurable: true,
       writable: true,