feat(core): connect model providers over a Unix socket

packages/ai/src/protocols/utils/responses-compaction.ts

@@ -124,7 +124,7 @@ export const make = (
         headers: parts.headers,
         encoding: request.http?.encoding,
       }),
-      RequestExecutor.withFetch(request.fetch, options?.http),
+      RequestExecutor.withTransport(request.http?.socketPath, request.fetch, options?.http),
     )
     const text = yield* RequestExecutor.responseStream(response).pipe(
       Stream.decodeText(),

packages/ai/src/route/client.ts

@@ -402,7 +402,7 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
           endpoint: routeInput.endpoint,
           auth: routeInput.auth ?? Auth.none,
           encodeBody: ProviderShared.encodeJson,
-          middleware: RequestExecutor.withFetch(request.fetch, options?.http),
+          middleware: RequestExecutor.withTransport(request.http?.socketPath, request.fetch, options?.http),
           webSocket: options?.webSocket,
         }),
       streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime, options?: StreamOptions) => {

packages/ai/src/route/executor.ts

@@ -10,6 +10,7 @@ import {
 import { HttpContext, HttpRateLimitDetails, AIError, TransportError } from "../schema/index.js"
 import { classifyProviderFailure, providerErrorMessage } from "../provider-error.js"
 import { headerDetails, responseHttp, Service, type HttpMiddleware, type Interface } from "./executor-service.js"
+import { unixSocketFetch } from "./unix-socket.js"
 
 export { responseHttp, Service } from "./executor-service.js"
 export type { HttpHandler, HttpMiddleware, Interface, RequestExecutorService } from "./executor-service.js"
@@ -260,16 +261,35 @@ export const fetchLayer = layer.pipe(Layer.provide(FetchHttpClient.layer))
 export const withFetch = (
   fetch: typeof globalThis.fetch | undefined,
   middleware?: HttpMiddleware,
+  requestInit?: RequestInit,
 ): HttpMiddleware | undefined => {
   if (fetch === undefined) return middleware
-  const send = (request: HttpClientRequest.HttpClientRequest) =>
-    HttpClient.execute(request).pipe(
+  const send = (request: HttpClientRequest.HttpClientRequest) => {
+    const executed = HttpClient.execute(request).pipe(
       Effect.provide(FetchHttpClient.layer),
       Effect.provideService(FetchHttpClient.Fetch, fetch),
     )
+    return requestInit === undefined
+      ? executed
+      : executed.pipe(Effect.provideService(FetchHttpClient.RequestInit, requestInit))
+  }
   return middleware === undefined ? send : (request) => middleware(request, send)
 }
 
+/**
+ * The per-call middleware that chooses the transport. A configured Unix socket wins over a per-call `fetch`, so a
+ * provider request cannot silently fall back to TCP. Keeping the decision here also means a new send site only has to
+ * pass its `HttpOptions` through.
+ */
+export const withTransport = (
+  socketPath: string | undefined,
+  fetch: typeof globalThis.fetch | undefined,
+  middleware?: HttpMiddleware,
+): HttpMiddleware | undefined =>
+  socketPath === undefined
+    ? withFetch(fetch, middleware)
+    : withFetch(unixSocketFetch(socketPath), middleware, { timeout: false } as BunFetchRequestInit)
+
 /** Run `fn` on every request: it sees the raw response before status classification, inside middleware already on `executor`, and outside per-call middleware. */
 export const middleware = (fn: HttpMiddleware, executor: Layer.Layer<Service> = fetchLayer): Layer.Layer<Service> =>
   Layer.effect(

packages/ai/src/route/index.ts

@@ -13,6 +13,7 @@ export type {
   TriggerCompactOptions,
 } from "./client.js"
 export * from "./executor.js"
+export { unixSocketFetch } from "./unix-socket.js"
 export { Auth } from "./auth.js"
 export { AuthOptions } from "./auth-options.js"
 export { Endpoint } from "./endpoint.js"

packages/ai/src/route/media.ts

@@ -331,7 +331,7 @@ const makeTransport = <Request extends MediaRequest>(
     })
     const response = yield* execute(
       encoded.apply(HttpClientRequest.make(call.method)(url).pipe(HttpClientRequest.setHeaders(headers))),
-      RequestExecutor.withFetch(fetch),
+      RequestExecutor.withTransport(call.request.http?.socketPath, fetch),
     )
     return { response }
   })
@@ -357,9 +357,11 @@ const makeTransport = <Request extends MediaRequest>(
         protocol.prepare === undefined
           ? request
           : yield* protocol.prepare(request, (path, body) =>
-              send({ method: "POST", url: baseURL(path), headers, request, body }, request.fetch, execute).pipe(
-                Effect.map((sent) => sent.response),
-              ),
+              send(
+                { method: "POST", url: baseURL(path), headers, request: { ...request, http }, body },
+                request.fetch,
+                execute,
+              ).pipe(Effect.map((sent) => sent.response)),
             )
       // Sanitize after merging so model-level overlays are covered; the model value is restored, not sanitized.
       const resolved: Request = {

packages/ai/src/route/unix-socket.ts

@@ -0,0 +1,21 @@
+/**
+ * `fetch` over an unencrypted Unix domain socket. The URL supplies the request path, method and headers, and must use
+ * `http:`; its host and port are ignored and the socket is dialed directly, so no TCP port is opened or reached.
+ * Redirects are not followed: a redirect target would be a normal TCP request.
+ */
+export const unixSocketFetch = (socketPath: string): typeof fetch => {
+  if (socketPath.length === 0)
+    return (() => Promise.reject(new TypeError("Unix socket path must not be empty"))) as unknown as typeof fetch
+  // Node's fetch has no socket option, and a hand-written node:http client would be untested code in a Bun-first CLI.
+  // Reject clearly instead of letting the request fall back to the URL's TCP host.
+  if (typeof Bun === "undefined")
+    return (() => Promise.reject(new TypeError("Unix socket transport requires the Bun runtime"))) as unknown as typeof fetch
+  return ((input, init) => {
+    const protocol = new URL(input instanceof Request ? input.url : String(input)).protocol
+    if (protocol !== "http:") return Promise.reject(unsupportedProtocol(protocol))
+    return fetch(input, { ...init, unix: socketPath, redirect: "manual" })
+  }) as typeof fetch
+}
+
+const unsupportedProtocol = (protocol: string) =>
+  new TypeError(`Unix socket transport requires an http URL, received ${protocol}`)

packages/ai/src/schema/options.ts

@@ -72,6 +72,11 @@ export interface HttpOptions {
   readonly body?: Record<string, unknown>
   /** Compression for JSON request bodies. Only set it for endpoints known to decode it. */
   readonly encoding?: RequestEncoding
+  /**
+   * Unix domain socket to dial instead of the URL's host and port, such as a local model server's socket. The URL still
+   * supplies the request path and must use `http:`; its host and port are ignored.
+   */
+  readonly socketPath?: string
 }
 
 /** Per-phase time limits in milliseconds. */
@@ -127,6 +132,7 @@ export const mergeTransport = (...items: ReadonlyArray<TransportOptions | undefi
       query: mergeStringRecords(...http.map((item) => item?.query)),
       body: mergeJsonRecords(...http.map((item) => item?.body)),
       encoding: last(http.map((item) => item?.encoding)),
+      socketPath: last(http.map((item) => item?.socketPath)),
     }),
   }
 }

