From a8dde001d3b89330cfd67f0ed226efba00627d9d Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Thu, 12 Nov 2026 17:50:06 -0600 Subject: [PATCH] feat(codemode): stream fetch response bodies (#63436) --- packages/codemode/interpreter-support.md | 1 + packages/codemode/src/extension.ts | 8 ++ packages/codemode/src/runtime/extensions.ts | 55 +++++++++- packages/core/src/codemode/execution.ts | 4 +- packages/core/src/codemode/fetch.ts | 74 +++++-------- packages/core/src/codemode/tool.ts | 3 +- packages/core/test/tool-execute.test.ts | 114 ++++++++++++++++++++ 7 files changed, 205 insertions(+), 54 deletions(-) diff --git a/packages/codemode/interpreter-support.md b/packages/codemode/interpreter-support.md index 099d615ff52..a2298fd5109 100644 --- a/packages/codemode/interpreter-support.md +++ b/packages/codemode/interpreter-support.md @@ -604,6 +604,7 @@ with a hint to encode as text first (`TextDecoder`, `toBase64`, `toHex`). `[Symbol.iterator]`; iteration is live and sorted by name, with `set-cookie` values kept apart. - [x] Headers serialize to a `{ name: value }` object in JSON, in results, and in tool arguments. - [ ] `Request`, `Response`, and `Blob`. +- [x] A host-provided streaming fetch response exposes `bodyUsed`, whole-body readers, a default stream reader and `for await` iteration. `arrayBuffer()` crosses as a `Uint8Array`. - [ ] `crypto.subtle` and non-UTF-8 `TextDecoder` encodings. ## Extensions diff --git a/packages/codemode/src/extension.ts b/packages/codemode/src/extension.ts index 8ca6b9b8332..eb4342b8dc0 100644 --- a/packages/codemode/src/extension.ts +++ b/packages/codemode/src/extension.ts @@ -61,6 +61,14 @@ export const make = >>( } const asyncGlobals = new WeakSet() + +/** A live fetch response whose I/O is owned by the host for the lifetime of the program. */ +export class StreamingResponse { + constructor( + readonly response: Response, + readonly redirected = response.redirected, + ) {} +} const AsyncFunction = (async () => {}).constructor /** Marks a global that returns a Promise or an Effect, so a denied call rejects its promise instead of throwing. */ diff --git a/packages/codemode/src/runtime/extensions.ts b/packages/codemode/src/runtime/extensions.ts index dc6f776f203..1a4e8500dc4 100644 --- a/packages/codemode/src/runtime/extensions.ts +++ b/packages/codemode/src/runtime/extensions.ts @@ -1,14 +1,15 @@ import { Effect } from "effect" -import { type Extension, isAsync } from "../extension.js" +import { type Extension, isAsync, StreamingResponse } from "../extension.js" import { hooked } from "../tool-runtime.js" import { errorMessage, errorType } from "./errors.js" import type { Interpreter } from "./interpreter.js" import { createErrorValue } from "./intrinsics.js" import { MAX_ARRAY_LENGTH, MAX_VALUE_DEPTH } from "./limits.js" -import { type AstNode, PendingThrow, rangeError, Throw, typeError } from "./model.js" +import { type AstNode, AsyncIteratorSymbol, PendingThrow, rangeError, Throw, typeError } from "./model.js" import { native } from "./native.js" import { define, + defineAccessor, get, has, hidden, @@ -112,6 +113,56 @@ export const extensionGlobals = ( // Nothing a host value hands back is kept: results are read as primitives or copied. Where the value's own // method could hand back host storage or an object (`getTime`, `slice`, `map`), its internal data is read. if (value instanceof Date) return new DateObj(builtins.Date, Date.prototype.getTime.call(value)) + if (value instanceof StreamingResponse) { + const response = value.response + const body = response.body + const stream = new Obj(builtins.Object) + if (body) { + defineAccessor(stream, "locked", () => body.locked) + define( + stream, + "cancel", + next(async (reason?: unknown) => body.cancel(reason), `${label}.body.cancel`), + ) + define( + stream, + "getReader", + next((options?: { mode?: string }) => { + if (options?.mode !== undefined) throw new TypeError("Only the default stream reader is supported.") + const reader = body.getReader() + return { + read: async () => reader.read(), + cancel: async (reason?: unknown) => reader.cancel(reason), + releaseLock: () => reader.releaseLock(), + } + }, `${label}.body.getReader`), + ) + define( + stream, + AsyncIteratorSymbol, + next(() => { + const iterator = body.values() + return { + next: async () => iterator.next(), + return: async () => iterator.return!(), + } + }, `${label}.body[Symbol.asyncIterator]`), + ) + } + const result = new Obj(builtins.Object) + for (const key of ["url", "status", "statusText", "ok", "headers"] as const) + define(result, key, next(response[key], `${label}.${key}`)) + define(result, "redirected", value.redirected) + define(result, "body", body ? stream : null) + defineAccessor(result, "bodyUsed", () => response.bodyUsed) + for (const key of ["text", "json", "bytes", "arrayBuffer"] as const) + define( + result, + key, + next(async () => response[key](), `${label}.${key}`), + ) + return result + } if (value instanceof RegExp) return new RegExpObj(builtins.RegExp, value.source, value.flags) if (value instanceof Uint8Array) return new Bytes(builtins.Uint8Array, new Uint8Array(value)) if (value instanceof ArrayBuffer) { diff --git a/packages/core/src/codemode/execution.ts b/packages/core/src/codemode/execution.ts index b41e0de4e85..e9c0c4cf0d2 100644 --- a/packages/core/src/codemode/execution.ts +++ b/packages/core/src/codemode/execution.ts @@ -4,4 +4,6 @@ import type { Context as ToolContext } from "@opencode/schema/tool" import { Context } from "effect" /** The `execute` call a Code Mode program runs under, so extensions can act for its session like a tool does. */ -export class Service extends Context.Service()("@opencode/CodeModeExecution") {} +export class Service extends Context.Service()( + "@opencode/CodeModeExecution", +) {} diff --git a/packages/core/src/codemode/fetch.ts b/packages/core/src/codemode/fetch.ts index 2f4ca6b32fc..871500199ef 100644 --- a/packages/core/src/codemode/fetch.ts +++ b/packages/core/src/codemode/fetch.ts @@ -20,7 +20,8 @@ const fetch = (input: string | URL, init: Init = {}) => // matching. const url = String(input).split("#")[0] const parsed = URL.parse(url) - if (parsed?.protocol === "file:") return yield* Effect.fail(new Error("fetch cannot be used for filesystem access.")) + if (parsed?.protocol === "file:") + return yield* Effect.fail(new Error("fetch cannot be used for filesystem access.")) // Rules match the URL, so `https://user@evil.com` would slip past a rule for `https://evil.com/*`. if (parsed?.username || parsed?.password) return yield* Effect.fail(new TypeError("fetch cannot be used with credentials in the URL.")) @@ -54,35 +55,30 @@ const fetch = (input: string | URL, init: Init = {}) => agent: execution.agent, source, }) - const signal = AbortSignal.timeout(TIMEOUT_MS) - const follow = ( - target: string, - request: Init, - redirects: number, - ): Effect.Effect<{ response: Response; url: string; redirects: number }, unknown> => - Effect.tryPromise({ + const signal = AbortSignal.any([execution.signal, AbortSignal.timeout(TIMEOUT_MS)]) + let target = url + let request = init + for (let redirects = 0; ; redirects++) { + const response = yield* Effect.tryPromise({ try: () => globalThis.fetch(target, { ...request, redirect: "manual", signal }), catch: (error) => error, - }).pipe( - Effect.flatMap((response) => { - const location = response.headers.get("location") - if (!redirectStatuses.has(response.status) || location === null) - return Effect.succeed({ response, url: target, redirects }) - if (redirects >= MAX_REDIRECTS) - return Effect.fail(new Error(`Too many redirects (more than ${MAX_REDIRECTS})`)) - const next = new URL(location, target) - if (next.protocol !== "http:" && next.protocol !== "https:") - return Effect.fail(new Error(`fetch cannot follow a redirect to ${next.protocol} URLs.`)) - if (next.username || next.password) - return Effect.fail(new Error("fetch cannot follow a redirect with credentials in the URL.")) - next.hash = "" - return check(next.href).pipe( - Effect.andThen(follow(next.href, redirected(request, response.status, target, next), redirects + 1)), - ) - }), - ) - const result = yield* follow(url, init, 0) - return yield* Effect.tryPromise({ try: () => read(result), catch: (error) => error }) + }) + const location = response.headers.get("location") + if (!redirectStatuses.has(response.status) || location === null) + return new Extension.StreamingResponse(response, redirects > 0) + yield* Effect.tryPromise({ try: async () => response.body?.cancel(), catch: (error) => error }) + if (redirects >= MAX_REDIRECTS) + return yield* Effect.fail(new Error(`Too many redirects (more than ${MAX_REDIRECTS})`)) + const next = new URL(location, target) + if (next.protocol !== "http:" && next.protocol !== "https:") + return yield* Effect.fail(new Error(`fetch cannot follow a redirect to ${next.protocol} URLs.`)) + if (next.username || next.password) + return yield* Effect.fail(new Error("fetch cannot follow a redirect with credentials in the URL.")) + next.hash = "" + yield* check(next.href) + request = redirected(request, response.status, target, next) + target = next.href + } }) export const MAX_REDIRECTS = 20 @@ -107,28 +103,6 @@ const redirected = (request: Init, status: number, from: string, to: URL): Init } } -const read = async (result: { response: Response; url: string; redirects: number }) => { - const response = result.response - const bytes = await response.bytes() - const headers = Object.fromEntries(response.headers) - const text = () => new TextDecoder().decode(bytes) - return { - url: result.url, - status: response.status, - statusText: response.statusText, - ok: response.ok, - redirected: result.redirects > 0, - headers: { - get: (name: string) => headers[name.toLowerCase()] ?? null, - has: (name: string) => name.toLowerCase() in headers, - entries: () => Object.entries(headers), - }, - text: async () => text(), - json: async () => JSON.parse(text()) as unknown, - bytes: async () => bytes, - } -} - export const extension = Extension.make({ name: "fetch", globals: { fetch: Extension.async(fetch) } }) /** What to show for a fetch call: its method and URL. */ diff --git a/packages/core/src/codemode/tool.ts b/packages/core/src/codemode/tool.ts index 93451c6b850..2d24511228b 100644 --- a/packages/core/src/codemode/tool.ts +++ b/packages/core/src/codemode/tool.ts @@ -173,7 +173,8 @@ export const create = ( Effect.onError(() => lock.withPermit(Ref.get(trailing).pipe(Effect.flatMap((pending) => (pending ? publish : Effect.void)))), ), - Effect.provideService(CodeModeExecution.Service, context), + // Readers outlive the fetch call that returned them, up to the end of this program's scope. + Effect.provideService(CodeModeExecution.Service, { ...context, signal: yield* Effect.abortSignal }), Effect.provideService(Permission.Service, permission), ) const toolCalls = yield* Ref.get(calls) diff --git a/packages/core/test/tool-execute.test.ts b/packages/core/test/tool-execute.test.ts index b0a3a989652..4aa0ab68b4a 100644 --- a/packages/core/test/tool-execute.test.ts +++ b/packages/core/test/tool-execute.test.ts @@ -152,6 +152,120 @@ test("execute rejects filesystem fetches", async () => { expect(result.output.output).toContain("fetch cannot be used for filesystem access.") }) +describe("streaming fetch", () => { + const run = (url: URL, code: string, permission = allowAll) => + createCodeMode(new Map(), permission).execute( + { code: `const res = await fetch(${JSON.stringify(url.href)}); ${code}` }, + context, + ) + + // Stays open after the first chunk so buffering would block until the test times out. + const openBody = (cancel: () => void) => + new ReadableStream({ + start(controller) { + controller.enqueue(Uint8Array.of(42)) + }, + cancel, + }) + + test.each([ + ["return with a live reader", "const reader = res.body.getReader(); return (await reader.read()).value[0]"], + [ + "reader cancellation", + "const reader = res.body.getReader(); const chunk = await reader.read(); await reader.cancel(); return chunk.value[0]", + ], + [ + "break from async iteration", + "let first; for await (const chunk of res.body) { first = chunk[0]; break } return res.body.locked ? 'locked' : first", + ], + ])("reads the first chunk before EOF: %s", async (_, code) => { + const closed = Promise.withResolvers() + await using server = Bun.serve({ + port: 0, + fetch: () => new Response(openBody(closed.resolve)), + }) + const result = await Effect.runPromise(run(server.url, code)) + expect(result.output.output).toBe("42") + await closed.promise + }) + + test("interrupting execute aborts a pending request", async () => { + const started = Promise.withResolvers() + const closed = Promise.withResolvers() + await using server = Bun.serve({ + port: 0, + fetch: async (req) => { + req.signal.addEventListener("abort", () => closed.resolve(), { once: true }) + started.resolve() + await closed.promise + return new Response(null) + }, + }) + const fiber = Effect.runFork(run(server.url, "return res.status")) + await started.promise + await Effect.runPromise(Fiber.interrupt(fiber)) + await closed.promise + }) + + test("whole-body methods share consumption and expose Bun's empty body", async () => { + await using server = Bun.serve({ + port: 0, + fetch: (req) => (req.method === "HEAD" ? new Response(null) : Response.json(42)), + }) + const result = await Effect.runPromise( + run( + server.url, + ` + const before = res.bodyUsed + const number = await res.json() + const second = await res.text().catch(e => e.name) + const bytes = await (await fetch(res.url)).bytes() + const buffer = await (await fetch(res.url)).arrayBuffer() + const empty = await fetch(res.url, { method: "HEAD" }) + return [res.headers.get("Content-Type"), before, number, res.bodyUsed, second, [...bytes], [...buffer], empty.body === null] + `, + ), + ) + expect(JSON.parse(result.output.output)).toEqual([ + "application/json;charset=utf-8", + false, + 42, + true, + "TypeError", + [52, 50], + [52, 50], + true, + ]) + }) + + test("discards a redirect body before reading the destination", async () => { + const discarded = Promise.withResolvers() + await using server = Bun.serve({ + port: 0, + async fetch(req) { + if (new URL(req.url).pathname === "/") + return new Response(openBody(discarded.resolve), { status: 302, headers: { location: "/next" } }) + await discarded.promise + return new Response("ok") + }, + }) + const result = await Effect.runPromise( + run( + server.url, + ` + const reader = res.body.getReader() + const first = await reader.read() + const end = await reader.read() + reader.releaseLock() + return [res.redirected, [...first.value], end.done, res.body.locked, await res.text().catch(e => e.name)] + `, + permissionService({ assert: () => Effect.void, assertNotDenied: () => Effect.void }), + ), + ) + expect(JSON.parse(result.output.output)).toEqual([true, [111, 107], true, false, "TypeError"]) + }) +}) + describe("fetch permissions", () => { const hits: Array = [] let server: ReturnType