feat(codemode): stream fetch response bodies (#63436)

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

packages/codemode/src/extension.ts

@@ -61,6 +61,14 @@ export const make = <const Globals extends Readonly<Record<string, unknown>>>(
 }
 
 const asyncGlobals = new WeakSet<Function>()
+
+/** 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. */

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 = <R>(
       // 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) {

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<Service, ToolContext>()("@opencode/CodeModeExecution") {}
+export class Service extends Context.Service<Service, ToolContext & { signal: AbortSignal }>()(
+  "@opencode/CodeModeExecution",
+) {}

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

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)

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<void>()
+    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<void>()
+    const closed = Promise.withResolvers<void>()
+    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<void>()
+    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<string> = []
   let server: ReturnType<typeof Bun.serve>