From 49e486ec843893f6ea3b0c6c25b8f92faed37368 Mon Sep 17 00:00:00 2001 From: Alex Nahas Date: Mon, 7 Sep 2026 22:42:39 -0700 Subject: [PATCH 1/5] Wait for preview navigation before exercising the example API The built status is written immediately after assigning iframe.srcdoc, while the previous document can still be active. Register for navigation before each build, save, and reload, then wait for the new preview root so API calls use its installed bridge. This also synchronizes the initial automatic build and uses the correct Playwright timeout argument. --- examples/vibe-platform/scripts/e2e.mjs | 41 +++++++++++++++----------- 1 file changed, 24 insertions(+), 17 deletions(-) diff --git a/examples/vibe-platform/scripts/e2e.mjs b/examples/vibe-platform/scripts/e2e.mjs index e518d23..0f71f69 100644 --- a/examples/vibe-platform/scripts/e2e.mjs +++ b/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( From 5fd09689ebec60972ca46bddca879273e55a720e Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 16:04:50 +0000 Subject: [PATCH 2/5] ci(deps): bump pnpm/action-setup in the actions group Bumps the actions group with 1 update: [pnpm/action-setup](https://github.com/pnpm/action-setup). Updates `pnpm/action-setup` from 6.0.10 to 6.1.0 - [Release notes](https://github.com/pnpm/action-setup/releases) - [Commits](https://github.com/pnpm/action-setup/compare/0977fd99725f1db4007ccb2928dbb4e90d06cc86...ea17c68df8912ef543352723c149a84f56e3d413) --- updated-dependencies: - dependency-name: pnpm/action-setup dependency-version: 6.1.0 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: actions ... Signed-off-by: dependabot[bot] --- .github/workflows/ci.yml | 2 +- .github/workflows/release.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index db9f646..8cb5c0b 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index df1717b..5d4594b 100644 --- a/.github/workflows/release.yml +++ b/.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 From a423fd285aa45af5236635b01b23a2314a11b172 Mon Sep 17 00:00:00 2001 From: Alex Nahas Date: Wed, 16 Sep 2026 21:12:33 -0700 Subject: [PATCH 3/5] Keep native stream callbacks inside their creating actor Delayed TransformStream input bypasses transformed awaits and can enter streamed tool execution without an actor input lock. Capture the creating context at construction and use the existing reentry gate for asynchronous callbacks, preserving native receivers and synchronous start timing. --- .changeset/stream-callback-actor-gates.md | 7 +++++ docs/browser-async-context.md | 7 +++++ docs/gating-coverage.md | 1 + src/api/global-scope.test.ts | 34 ++++++++++++++++++++++ src/api/global-scope.ts | 35 +++++++++++++++++++++++ src/api/http.ts | 5 ++-- 6 files changed, 87 insertions(+), 2 deletions(-) create mode 100644 .changeset/stream-callback-actor-gates.md diff --git a/.changeset/stream-callback-actor-gates.md b/.changeset/stream-callback-actor-gates.md new file mode 100644 index 0000000..6bbb440 --- /dev/null +++ b/.changeset/stream-callback-actor-gates.md @@ -0,0 +1,7 @@ +--- +"@mcp-b/do-runtime": patch +--- + +Preserve the creating actor's input gate and async context in TransformStream +callbacks. Delayed model response chunks previously entered tool execution +without a lock, causing valid title and routine actions to fail at SQL access. diff --git a/docs/browser-async-context.md b/docs/browser-async-context.md index 1cca27f..bbab8eb 100644 --- a/docs/browser-async-context.md +++ b/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 installs the native `TransformStream` constructor with +actor callback binding. A transformer created inside an actor retains its actor +and async stores when network chunks arrive later. Transform, flush and cancel +callbacks 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 diff --git a/docs/gating-coverage.md b/docs/gating-coverage.md index 0bf3fa1..5995881 100644 --- a/docs/gating-coverage.md +++ b/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 `TransformStream` callbacks | constructor captures the current actor and async stores; `transform`, `flush`, and `cancel` use `makeReentryCallback`; synchronous `start` retains its native timing and receiver | `api/global-scope.ts`; delayed input regression 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` | diff --git a/src/api/global-scope.test.ts b/src/api/global-scope.test.ts index 428ba73..02b6361 100644 --- a/src/api/global-scope.test.ts +++ b/src/api/global-scope.test.ts @@ -25,6 +25,39 @@ import { NO_GLOBAL_OUTBOUND_MESSAGE, } from "./global-scope"; import { HibernatableWebSocketRegistry } from "./web-socket"; +import { AsyncLocalStorage } from "../browser/async-hooks"; + +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(); + const seen: unknown[] = []; + const transformer: Transformer = { + 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 +498,7 @@ describe("installActorScope", () => { installActorScope(target, () => scope); expect(Object.keys(target).sort()).toEqual([ + "TransformStream", "WebSocket", "WebSocketPair", "WebSocketRequestResponsePair", diff --git a/src/api/global-scope.ts b/src/api/global-scope.ts index 67cd74c..f044b0a 100644 --- a/src/api/global-scope.ts +++ b/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,7 @@ export type ActorScopeBindings = { readonly WebSocket: typeof globalThis.WebSocket; readonly WebSocketPair: WebSocketPairConstructor; readonly WebSocketRequestResponsePair: typeof WebSocketRequestResponsePair; + readonly TransformStream: typeof globalThis.TransformStream; readonly currentExternalEntry?: object | undefined; }; @@ -572,6 +574,7 @@ export function actorScopeBindings(resolve: () => ActorGlobalScope): ActorScopeB }), WebSocketPair: BoundWebSocketPair, WebSocketRequestResponsePair, + TransformStream: ActorTransformStream, get currentExternalEntry(): object | undefined { return resolve().currentExternalEntry; }, @@ -613,6 +616,38 @@ function scopeCrypto(resolve: () => ActorGlobalScope): Crypto { /** Captured at import, before any host installs a scope over it. */ const platformCrypto = globalThis.crypto; +// Native stream machinery invokes transformer callbacks without going through +// getReader() or a transformed await. Capture the creating actor here, before +// network input arrives on a later task. Streams created outside actors remain +// native, and the constructor/prototype/brand and callback receiver are retained. +const ActorTransformStream = new Proxy(globalThis.TransformStream, { + construct(target, args, newTarget) { + const context = tryCurrentIoContext(); + const [transformer, ...strategies] = args; + if ( + !context || transformer == null || + (typeof transformer !== "object" && typeof transformer !== "function") + ) { + return Reflect.construct(target, args, newTarget); + } + const callbacks = new Proxy({}, { + get(_target, name) { + const callback = Reflect.get(transformer, name, transformer); + if (typeof callback !== "function") return callback; + // start runs synchronously during construction, already inside the actor. + if (name === "start") return callback.bind(transformer); + if (name === "transform" || name === "flush" || name === "cancel") { + return context.makeReentryCallback((_lock, ...values: unknown[]) => + Reflect.apply(callback, transformer, 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", diff --git a/src/api/http.ts b/src/api/http.ts index 8e8fa45..f6c7849 100644 --- a/src/api/http.ts +++ b/src/api/http.ts @@ -246,8 +246,9 @@ export function gateReadableStream(ctx: IoAwaiter, stream: ReadableStream) // 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, From 01102f6435ff5eb5002fb1ee00edc48cc908c1cf Mon Sep 17 00:00:00 2001 From: Alex Nahas Date: Wed, 16 Sep 2026 21:28:36 -0700 Subject: [PATCH 4/5] Bind stream source callbacks to their creator as well Full extension CI exposed the corresponding ReadableStream gap: source pull callbacks inherited consumer context after transformed chunks re-entered their own gates. Bind native sources and transformers through the same constructor seam. A focused regression reproduces the lost async store, and the real streamed Think example now completes tools, Stop, and recovery. --- .changeset/stream-callback-actor-gates.md | 7 ++- docs/browser-async-context.md | 8 +-- docs/gating-coverage.md | 2 +- src/api/global-scope.test.ts | 29 +++++++++ src/api/global-scope.ts | 71 +++++++++++++---------- 5 files changed, 78 insertions(+), 39 deletions(-) diff --git a/.changeset/stream-callback-actor-gates.md b/.changeset/stream-callback-actor-gates.md index 6bbb440..4f917f6 100644 --- a/.changeset/stream-callback-actor-gates.md +++ b/.changeset/stream-callback-actor-gates.md @@ -2,6 +2,7 @@ "@mcp-b/do-runtime": patch --- -Preserve the creating actor's input gate and async context in TransformStream -callbacks. Delayed model response chunks previously entered tool execution -without a lock, causing valid title and routine actions to fail at SQL access. +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. diff --git a/docs/browser-async-context.md b/docs/browser-async-context.md index bbab8eb..dcc05cb 100644 --- a/docs/browser-async-context.md +++ b/docs/browser-async-context.md @@ -13,10 +13,10 @@ 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 installs the native `TransformStream` constructor with -actor callback binding. A transformer created inside an actor retains its actor -and async stores when network chunks arrive later. Transform, flush and cancel -callbacks re-enter through the existing input gate; start remains synchronous. +`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. diff --git a/docs/gating-coverage.md b/docs/gating-coverage.md index 5995881..fb316d1 100644 --- a/docs/gating-coverage.md +++ b/docs/gating-coverage.md @@ -32,7 +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 `TransformStream` callbacks | constructor captures the current actor and async stores; `transform`, `flush`, and `cancel` use `makeReentryCallback`; synchronous `start` retains its native timing and receiver | `api/global-scope.ts`; delayed input regression in `global-scope.test.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` | diff --git a/src/api/global-scope.test.ts b/src/api/global-scope.test.ts index 02b6361..969a32d 100644 --- a/src/api/global-scope.test.ts +++ b/src/api/global-scope.test.ts @@ -27,6 +27,34 @@ import { 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(); + const seen: unknown[] = []; + const source: UnderlyingDefaultSource = { + 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 })), + ); + const reader = stream.getReader(); + expect(await store.run("unrelated", () => reader.read())).toEqual({ value: "chunk", done: false }); + await reader.cancel(); + expect(seen).toEqual([ + ["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 }; @@ -498,6 +526,7 @@ describe("installActorScope", () => { installActorScope(target, () => scope); expect(Object.keys(target).sort()).toEqual([ + "ReadableStream", "TransformStream", "WebSocket", "WebSocketPair", diff --git a/src/api/global-scope.ts b/src/api/global-scope.ts index f044b0a..781ab80 100644 --- a/src/api/global-scope.ts +++ b/src/api/global-scope.ts @@ -532,6 +532,7 @@ 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; }; @@ -574,6 +575,7 @@ export function actorScopeBindings(resolve: () => ActorGlobalScope): ActorScopeB }), WebSocketPair: BoundWebSocketPair, WebSocketRequestResponsePair, + ReadableStream: ActorReadableStream, TransformStream: ActorTransformStream, get currentExternalEntry(): object | undefined { return resolve().currentExternalEntry; @@ -616,37 +618,44 @@ function scopeCrypto(resolve: () => ActorGlobalScope): Crypto { /** Captured at import, before any host installs a scope over it. */ const platformCrypto = globalThis.crypto; -// Native stream machinery invokes transformer callbacks without going through -// getReader() or a transformed await. Capture the creating actor here, before -// network input arrives on a later task. Streams created outside actors remain -// native, and the constructor/prototype/brand and callback receiver are retained. -const ActorTransformStream = new Proxy(globalThis.TransformStream, { - construct(target, args, newTarget) { - const context = tryCurrentIoContext(); - const [transformer, ...strategies] = args; - if ( - !context || transformer == null || - (typeof transformer !== "object" && typeof transformer !== "function") - ) { - return Reflect.construct(target, args, newTarget); - } - const callbacks = new Proxy({}, { - get(_target, name) { - const callback = Reflect.get(transformer, name, transformer); - if (typeof callback !== "function") return callback; - // start runs synchronously during construction, already inside the actor. - if (name === "start") return callback.bind(transformer); - if (name === "transform" || name === "flush" || name === "cancel") { - return context.makeReentryCallback((_lock, ...values: unknown[]) => - Reflect.apply(callback, transformer, values), - ); - } - return callback; - }, - }); - return Reflect.construct(target, [callbacks, ...strategies], newTarget); - }, -}); +// 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( + 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 = [ From 7c7722e7281047ee7d037e569d6de3b7ad75b481 Mon Sep 17 00:00:00 2001 From: Alex Nahas Date: Wed, 16 Sep 2026 21:33:10 -0700 Subject: [PATCH 5/5] Make the readable stream regression cross a real task boundary Let the creating input lock expire before pulling so the test proves reentry instead of accidentally sharing the original microtask checkpoint. Also assert that start keeps its synchronous actor scope and receiver. --- src/api/global-scope.test.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/api/global-scope.test.ts b/src/api/global-scope.test.ts index 969a32d..8ed532c 100644 --- a/src/api/global-scope.test.ts +++ b/src/api/global-scope.test.ts @@ -34,6 +34,9 @@ test("readable stream callbacks re-enter their creator when consumed outside its const store = new AsyncLocalStorage(); const seen: unknown[] = []; const source: UnderlyingDefaultSource = { + start() { + seen.push(["start", ctx.hasCurrent(), store.getStore(), this === source]); + }, pull(controller) { seen.push(["pull", ctx.hasCurrent(), store.getStore(), this === source]); controller.enqueue("chunk"); @@ -45,10 +48,13 @@ test("readable stream callbacks re-enter their creator when consumed outside its 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], ]);