packages/ai/test/unix-socket.test.ts

@@ -0,0 +1,207 @@
+import { afterEach, describe, expect, test } from "bun:test"
+import { Effect, Layer } from "effect"
+import { ai, LanguageModel } from "../src/effect.js"
+import { OpenAICompatible } from "../src/providers.js"
+import { LLMClient, RequestExecutor, mergeTransport, unixSocketFetch } from "../src/route.js"
+import { deltaChunk, finishChunk } from "./lib/openai-chunks.js"
+import { sseEvents } from "./lib/sse.js"
+import { it } from "./lib/effect.js"
+import { randomUUID } from "node:crypto"
+import { rmSync } from "node:fs"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+
+const sockets: string[] = []
+const servers: Array<ReturnType<typeof Bun.serve>> = []
+
+const serveUnix = (fetch: (request: Request) => Response | Promise<Response>) => {
+  const path = join(tmpdir(), `oc-unix-${randomUUID()}.sock`)
+  const server = Bun.serve({ unix: path, fetch })
+  servers.push(server)
+  sockets.push(path)
+  return path
+}
+
+const serveTcp = (fetch: (request: Request) => Response | Promise<Response>) => {
+  const server = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch })
+  servers.push(server)
+  return server
+}
+
+afterEach(() => {
+  for (const server of servers.splice(0)) server.stop(true)
+  for (const path of sockets.splice(0)) rmSync(path, { force: true })
+})
+
+const encoder = new TextEncoder()
+
+describe.skipIf(process.platform === "win32")("unixSocketFetch", () => {
+  test("round-trips a JSON POST over the socket and ignores the URL host", async () => {
+    const seen: Array<{ method: string; path: string; token: string | null; body: string }> = []
+    const path = serveUnix(async (request) => {
+      seen.push({
+        method: request.method,
+        path: new URL(request.url).pathname + new URL(request.url).search,
+        token: request.headers.get("x-token"),
+        body: await request.text(),
+      })
+      return Response.json({ ok: true })
+    })
+
+    const response = await unixSocketFetch(path)("http://ignored.invalid/v1/chat?stream=1", {
+      method: "POST",
+      headers: { "x-token": "secret" },
+      body: JSON.stringify({ hello: "world" }),
+    })
+
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({ ok: true })
+    expect(seen).toEqual([
+      { method: "POST", path: "/v1/chat?stream=1", token: "secret", body: '{"hello":"world"}' },
+    ])
+  })
+
+  test("delivers SSE chunks before the response completes", async () => {
+    let release!: () => void
+    const gate = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    let pull = 0
+    const path = serveUnix(
+      () =>
+        new Response(
+          new ReadableStream({
+            async pull(controller) {
+              if (pull++ === 0) {
+                controller.enqueue(encoder.encode("data: first\n\n"))
+                return
+              }
+              await gate
+              controller.enqueue(encoder.encode("data: second\n\ndata: [DONE]\n\n"))
+              controller.close()
+            },
+          }),
+          { headers: { "content-type": "text/event-stream" } },
+        ),
+    )
+
+    const response = await unixSocketFetch(path)("http://localhost/v1/chat")
+    const reader = response.body!.getReader()
+    const first = await reader.read()
+    expect(new TextDecoder().decode(first.value)).toBe("data: first\n\n")
+
+    release()
+    const rest: string[] = []
+    while (true) {
+      const part = await reader.read()
+      if (part.done) break
+      rest.push(new TextDecoder().decode(part.value))
+    }
+
+    expect(rest.join("")).toBe("data: second\n\ndata: [DONE]\n\n")
+  })
+
+  test("aborts while waiting for response headers", async () => {
+    let arrived!: () => void
+    const started = new Promise<void>((resolve) => {
+      arrived = resolve
+    })
+    const path = serveUnix(() => {
+      arrived()
+      return new Promise<Response>(() => {})
+    })
+    const controller = new AbortController()
+
+    const call = unixSocketFetch(path)("http://localhost/v1/chat", { signal: controller.signal })
+    await started
+    controller.abort()
+
+    await expect(call).rejects.toMatchObject({ name: "AbortError" })
+  })
+
+  test("aborts while reading the response body", async () => {
+    const path = serveUnix(
+      () =>
+        new Response(
+          new ReadableStream({
+            start(controller) {
+              controller.enqueue(encoder.encode("data: first\n\n"))
+            },
+          }),
+          { headers: { "content-type": "text/event-stream" } },
+        ),
+    )
+    const controller = new AbortController()
+
+    const response = await unixSocketFetch(path)("http://localhost/v1/chat", { signal: controller.signal })
+    const reader = response.body!.getReader()
+    expect((await reader.read()).done).toBeFalse()
+
+    controller.abort()
+
+    await expect(reader.read()).rejects.toMatchObject({ name: "AbortError" })
+  })
+
+  test("returns redirects instead of following them over TCP", async () => {
+    const tcpRequests: string[] = []
+    const tcp = serveTcp((request) => {
+      tcpRequests.push(request.url)
+      return new Response("tcp")
+    })
+    const path = serveUnix(
+      () => new Response(null, { status: 302, headers: { location: `http://127.0.0.1:${tcp.port}/elsewhere` } }),
+    )
+
+    const response = await unixSocketFetch(path)("http://ignored.invalid/v1/chat")
+
+    expect(response.status).toBe(302)
+    expect(tcpRequests).toEqual([])
+  })
+
+  test("rejects an encrypted URL and an empty socket path", async () => {
+    const path = serveUnix(() => new Response("unused"))
+
+    await expect(unixSocketFetch(path)("https://localhost/v1/chat")).rejects.toThrow("requires an http URL")
+    await expect(unixSocketFetch("")("http://localhost/v1/chat")).rejects.toThrow("must not be empty")
+  })
+
+  test("fails when the socket does not exist", async () => {
+    const path = join(tmpdir(), `oc-missing-${randomUUID()}.sock`)
+
+    await expect(unixSocketFetch(path)("http://localhost/v1/chat")).rejects.toThrow()
+  })
+})
+
+describe.skipIf(process.platform === "win32")("native transport with a socket", () => {
+  it.effect("dials the socket instead of the base URL's TCP host", () =>
+    Effect.gen(function* () {
+      const socketRequests: string[] = []
+      const socket = serveUnix((request) => {
+        socketRequests.push(new URL(request.url).pathname)
+        return new Response(
+          sseEvents(deltaChunk({ role: "assistant", content: "over socket" }), finishChunk("stop")),
+          { headers: { "content-type": "text/event-stream" } },
+        )
+      })
+      const tcpRequests: string[] = []
+      const tcp = serveTcp((request) => {
+        tcpRequests.push(request.url)
+        return new Response("not the socket")
+      })
+
+      const baseURL = `http://127.0.0.1:${tcp.port}/v1`
+      const runtime = OpenAICompatible.model("local-model", { apiKey: "test", baseURL })
+      const model = LanguageModel.update(runtime, {
+        options: mergeTransport(runtime.options, { http: { socketPath: socket } }),
+      })
+
+      const response = yield* LLMClient.generate(ai.request({ model, prompt: "Hello" })).pipe(
+        Effect.provide(LLMClient.layer.pipe(Layer.provide(RequestExecutor.fetchLayer))),
+      )
+
+      expect(response.text).toBe("over socket")
+      expect(socketRequests).toEqual(["/v1/chat/completions"])
+      expect(tcpRequests).toEqual([])
+    }),
+  )
+})

