diff --git a/.changeset/stream-callback-actor-gates.md b/.changeset/stream-callback-actor-gates.md new file mode 100644 index 0000000..4f917f6 --- /dev/null +++ b/.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. 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 diff --git a/docs/browser-async-context.md b/docs/browser-async-context.md index 1cca27f..dcc05cb 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 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 diff --git a/docs/gating-coverage.md b/docs/gating-coverage.md index 0bf3fa1..fb316d1 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 `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/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( diff --git a/src/api/global-scope.test.ts b/src/api/global-scope.test.ts index 428ba73..8ed532c 100644 --- a/src/api/global-scope.test.ts +++ b/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(); + 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"); + }, + 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(); + 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 +532,8 @@ describe("installActorScope", () => { installActorScope(target, () => scope); expect(Object.keys(target).sort()).toEqual([ + "ReadableStream", + "TransformStream", "WebSocket", "WebSocketPair", "WebSocketRequestResponsePair", diff --git a/src/api/global-scope.ts b/src/api/global-scope.ts index 67cd74c..781ab80 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,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( + 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", 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,