packages/core/src/aisdk.ts

@@ -32,7 +32,15 @@ import {
   type ToolMessage,
   type UsageInput,
 } from "@opencode/ai/effect"
-import { Auth, Endpoint, ProviderShared, RequestExecutor, type AnyRoute, type HttpMiddleware } from "@opencode/ai/route"
+import {
+  Auth,
+  Endpoint,
+  ProviderShared,
+  RequestExecutor,
+  unixSocketFetch,
+  type AnyRoute,
+  type HttpMiddleware,
+} from "@opencode/ai/route"
 import { isRecord } from "@opencode/util/record"
 import { Cause, Context, Effect, Layer, Option, Schema, Scope, Stream, Struct } from "effect"
 import { makeParser } from "effect/encoding/Sse"
@@ -133,6 +141,7 @@ function prepareOptions(model: RuntimeInfo, pkg: string, outbound: Fetch) {
   }
 
   const customFetch = settings.fetch
+  const socketPath = Provider.socketPath(settings)
   const timeouts = Provider.timeouts(settings)
   options.fetch = async (input: Parameters<typeof fetch>[0], init?: RequestInit) => {
     const opts = { ...(init ?? {}) }
@@ -160,7 +169,9 @@ function prepareOptions(model: RuntimeInfo, pkg: string, outbound: Fetch) {
       }
     }
 
-    const send: Fetch = isFetch(customFetch) ? customFetch : outbound
+    // A configured socket wins over a custom fetch so a provider never silently falls back to TCP.
+    const send: Fetch =
+      socketPath === undefined ? (isFetch(customFetch) ? customFetch : outbound) : (unixSocketFetch(socketPath) as Fetch)
     const middleware = httpMiddleware.getStore()
     const res = await (
       middleware