feat(desktop): use WebSocket RPC for server requests

packages/app/src/runtime/platform/platform.tsx

@@ -6,6 +6,7 @@ import { ServerConnection } from "@/runtime/server/registry"
 import type { WslServersPlatform } from "@/servers/wsl/types"
 import type { UpdaterPlatform } from "@/shell/updates/types"
 import type { DraftStore } from "@/runtime/persistence/drafts"
+import type { ServerApiConnection } from "@/runtime/server/api"
 
 type PickerPaths = string | string[] | null
 type OpenDirectoryPickerOptions = { title?: string; multiple?: boolean }
@@ -74,6 +75,9 @@ type PlatformBase = {
   /** Fetch override */
   fetch?: typeof fetch
 
+  /** Optional owned server transport; browser web defaults to HTTP. */
+  createServerApi?(server: ServerConnection.HttpBase): ServerApiConnection
+
   /** Get the configured default server URL (platform-specific) */
   getDefaultServer?(): Promise<ServerConnection.Key | null>
 

packages/app/src/runtime/server/api.ts

@@ -36,3 +36,8 @@ export function createApiForServer(input: {
 }
 
 export type ServerApi = OpenCodeClient
+
+export type ServerApiConnection = {
+  readonly api: ServerApi
+  readonly dispose: () => Promise<void>
+}

packages/app/src/runtime/server/client.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, test } from "bun:test"
-import type { OpenCodeEvent } from "@opencode-ai/client/promise"
+import { OpenCode, type OpenCodeEvent } from "@opencode-ai/client/promise"
 import { createRoot } from "solid-js"
 import { createOpenCodeEventSource, createServerTransport } from "./client"
 
@@ -99,7 +99,7 @@ test("rotates HTTP and PTY clients together", async () => {
   const initialPty = transport.pty
 
   await transport.api.health.get()
-  const replacement = transport.update({
+  const replacement = await transport.update({
     url: "http://127.0.0.1:4200",
     username: "opencode",
     password: "second",
@@ -120,3 +120,68 @@ test("rotates HTTP and PTY clients together", async () => {
     },
   ])
 })
+
+test("uses the platform transport and disposes it before rotating credentials", async () => {
+  const lifecycle: string[] = []
+  const transport = createServerTransport({
+    http: { url: "http://127.0.0.1:4100", password: "first" },
+    fetch: (async (_input: string | URL | Request, _init?: RequestInit): Promise<Response> => {
+      throw new Error("The injected transport must not fall back to HTTP")
+    }) as typeof fetch,
+    createApi: (http) => {
+      lifecycle.push(`create:${http.password}`)
+      return {
+        api: OpenCode.make({ baseUrl: http.url }),
+        dispose: async () => {
+          lifecycle.push(`dispose:${http.password}`)
+        },
+      }
+    },
+  })
+  const first = transport.api
+  const pty = transport.pty
+  await transport.update({ url: "http://127.0.0.1:4200", password: "second" })
+  expect(transport.api).not.toBe(first)
+  expect(transport.pty).not.toBe(pty)
+  expect(transport.http.password).toBe("second")
+  await transport.dispose()
+  await transport.dispose()
+  expect(lifecycle).toEqual(["create:first", "dispose:first", "create:second", "dispose:second"])
+})
+
+test("cannot reopen a disposed owner while reconnection is resolving", async () => {
+  const released = Promise.withResolvers<void>()
+  const clients: string[] = []
+  const transport = createServerTransport({
+    http: { url: "http://127.0.0.1:4100" },
+    createApi: (http) => {
+      clients.push(http.url)
+      return { api: OpenCode.make({ baseUrl: http.url }), dispose: () => released.promise }
+    },
+  })
+  const update = transport.update({ url: "http://127.0.0.1:4200" })
+  const result = update.catch((error) => error)
+  const disposal = transport.dispose()
+  released.resolve()
+  await disposal
+  expect(await result).toMatchObject({ name: "AbortError" })
+  expect(clients).toEqual(["http://127.0.0.1:4100"])
+})
+
+test("an aborted reconnect does not create another client", async () => {
+  const abort = new AbortController()
+  const clients: string[] = []
+  const transport = createServerTransport({
+    http: { url: "http://127.0.0.1:4100" },
+    createApi: (http) => {
+      clients.push(http.url)
+      return { api: OpenCode.make({ baseUrl: http.url }), dispose: () => Promise.resolve() }
+    },
+  })
+  abort.abort()
+  await expect(transport.update({ url: "http://127.0.0.1:4200" }, abort.signal)).rejects.toMatchObject({
+    name: "AbortError",
+  })
+  expect(clients).toEqual(["http://127.0.0.1:4100"])
+  await transport.dispose()
+})

packages/app/src/runtime/server/client.tsx

@@ -2,7 +2,7 @@ import type { OpenCodeEvent } from "@opencode-ai/client/promise"
 import { createClientConnection, createPtyClient, type ClientConnectionStatus } from "@opencode-ai/client/solid"
 import { createGlobalEmitter } from "@solid-primitives/event-bus"
 import { type Accessor, onCleanup } from "solid-js"
-import { createApiForServer, type ServerApi } from "@/runtime/server/api"
+import { createApiForServer, type ServerApi, type ServerApiConnection } from "@/runtime/server/api"
 import { usePlatform } from "@/runtime/platform/platform"
 import { ServerConnection } from "./registry"
 import { createRefCountMap } from "@/runtime/server/refcount"
@@ -72,12 +72,20 @@ type ServerSDKBase = {
 
 function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerScope): ServerSDKBase {
   const platform = usePlatform()
-  const transport = createServerTransport({ http: server.http, fetch: platform.fetch })
+  const transport = createServerTransport({
+    http: server.http,
+    fetch: platform.fetch,
+    createApi: platform.createServerApi,
+  })
+  onCleanup(() => void transport.dispose())
   const events = createOpenCodeEventSource()
   const reconnect = server.type === "sidecar" && server.variant === "base" ? server.reconnect : undefined
 
   const connection = createClientConnection(transport.api, {
-    reconnect: reconnect ? async (signal) => transport.update(await reconnect(signal)) : undefined,
+    reconnect:
+      reconnect || platform.createServerApi
+        ? async (signal) => transport.update(reconnect ? await reconnect(signal) : transport.http, signal)
+        : undefined,
     flushInterval: 16,
     pageLifecycle: true,
     onEvent(event) {
@@ -108,22 +116,44 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
   }
 }
 
-export function createServerTransport(input: { http: ServerConnection.HttpBase; fetch?: typeof globalThis.fetch }): {
-  update(http: ServerConnection.HttpBase): ServerApi
+export function createServerTransport(input: {
+  http: ServerConnection.HttpBase
+  fetch?: typeof globalThis.fetch
+  createApi?: (http: ServerConnection.HttpBase) => ServerApiConnection
+}): {
+  update(http: ServerConnection.HttpBase, signal?: AbortSignal): Promise<ServerApi>
+  dispose(): Promise<void>
+  readonly http: ServerConnection.HttpBase
   readonly url: string
   readonly api: ServerApi
   readonly pty: ReturnType<typeof createPtyClient>
 } {
   const build = (http: ServerConnection.HttpBase) => {
-    const api = createApiForServer({ server: http, fetch: input.fetch })
-    return { http, api, pty: createPtyClient(api, { url: http.url }) }
+    const connection = input.createApi?.(http) ?? {
+      api: createApiForServer({ server: http, fetch: input.fetch }),
+      dispose: () => Promise.resolve(),
+    }
+    return { http, ...connection, pty: createPtyClient(connection.api, { url: http.url }) }
   }
-  const state = { current: build(input.http) }
+  const state = { current: build(input.http), disposed: false }
   return {
-    update(http: ServerConnection.HttpBase) {
+    async update(http: ServerConnection.HttpBase, signal?: AbortSignal) {
+      signal?.throwIfAborted()
+      if (state.disposed) throw new DOMException(undefined, "AbortError")
+      await state.current.dispose()
+      signal?.throwIfAborted()
+      if (state.disposed) throw new DOMException(undefined, "AbortError")
       state.current = build(http)
       return state.current.api
     },
+    dispose() {
+      if (state.disposed) return Promise.resolve()
+      state.disposed = true
+      return state.current.dispose()
+    },
+    get http() {
+      return state.current.http
+    },
     get url() {
       return state.current.http.url
     },

packages/client/README.md

@@ -5,7 +5,9 @@ Private generation target for clients derived directly from OpenCode's authorita
 ## Entrypoints
 
 - `@opencode-ai/client`: zero-Effect Promise client using `fetch`.
+- `@opencode-ai/client/promise/rpc`: the same Promise DTO surface over a lazy, shared WebSocket (imports Effect).
 - `@opencode-ai/client/effect`: rich Effect network client using an environment-provided `HttpClient`.
+- `@opencode-ai/client/effect/rpc`: native Effect RPC over a single scoped WebSocket.
 
 The generated surface includes every standard HTTP group from Server's concrete API. The build compiler reads `@opencode-ai/server/api`; the generated Effect runtime imports a client-local projection built from Protocol, with a generation-equivalence test preventing transport drift. Custom transports such as the PTY WebSocket connection remain outside the generic HTTP client. Run `bun run generate` after changing the contract and `bun run check:generated` to detect committed-output drift.
 
@@ -25,3 +27,58 @@ yield *
   })
 yield * client.sessions.prompt({ sessionID, prompt: Prompt.make({ text: "Hello" }) })
 ```
+
+## WebSocket RPC
+
+Promise consumers can keep their existing method calls and wire DTOs:
+
+```ts
+import { OpenCodeRpc } from "@opencode-ai/client/promise/rpc"
+
+const client = OpenCodeRpc.make({
+  baseUrl: "https://opencode.example",
+  headers: { authorization: `Basic ${token}` },
+})
+try {
+  const session = await client.session.create()
+  // Numeric timestamps, not Effect DateTime values.
+  console.log(session.time.created)
+} finally {
+  await client.dispose()
+}
+```
+
+`make` returns synchronously without opening a socket. The first call or stream iteration opens one shared connection. Basic authorization is used only for the upgrade's `auth_token`; other default and per-call headers travel as RPC frame metadata. Per-call headers override endpoint headers, which override facade defaults. Binary `file.read` results remain `Uint8Array`, and declared errors remain plain wire objects accepted by the Promise error guards.
+
+Breaking or returning from an async iterator cancels its subscription. `RequestOptions.signal` cancels one call or stream without closing unrelated work. `dispose()` is idempotent, closes the connection, and rejects subsequent calls. A failed connection is never retried and in-flight requests are never replayed: dispose the failed facade and create a new one to reconnect. The Promise root stays zero-Effect; only the explicit `/promise/rpc` entrypoint imports the native RPC runtime.
+
+The additive `/api/rpc` transport derives its operation contracts from Protocol's HTTP schemas; existing HTTP clients are unchanged. Use operation identifiers directly, with decoded `params`, `query`, and `payload` fields where declared. Numeric queries are numbers, not HTTP strings. Optional `location: { directory, workspace? }` selects per-call location context; session-specific operations retain their existing session location rules.
+
+```ts
+import { OpenCodeRpc } from "@opencode-ai/client/effect/rpc"
+import { Effect, Redacted, Stream } from "effect"
+
+declare const token: string // Base64 of "opencode:<server password>", not the raw password.
+
+const program = Effect.gen(function* () {
+  const client = yield* OpenCodeRpc.make({
+    url: "wss://opencode.example/api/rpc",
+    authToken: Redacted.make(token),
+  })
+  const events = yield* client["event.subscribe"]({}).pipe(
+    Stream.runForEach((event) => Effect.log(event.type)),
+    Effect.forkScoped,
+  )
+  const sessions = yield* client["session.list"]({ query: {} })
+  // Both requests share the connection; interrupting events cancels only that subscription.
+  return sessions
+})
+
+Effect.runPromise(Effect.scoped(program))
+```
+
+Keep the scope open while consuming streams. Scope closure closes the socket and cancels outstanding work. Connection failures are surfaced rather than replaying potentially mutating requests; create a new scoped client to reconnect. Browsers use their native WebSocket constructor, with an optional `webSocketConstructor` override for other runtimes. Authentication uses the server's existing `auth_token` upgrade mechanism; treat the resulting URL as sensitive and do not log it.
+
+Streaming operations return native Effect Streams of the original typed items, not SSE text. No-content operations return `void`. `fs.read` takes `params: { path: "relative/file" }` plus `query: {}` and returns `{ content: Uint8Array, mime: string }`, with bytes encoded as base64 on the wire. Raw `pty.connect` and `persistentPty.connect` remain on their existing WebSocket routes.
+
+From `packages/server`, run `bun run script/benchmark-rpc.ts` for an isolated loopback comparison of HTTP and RPC session-list calls. It uses an in-memory database, temporary configuration, and schema-decoding clients at concurrency 1 and 16. This measures transport overhead, not end-to-end desktop/web speed. Desktop injects the Promise RPC transport; browser web retains HTTP. Desktop service discovery/readiness checks and raw terminal attachments keep their existing transports.

packages/client/package.json

@@ -19,10 +19,12 @@
     ".": "./src/promise/index.ts",
     "./promise": "./src/promise/index.ts",
     "./promise/api": "./src/promise/api.ts",
+    "./promise/rpc": "./src/promise/rpc.ts",
     "./service": "./src/promise/service.ts",
     "./solid": "./src/solid/index.ts",
     "./effect": "./src/effect/index.ts",
     "./effect/api": "./src/effect/api.ts",
+    "./effect/rpc": "./src/effect/rpc.ts",
     "./effect/service": "./src/effect/service.ts"
   },
   "scripts": {

packages/client/src/effect/rpc.ts

@@ -0,0 +1,35 @@
+export * as OpenCodeRpc from "./rpc.js"
+
+import { Group, type Rpcs } from "@opencode-ai/protocol/rpc"
+import { Effect, Redacted, Schedule, Scope } from "effect"
+import { RpcClient, RpcClientError, RpcSerialization } from "effect/unstable/rpc"
+import { Socket } from "effect/unstable/socket"
+
+export interface Options {
+  /** Full WebSocket endpoint, for example wss://opencode.example/api/rpc. */
+  readonly url: string | URL
+  /** Existing server auth_token, sent only during the WebSocket upgrade. */
+  readonly authToken?: Redacted.Redacted<string>
+  readonly webSocketConstructor?: Socket.WebSocketConstructor["Service"]
+}
+
+export type Client = RpcClient.RpcClient<Rpcs, RpcClientError.RpcClientError>
+
+/** One scoped connection multiplexes all unary calls and streaming subscriptions. */
+export const make: (options: Options) => Effect.Effect<Client, never, Scope.Scope> = Effect.fnUntraced(
+  function* (options) {
+    const url = new URL(options.url)
+    if (options.authToken) url.searchParams.set("auth_token", Redacted.value(options.authToken))
+    const socket = yield* Socket.makeWebSocket(url.toString()).pipe(
+      Effect.provideService(
+        Socket.WebSocketConstructor,
+        options.webSocketConstructor ?? ((url, protocols) => new WebSocket(url, protocols)),
+      ),
+    )
+    const protocol = yield* RpcClient.makeProtocolSocket({ retryPolicy: Schedule.recurs(0) }).pipe(
+      Effect.provideService(Socket.Socket, socket),
+      Effect.provideService(RpcSerialization.RpcSerialization, RpcSerialization.json),
+    )
+    return yield* RpcClient.make(Group).pipe(Effect.provideService(RpcClient.Protocol, protocol))
+  },
+)

packages/client/src/promise/rpc.ts

@@ -0,0 +1,194 @@
+export * as OpenCodeRpc from "./rpc.js"
+
+import { ClientApi } from "@opencode-ai/protocol/client"
+import { Group } from "@opencode-ai/protocol/rpc"
+import { Effect, Exit, Redacted, Schema, Scope, Stream } from "effect"
+import type { HttpApi } from "effect/unstable/httpapi"
+import { RpcClientError, RpcSchema } from "effect/unstable/rpc"
+import { OpenCodeRpc } from "../effect/rpc.js"
+import { ClientError } from "./generated/client-error.js"
+import { OpenCode } from "./generated/index.js"
+import type { ClientOptions, RequestDescriptor, RequestOptions } from "./generated/client.js"
+import type { OpenCodeClient } from "./index.js"
+
+export interface Options extends Pick<ClientOptions, "baseUrl" | "headers"> {
+  readonly webSocketConstructor?: OpenCodeRpc.Options["webSocketConstructor"]
+}
+
+export type Client = OpenCodeClient & { readonly dispose: () => Promise<void> }
+
+const endpoints = new Map(
+  Object.values((ClientApi as unknown as HttpApi.Top).groups).flatMap((group) =>
+    Object.values(group.endpoints).map((endpoint) => [endpoint.identifier, endpoint] as const),
+  ),
+)
+
+const codecs = new Map<
+  string,
+  {
+    readonly input: Schema.Codec<unknown, unknown>
+    readonly output: Schema.Codec<unknown, unknown>
+    readonly error: Schema.Codec<unknown, unknown>
+  }
+>()
+
+function operation(name: string) {
+  const cached = codecs.get(name)
+  if (cached) return cached
+  const endpoint = endpoints.get(name)
+  const rpc = Group.requests.get(name)
+  if (!endpoint || !rpc) throw new ClientError("Transport", { cause: new Error(`Unknown RPC operation: ${name}`) })
+  const payloads = Array.from(endpoint.payload.values()).flatMap((entry) => entry.schemas)
+  const output = RpcSchema.isStreamSchema(rpc.successSchema) ? rpc.successSchema.success : rpc.successSchema
+  // Promise query inputs are decoded values; payloads and path/header inputs are HTTP wire values.
+  const input = Schema.Struct({
+    ...(endpoint.params && name !== "fs.read" ? { params: Schema.toCodecJson(endpoint.params) } : {}),
+    ...(name === "fs.read" ? { params: Schema.Struct({ path: Schema.String }) } : {}),
+    ...(endpoint.query ? { query: Schema.toCodecJson(Schema.toType(endpoint.query)) } : {}),
+    ...(endpoint.headers ? { headers: Schema.toCodecJson(endpoint.headers) } : {}),
+    ...(payloads.length ? { payload: Schema.toCodecJson(Schema.Union(payloads)) } : {}),
+  })
+  const error = RpcSchema.isStreamSchema(rpc.successSchema)
+    ? Schema.Union([rpc.errorSchema, rpc.successSchema.error])
+    : rpc.errorSchema
+  const value = { input: Schema.fromJsonString(input), output, error } as unknown as {
+    input: Schema.Codec<unknown, unknown>
+    output: Schema.Codec<unknown, unknown>
+    error: Schema.Codec<unknown, unknown>
+  }
+  codecs.set(name, value)
+  return value
+}
+
+/** A lazy, single-connection Promise facade. Replace it after a disconnect; calls are never replayed. */
+export function make(options: Options): Client {
+  const scope = Scope.makeUnsafe()
+  let client: Promise<OpenCodeRpc.Client> | undefined
+  let disposal: Promise<void> | undefined
+  const connect = Effect.suspend(() => {
+    if (disposal) return Effect.fail(new ClientError("Transport", { cause: new Error("RPC client disposed") }))
+    return Effect.promise(
+      () =>
+        (client ??= Effect.runPromise(
+          Effect.gen(function* () {
+            const url = new URL("/api/rpc", options.baseUrl)
+            url.protocol = url.protocol === "https:" || url.protocol === "wss:" ? "wss:" : "ws:"
+            const authorization = new Headers(options.headers).get("authorization")
+            const token = /^Basic\s+(.+)$/i.exec(authorization ?? "")?.[1]
+            return yield* OpenCodeRpc.make({
+              url,
+              authToken: token ? Redacted.make(token) : undefined,
+              webSocketConstructor: options.webSocketConstructor,
+            })
+          }).pipe(Effect.provideService(Scope.Scope, scope)),
+        )),
+    )
+  })
+
+  const invoke = Effect.fnUntraced(function* (descriptor: RequestDescriptor, requestOptions: RequestOptions) {
+    if (requestOptions.signal?.aborted)
+      return yield* Effect.fail(new ClientError("Transport", { cause: requestOptions.signal.reason }))
+    const codec = operation(descriptor.operation)
+    const headers = new Headers(requestOptions.headers)
+    // Authentication belongs only to the upgrade, not to user-controlled RPC frames.
+    headers.delete("authorization")
+    // Match the Promise wire boundary: omit undefined optional fields before decoding JSON codecs.
+    const input = yield* Schema.decodeUnknownEffect(codec.input)(
+      JSON.stringify({
+        params: descriptor.params,
+        query: descriptor.query ?? {},
+        payload: descriptor.body === undefined ? {} : descriptor.body,
+        headers: Object.fromEntries(headers),
+      }),
+    ).pipe(Effect.mapError((cause) => new ClientError("Transport", { cause })))
+    const api = yield* connect
+    // The generated operation/descriptor and protocol codec jointly establish this dynamic boundary.
+    const call = (
+      api as unknown as Record<
+        string,
+        (
+          input: unknown,
+          options: { headers: Record<string, string> },
+        ) => Effect.Effect<unknown, unknown> | Stream.Stream<unknown, unknown>
+      >
+    )[descriptor.operation]!
+    return call(input, { headers: Object.fromEntries(headers) })
+  })
+
+  return Object.assign(
+    OpenCode.make({
+      ...options,
+      transport: {
+        request(descriptor, requestOptions) {
+          const codec = operation(descriptor.operation)
+          return Effect.runPromise(
+            Effect.gen(function* () {
+              const result = yield* invoke(descriptor, requestOptions)
+              if (Stream.isStream(result)) return yield* Effect.fail(new ClientError("MalformedResponse"))
+              const value = yield* result
+              if (descriptor.empty) return undefined
+              if (descriptor.binary) return (value as { content: Uint8Array }).content
+              return yield* Schema.encodeUnknownEffect(codec.output)(value).pipe(
+                Effect.mapError((cause) => new ClientError("MalformedResponse", { cause })),
+              )
+            }),
+            { signal: requestOptions.signal },
+          ).catch((error) => {
+            throw wireError(error, codec.error)
+          })
+        },
+        stream(descriptor, requestOptions) {
+          const codec = operation(descriptor.operation)
+          const stream = Stream.unwrap(
+            invoke(descriptor, requestOptions).pipe(
+              Effect.map((result) =>
+                Stream.isStream(result) ? result : Stream.fail(new ClientError("MalformedResponse")),
+              ),
+            ),
+          ).pipe(
+            Stream.mapEffect((value) =>
+              Schema.encodeUnknownEffect(codec.output)(value).pipe(
+                Effect.mapError((cause) => new ClientError("MalformedResponse", { cause })),
+              ),
+            ),
+          )
+          const signal = requestOptions.signal
+          const abort = Effect.callback<never, ClientError>((resume) => {
+            if (!signal) return
+            const fail = () => resume(Effect.fail(new ClientError("Transport", { cause: signal.reason })))
+            if (signal.aborted) return fail()
+            signal.addEventListener("abort", fail, { once: true })
+            return Effect.sync(() => signal.removeEventListener("abort", fail))
+          })
+          return {
+            [Symbol.asyncIterator]() {
+              const iterator = Stream.toAsyncIterable(stream.pipe(Stream.interruptWhen(abort)))[Symbol.asyncIterator]()
+              return {
+                next: () =>
+                  iterator.next().catch((error) => {
+                    throw wireError(error, codec.error)
+                  }),
+                return: () => iterator.return!(),
+                throw: (error?: unknown) => iterator.throw!(error),
+              }
+            },
+          }
+        },
+      },
+    }),
+    {
+      dispose: () =>
+        (disposal ??= (async () => {
+          await client?.catch(() => {})
+          await Effect.runPromise(Scope.close(scope, Exit.void))
+        })()),
+    },
+  )
+}
+
+function wireError(error: unknown, schema: Schema.Codec<unknown, unknown>) {
+  if (error instanceof ClientError) return error
+  if (error instanceof RpcClientError.RpcClientError) return new ClientError("Transport", { cause: error })
+  const encoded = Schema.encodeUnknownExit(schema)(error)
+  return Exit.isSuccess(encoded) ? encoded.value : new ClientError("Transport", { cause: error })
+}

packages/client/test/import-boundaries.test.ts

@@ -28,6 +28,18 @@ describe("public import boundaries", () => {
     expect(within(network, core)).toEqual([])
     expect(within(network, server)).toEqual([])
 
+    const rpc = await bundleInputs("@opencode-ai/client/effect/rpc", "browser")
+    expect(within(rpc, effect).length).toBeGreaterThan(0)
+    expect(within(rpc, protocol).length).toBeGreaterThan(0)
+    expect(within(rpc, core)).toEqual([])
+    expect(within(rpc, server)).toEqual([])
+
+    const promiseRpc = await bundleInputs("@opencode-ai/client/promise/rpc", "browser")
+    expect(within(promiseRpc, effect).length).toBeGreaterThan(0)
+    expect(within(promiseRpc, protocol).length).toBeGreaterThan(0)
+    expect(within(promiseRpc, core)).toEqual([])
+    expect(within(promiseRpc, server)).toEqual([])
+
     const promiseService = await bundleInputs("@opencode-ai/client/service", "bun")
 
     expect(within(promiseService, effect)).toEqual([])

packages/client/test/promise-rpc.test.ts

@@ -0,0 +1,285 @@
+import { expect, test } from "bun:test"
+import { createServer } from "node:http"
+import { NodeHttpServer } from "@effect/platform-node"
+import { ClientApi } from "@opencode-ai/protocol/client"
+import { SessionNotFoundError } from "@opencode-ai/protocol/errors"
+import { fromEndpoint } from "@opencode-ai/protocol/rpc"
+import { Event } from "@opencode-ai/schema/event"
+import { Session } from "@opencode-ai/schema/session"
+import { DateTime, Effect, Schema, Stream } from "effect"
+import { RpcGroup, RpcSerialization, RpcServer } from "effect/unstable/rpc"
+import { ClientError, isSessionNotFoundError, type OpenCodeClient } from "../src/promise/index.js"
+import { OpenCodeRpc } from "../src/promise/rpc.js"
+
+const wire = {
+  id: "ses_test",
+  projectID: "prj_test",
+  cost: 0,
+  tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
+  time: { created: 1700000000000, updated: 1700000001000 },
+  location: { directory: "/project" },
+}
+const info = Schema.decodeUnknownSync(Schema.toCodecJson(Session.Info))(wire)
+
+test("Promise RPC preserves DTOs, numeric queries, optional payloads, errors, bytes and frame headers", async () => {
+  await withServer(async (baseUrl, state) => {
+    const sockets: WebSocket[] = []
+    const client = OpenCodeRpc.make({
+      baseUrl,
+      headers: {
+        authorization: "Basic dXNlcjpwYXNz",
+        "x-opencode-directory": "%2Fdefault",
+        "x-opencode-workspace": "wrk_default",
+      },
+      webSocketConstructor: (url, protocols) => {
+        expect(new URL(url).pathname).toBe("/api/rpc")
+        expect(new URL(url).searchParams.get("auth_token")).toBe("dXNlcjpwYXNz")
+        const socket = new WebSocket(url, protocols)
+        sockets.push(socket)
+        return socket
+      },
+    })
+    const api: OpenCodeClient = client
+    expect(sockets).toHaveLength(0)
+    try {
+      const [health, created, sessions, bytes] = await Promise.all([
+        api.health.get(),
+        api.session.create(),
+        api.session.list({ limit: 2, parentID: null }),
+        api.file.read({ path: "dir/space %25.bin", location: { directory: "/explicit", workspace: "wrk_explicit" } }),
+      ])
+      expect(health.healthy).toBe(true)
+      expect(created).toEqual(wire)
+      expect(created.time.created).toBeNumber()
+      expect(sessions.data).toEqual([wire])
+      expect(bytes).toEqual(new Uint8Array([0, 255, 128]))
+      expect(await api.session.import({ info: wire, messages: [] })).toEqual(wire)
+      expect(await api.session.rename({ sessionID: wire.id, title: "new title" })).toBeUndefined()
+      expect(
+        await api.form.list(
+          { sessionID: "global" },
+          {
+            headers: {
+              "x-opencode-directory": "%2Foverride",
+              "x-opencode-workspace": "wrk_override",
+              "x-opencode-ticket": "test-ticket",
+            },
+          },
+        ),
+      ).toEqual([])
+      const form = state.requests.find((request) => request.operation === "session.form.list")!
+      expect(form.input).toEqual({ params: { sessionID: "global" } })
+      expect(form.headers["x-opencode-directory"]).toBe("%2Foverride")
+      expect(form.headers["x-opencode-workspace"]).toBe("wrk_override")
+      expect(form.headers["x-opencode-ticket"]).toBe("test-ticket")
+      expect(form.headers.authorization).toBeUndefined()
+      expect(state.requests.find((request) => request.operation === "session.create")?.input).toEqual({ payload: {} })
+      expect(state.requests.find((request) => request.operation === "session.list")?.input).toEqual({
+        query: { limit: 2, parentID: null },
+      })
+      expect(state.requests.find((request) => request.operation === "fs.read")?.input).toEqual({
+        params: { path: "dir/space %25.bin" },
+        query: { location: { directory: "/explicit", workspace: "wrk_explicit" } },
+      })
+      const error = await api.session.get({ sessionID: "ses_missing" }).catch((error: unknown) => error)
+      expect(isSessionNotFoundError(error)).toBe(true)
+      expect(error).toEqual({ _tag: "SessionNotFoundError", sessionID: "ses_missing", message: "missing" })
+      expect(error).not.toBeInstanceOf(SessionNotFoundError)
+      expect(sockets).toHaveLength(1)
+    } finally {
+      await client.dispose()
+    }
+    expect(sockets[0]!.readyState).toBe(WebSocket.CLOSED)
+    await expect(client.health.get()).rejects.toBeInstanceOf(ClientError)
+    await client.dispose()
+    expect(sockets).toHaveLength(1)
+  })
+})
+
+test("iterator return, AbortSignal and disposal cancel only their owned RPC work", async () => {
+  await withServer(async (baseUrl, state) => {
+    const client = OpenCodeRpc.make({ baseUrl })
+    try {
+      const first = client.event.subscribe()[Symbol.asyncIterator]()
+      expect((await first.next()).value?.type).toBe("server.connected")
+      const pending = first.next()
+      await first.return!()
+      expect((await pending).done).toBe(true)
+      await state.streams[0]!.promise
+      expect((await client.health.get()).healthy).toBe(true)
+
+      const abort = new AbortController()
+      const second = client.event.subscribe({ signal: abort.signal })[Symbol.asyncIterator]()
+      await second.next()
+      const blocked = second.next().catch((error: unknown) => error)
+      abort.abort()
+      expect(await blocked).toBeInstanceOf(ClientError)
+      await state.streams[1]!.promise
+
+      const mutationAbort = new AbortController()
+      const mutation = client.session
+        .remove({ sessionID: wire.id }, { signal: mutationAbort.signal })
+        .catch((error: unknown) => error)
+      await state.removed.promise
+      mutationAbort.abort()
+      expect(await mutation).toBeInstanceOf(ClientError)
+      await state.removeStopped.promise
+      expect((await client.health.get()).healthy).toBe(true)
+
+      const third = client.event.subscribe()[Symbol.asyncIterator]()
+      await third.next()
+      const disposed = third.next().catch((error: unknown) => error)
+      await client.dispose()
+      expect(await disposed).toBeInstanceOf(ClientError)
+      await state.streams[2]!.promise
+      await expect(client.health.get()).rejects.toBeInstanceOf(ClientError)
+    } finally {
+      await client.dispose()
+    }
+  })
+})
+
+test("disconnect rejects in-flight mutations without replay; only a new facade reconnects", async () => {
+  await withServer(async (baseUrl, state) => {
+    const sockets: WebSocket[] = []
+    const options = {
+      baseUrl,
+      webSocketConstructor: (url: string, protocols?: string | string[]) => {
+        const socket = new WebSocket(url, protocols)
+        sockets.push(socket)
+        return socket
+      },
+    }
+    const client = OpenCodeRpc.make(options)
+    try {
+      const mutation = client.session.remove({ sessionID: wire.id }).catch((error: unknown) => error)
+      await state.removed.promise
+      sockets[0]!.close(1011, "disconnect")
+      expect(await mutation).toBeInstanceOf(ClientError)
+      await state.removeStopped.promise
+      await expect(client.health.get()).rejects.toBeInstanceOf(ClientError)
+      expect(sockets).toHaveLength(1)
+      const replacement = OpenCodeRpc.make(options)
+      try {
+        expect((await replacement.health.get()).healthy).toBe(true)
+        expect(sockets).toHaveLength(2)
+        expect(state.requests.filter((request) => request.operation === "session.remove")).toHaveLength(1)
+        await client.dispose()
+        expect((await replacement.health.get()).healthy).toBe(true)
+      } finally {
+        await replacement.dispose()
+      }
+    } finally {
+      await client.dispose()
+    }
+  })
+})
+
+test("disposing an unused facade and pre-aborted calls never open a socket", async () => {
+  let sockets = 0
+  const options = {
+    baseUrl: "http://localhost:1",
+    webSocketConstructor: (url: string) => {
+      sockets++
+      return new WebSocket(url)
+    },
+  }
+  const unused = OpenCodeRpc.make(options)
+  await unused.dispose()
+  await expect(unused.health.get()).rejects.toBeInstanceOf(ClientError)
+  const aborted = OpenCodeRpc.make(options)
+  try {
+    await expect(aborted.health.get({ signal: AbortSignal.abort() })).rejects.toBeInstanceOf(ClientError)
+    await expect(
+      aborted.event.subscribe({ signal: AbortSignal.abort() })[Symbol.asyncIterator]().next(),
+    ).rejects.toBeInstanceOf(ClientError)
+    expect(sockets).toBe(0)
+  } finally {
+    await aborted.dispose()
+  }
+})
+
+type State = {
+  requests: Array<{ operation: string; input: unknown; headers: Record<string, string> }>
+  streams: Array<ReturnType<typeof Promise.withResolvers<void>>>
+  removed: ReturnType<typeof Promise.withResolvers<void>>
+  removeStopped: ReturnType<typeof Promise.withResolvers<void>>
+}
+
+async function withServer(run: (baseUrl: string, state: State) => Promise<void>) {
+  await Effect.gen(function* () {
+    const state: State = {
+      requests: [],
+      streams: [],
+      removed: Promise.withResolvers(),
+      removeStopped: Promise.withResolvers(),
+    }
+    const record = (operation: string, input: unknown, headers: Record<string, string>) =>
+      state.requests.push({ operation, input, headers })
+    const group = RpcGroup.make(
+      fromEndpoint(ClientApi.groups["server.health"].endpoints["health.get"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.list"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.create"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.import"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.get"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.rename"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.remove"]),
+      fromEndpoint(ClientApi.groups["server.form"].endpoints["session.form.list"]),
+      fromEndpoint(ClientApi.groups["server.event"].endpoints["event.subscribe"]),
+      fromEndpoint(ClientApi.groups["server.fs"].endpoints["fs.read"]),
+    )
+    const app = yield* RpcServer.toHttpEffectWebsocket(group).pipe(
+      Effect.provide(
+        group.toLayer({
+          "health.get": () => Effect.succeed({ healthy: true, version: "test", pid: 0 }),
+          "session.list": (input, options) => {
+            record("session.list", input, options.headers)
+            return Effect.succeed({ data: [info], cursor: {} })
+          },
+          "session.create": (input, options) => {
+            record("session.create", input, options.headers)
+            return Effect.succeed({ data: info })
+          },
+          "session.import": (input) => {
+            expect(DateTime.isDateTime(input.payload.info.time.created)).toBe(true)
+            expect(DateTime.toEpochMillis(input.payload.info.time.created)).toBe(wire.time.created)
+            return Effect.succeed({ data: input.payload.info })
+          },
+          "session.get": ({ params }) =>
+            Effect.fail(new SessionNotFoundError({ sessionID: params.sessionID, message: "missing" })),
+          "session.rename": () => Effect.void,
+          "session.remove": (input, options) =>
+            Effect.gen(function* () {
+              record("session.remove", input, options.headers)
+              state.removed.resolve()
+              return yield* Effect.never
+            }).pipe(Effect.ensuring(Effect.sync(() => state.removeStopped.resolve()))),
+          "session.form.list": (input, options) => {
+            record("session.form.list", input, options.headers)
+            return Effect.succeed({ data: [] })
+          },
+          "fs.read": (input, options) => {
+            record("fs.read", input, options.headers)
+            return Effect.succeed({ content: new Uint8Array([0, 255, 128]), mime: "application/octet-stream" })
+          },
+          "event.subscribe": () => {
+            const stopped = Promise.withResolvers<void>()
+            state.streams.push(stopped)
+            return Stream.make({
+              id: Event.ID.make("evt_connected"),
+              type: "server.connected" as const,
+              data: {},
+            }).pipe(Stream.concat(Stream.never), Stream.ensuring(Effect.sync(() => stopped.resolve())))
+          },
+        }),
+      ),
+      Effect.provideService(RpcSerialization.RpcSerialization, RpcSerialization.json),
+    )
+    const server = yield* NodeHttpServer.make(createServer, { host: "127.0.0.1", port: 0 })
+    yield* server.serve(app)
+    if (server.address._tag !== "TcpAddress") return yield* Effect.die("Expected TCP listener")
+    yield* Effect.promise(() =>
+      run(`http://127.0.0.1:${server.address._tag === "TcpAddress" ? server.address.port : 0}`, state),
+    )
+  }).pipe(Effect.scoped, Effect.timeout("10 seconds"), Effect.runPromise)
+}

packages/client/test/rpc.test.ts

@@ -0,0 +1,105 @@
+import { expect, test } from "bun:test"
+import { createServer } from "node:http"
+import { NodeHttpServer } from "@effect/platform-node"
+import { ClientApi } from "@opencode-ai/protocol/client"
+import { SessionNotFoundError } from "@opencode-ai/protocol/errors"
+import { Group, fromEndpoint } from "@opencode-ai/protocol/rpc"
+import { Event } from "@opencode-ai/schema/event"
+import { Session } from "@opencode-ai/schema/session"
+import { Deferred, Effect, Fiber, Redacted, Stream } from "effect"
+import { RpcGroup, RpcSerialization, RpcServer } from "effect/unstable/rpc"
+import { OpenCodeRpc } from "../src/effect/rpc.js"
+
+test("one scoped socket multiplexes unary calls and cancellable typed streams", async () => {
+  const sockets: WebSocket[] = []
+  await Effect.gen(function* () {
+    const started = yield* Deferred.make<void>()
+    const stopped = yield* Deferred.make<void>()
+    const group = RpcGroup.make(
+      fromEndpoint(ClientApi.groups["server.health"].endpoints["health.get"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.list"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.get"]),
+      fromEndpoint(ClientApi.groups["server.session"].endpoints["session.remove"]),
+      fromEndpoint(ClientApi.groups["server.event"].endpoints["event.subscribe"]),
+      fromEndpoint(ClientApi.groups["server.fs"].endpoints["fs.read"]),
+    )
+    const app = yield* RpcServer.toHttpEffectWebsocket(group).pipe(
+      Effect.provide(
+        group.toLayer({
+          "health.get": () => Effect.succeed({ healthy: true, version: "test", pid: 0 }),
+          "session.list": () => Effect.succeed({ data: [], cursor: {} }),
+          "session.get": ({ params }) =>
+            Effect.fail(new SessionNotFoundError({ sessionID: params.sessionID, message: "missing" })),
+          "session.remove": () => Effect.void,
+          "fs.read": () => Effect.succeed({ content: new Uint8Array([0, 255, 128]), mime: "application/octet-stream" }),
+          "event.subscribe": () =>
+            Stream.fromEffect(Deferred.succeed(started, undefined)).pipe(
+              Stream.map(() => ({ id: Event.ID.make("evt_connected"), type: "server.connected" as const, data: {} })),
+              Stream.concat(Stream.never),
+              Stream.ensuring(Deferred.succeed(stopped, undefined)),
+            ),
+        }),
+      ),
+      Effect.provideService(RpcSerialization.RpcSerialization, RpcSerialization.json),
+    )
+    const server = yield* NodeHttpServer.make(createServer, { host: "127.0.0.1", port: 0 })
+    yield* server.serve(app)
+    if (server.address._tag !== "TcpAddress") return yield* Effect.die("Expected TCP listener")
+    yield* Effect.scoped(
+      Effect.gen(function* () {
+        const client = yield* OpenCodeRpc.make({
+          url: `ws://127.0.0.1:${server.address._tag === "TcpAddress" ? server.address.port : 0}/api/rpc`,
+          authToken: Redacted.make("test-token"),
+          webSocketConstructor: (url, protocols) => {
+            expect(new URL(url).searchParams.get("auth_token")).toBe("test-token")
+            const socket = new WebSocket(url, protocols)
+            sockets.push(socket)
+            return socket
+          },
+        })
+        const received = yield* Deferred.make<void>()
+        const events = yield* client["event.subscribe"]({}).pipe(
+          Stream.runForEach((event) => {
+            expect(event.type).toBe("server.connected")
+            return Deferred.succeed(received, undefined)
+          }),
+          Effect.forkScoped,
+        )
+        yield* Deferred.await(started)
+        yield* Deferred.await(received)
+        const [health, sessions, file] = yield* Effect.all(
+          [
+            client["health.get"]({}),
+            client["session.list"]({ query: {} }),
+            client["fs.read"]({ params: { path: "a.bin" }, query: {} }),
+          ],
+          { concurrency: "unbounded" },
+        )
+        expect(health).toEqual({ healthy: true, version: "test", pid: 0 })
+        expect(sessions).toEqual({ data: [], cursor: {} })
+        expect(file.content).toEqual(new Uint8Array([0, 255, 128]))
+        const error = yield* client["session.get"]({ params: { sessionID: Session.ID.make("ses_missing") } }).pipe(
+          Effect.flip,
+        )
+        expect(error).toBeInstanceOf(SessionNotFoundError)
+        expect(yield* client["session.remove"]({ params: { sessionID: Session.ID.make("ses_test") } })).toBeUndefined()
+        yield* Fiber.interrupt(events)
+        yield* Deferred.await(stopped)
+        expect((yield* client["health.get"]({})).healthy).toBe(true)
+        expect(sockets).toHaveLength(1)
+      }),
+    )
+  }).pipe(Effect.scoped, Effect.timeout("10 seconds"), Effect.runPromise)
+  expect(sockets[0]!.readyState).toBe(WebSocket.CLOSED)
+})
+
+// The derived client keeps operation-specific request and response types.
+type Client = Effect.Success<ReturnType<typeof OpenCodeRpc.make>>
+type HasRawPty = "pty.connect" extends keyof Client ? true : false
+const noRawPty: HasRawPty = false
+type Tags = RpcGroup.Rpcs<typeof Group>["_tag"]
+const fileTag: Tags = "fs.read"
+test("raw PTY is not part of the typed client", () => {
+  expect(noRawPty).toBe(false)
+  expect(fileTag).toBe("fs.read")
+})

packages/desktop/src/renderer/migration-status.tsx

@@ -1,4 +1,6 @@
-import { OpenCode, type MigrationV1StatusOutput } from "@opencode-ai/client/promise"
+import type { MigrationV1StatusOutput } from "@opencode-ai/client/promise"
+import { createDesktopServerApi } from "./platform/server-client"
+import { sidecarHttp } from "./startup/initialization"
 import { useLanguage } from "@opencode-ai/app/desktop"
 import { Loader } from "@opencode-ai/ui/loader"
 import { showToast, toaster, Toast } from "@opencode-ai/ui/toast"
@@ -11,6 +13,7 @@ export function MigrationStatus(props: { server: ServerReadyData }) {
   const language = useLanguage()
   const [progress, setProgress] = createSignal<Progress>()
   const abort = new AbortController()
+  const client = createDesktopServerApi(sidecarHttp(props.server))
   let toastID: number | undefined
   let disposeToast: (() => void) | undefined
 
@@ -49,16 +52,9 @@ export function MigrationStatus(props: { server: ServerReadyData }) {
     await wait(1_000, abort.signal)
     if (abort.signal.aborted) return
 
-    const client = OpenCode.make({
-      baseUrl: props.server.url,
-      headers: props.server.password
-        ? { Authorization: `Basic ${btoa(`${props.server.username ?? "opencode"}:${props.server.password}`)}` }
-        : undefined,
-    })
-
     void (async () => {
       while (true) {
-        const status = await client.migration.v1.status({ signal: abort.signal })
+        const status = await client.api.migration.v1.status({ signal: abort.signal })
         setProgress(status.status === "running" ? status.progress : undefined)
         if (status.status === "running") show()
         else hide()
@@ -66,20 +62,23 @@ export function MigrationStatus(props: { server: ServerReadyData }) {
         if (status.status === "error") throw new Error(status.error)
         await wait(1_000, abort.signal)
       }
-    })().catch((error) => {
-      if (abort.signal.aborted) return
-      hide()
-      showToast({
-        variant: "error",
-        title: language.t("toast.migration.failed.title"),
-        description: error instanceof Error ? error.message : String(error),
-        duration: 10_000,
+    })()
+      .catch((error) => {
+        if (abort.signal.aborted) return
+        hide()
+        showToast({
+          variant: "error",
+          title: language.t("toast.migration.failed.title"),
+          description: error instanceof Error ? error.message : String(error),
+          duration: 10_000,
+        })
       })
-    })
+      .finally(() => client.dispose())
   })
 
   onCleanup(() => {
     abort.abort()
+    void client.dispose()
     hide()
   })
 

packages/desktop/src/renderer/platform/index.ts

@@ -11,6 +11,7 @@ import { createDesktopFiles } from "./files"
 import { createDesktopMenuAction } from "./menu"
 import { createDesktopNotify } from "./notifications"
 import { createDesktopStorage } from "./storage"
+import { createDesktopServerApi } from "./server-client"
 
 export type DesktopWindowState = {
   id: string
@@ -40,6 +41,7 @@ export function createDesktopPlatform(
       if (input instanceof Request) return fetch(input)
       return fetch(input, init)
     },
+    createServerApi: createDesktopServerApi,
     getDefaultServer: async () => {
       const url = await api.getDefaultServerUrl().catch(() => null)
       if (!url) return null

packages/desktop/src/renderer/platform/server-client.ts

@@ -0,0 +1,12 @@
+import { OpenCodeRpc } from "@opencode-ai/client/promise/rpc"
+import type { ServerConnection } from "@opencode-ai/app/desktop"
+
+export function createDesktopServerApi(server: ServerConnection.HttpBase) {
+  const api = OpenCodeRpc.make({
+    baseUrl: server.url,
+    headers: server.password
+      ? { Authorization: `Basic ${btoa(`${server.username ?? "opencode"}:${server.password}`)}` }
+      : undefined,
+  })
+  return { api, dispose: () => api.dispose() }
+}

packages/httpapi-codegen/src/index.ts

@@ -930,13 +930,17 @@ function renderPromiseClient(groups: ReadonlyArray<Group>) {
             ? access(inputs[0].name)
             : `{ ${inputs.map((field) => `${JSON.stringify(field.name)}: ${access(field.name)}`).join(", ")} }`
       }
+      const params = promiseInput(endpoint).filter((field) => field.source === "params" || field.source === "wildcard")
       const parts = [
+        params.length === 0 && endpoint.params === undefined
+          ? undefined
+          : `params: { ${params.map((field) => `${JSON.stringify(field.name)}: ${access(field.name)}`).join(", ")} }`,
         endpoint.query === undefined ? undefined : `query: ${part("query")}`,
         endpoint.headers === undefined ? undefined : `headers: ${part("headers")}`,
         endpoint.payloads.length === 0 ? undefined : `body: ${part("payload")}`,
       ].filter((value): value is string => value !== undefined)
       const declaredStatuses = [...new Set(endpoint.errors.map((error) => error.status))]
-      const descriptor = `{ method: ${JSON.stringify(endpoint.endpoint.method)}, path: ${path}${parts.length === 0 ? "" : `, ${parts.join(", ")}`}, successStatus: ${resolveHttpApiStatus(endpoint.successes[0].ast) ?? 200}, declaredStatuses: [${declaredStatuses.join(", ")}], empty: ${endpoint.operation.success === "void"}${isBinarySchema(endpoint.successes[0]) ? ", binary: true" : ""} }`
+      const descriptor = `{ operation: ${JSON.stringify(endpoint.endpoint.identifier)}, method: ${JSON.stringify(endpoint.endpoint.method)}, path: ${path}${parts.length === 0 ? "" : `, ${parts.join(", ")}`}, successStatus: ${resolveHttpApiStatus(endpoint.successes[0].ast) ?? 200}, declaredStatuses: [${declaredStatuses.join(", ")}], empty: ${endpoint.operation.success === "void"}${isBinarySchema(endpoint.successes[0]) ? ", binary: true" : ""} }`
       if (endpoint.operation.success === "stream") {
         const success = endpoint.successes[0]
         if (!isStreamSchema(success) || success._tag !== "StreamSse" || success.sseMode !== "data") {
@@ -1224,7 +1228,54 @@ function normalizePromiseClientContent(content: string, groups: ReadonlyArray<Gr
   const usesBinary = endpoints.some((endpoint) => isBinarySchema(endpoint.successes[0]))
   const usesWildcard = endpoints.some((endpoint) => promiseWildcardInput(endpoint) !== undefined)
 
-  const sseReady = replaceOne(content, "let next: ReadableStreamReadResult<Uint8Array>", "let next")
+  const transportReady = [
+    [
+      "readonly fetch?: typeof globalThis.fetch",
+      "readonly fetch?: typeof globalThis.fetch\n  readonly transport?: ClientTransport",
+    ],
+    [
+      "interface RequestDescriptor {",
+      `export interface ClientTransport {
+  readonly request: (descriptor: RequestDescriptor, options: RequestOptions) => Promise<unknown>
+  readonly stream: (descriptor: RequestDescriptor, options: RequestOptions) => AsyncIterable<unknown>
+}
+
+export interface RequestDescriptor {
+  readonly operation: string
+  readonly params?: Record<string, unknown>`,
+    ],
+    [
+      "  const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n    const url = new URL(descriptor.path, options.baseUrl)\n    for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n",
+      "  const prepareHeaders = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n",
+    ],
+    [
+      '    if (descriptor.body !== undefined && !headers.has("content-type"))',
+      `    return headers
+  }
+
+  const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {
+    const url = new URL(descriptor.path, options.baseUrl)
+    for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)
+    const headers = prepareHeaders(descriptor, requestOptions)
+    if (descriptor.body !== undefined && !headers.has("content-type"))`,
+    ],
+    [
+      "    const response = await execute(descriptor, requestOptions)",
+      `    if (options.transport) return await options.transport.request(descriptor, {
+      ...requestOptions, headers: prepareHeaders(descriptor, requestOptions),
+    }) as A
+    const response = await execute(descriptor, requestOptions)`,
+    ],
+    [
+      "const sse = <A>(descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable<A> => ({",
+      `const sse = <A>(descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable<A> => options.transport
+    ? options.transport.stream(descriptor, {
+        ...requestOptions, headers: prepareHeaders(descriptor, requestOptions),
+      }) as AsyncIterable<A>
+    : ({`,
+    ],
+  ].reduce((source, [search, replacement]) => replaceOne(source, search!, replacement!), content)
+  const sseReady = replaceOne(transportReady, "let next: ReadableStreamReadResult<Uint8Array>", "let next")
   const binaryReady = usesBinary
     ? replaceOne(
         replaceOne(sseReady, "readonly empty: boolean\n}", "readonly empty: boolean\n  readonly binary?: true\n}"),

packages/httpapi-codegen/test/generate.test.ts

@@ -814,6 +814,75 @@ describe("HttpApiCodegen.generate", () => {
     ).toThrow("Unsupported Promise stream: session.events")
   })
 
+  test("passes operation descriptors and merged options directly to a Promise transport", async () => {
+    const output = emitPromise(
+      compileContract(
+        HttpApi.make("test").add(
+          HttpApiGroup.make("session").add(
+            HttpApiEndpoint.post("update", "/session/:sessionID", {
+              params: { sessionID: Schema.String },
+              query: { limit: Schema.NumberFromString },
+              headers: { "x-ticket": Schema.String },
+              payload: Schema.Struct({ title: Schema.String }),
+              success: Schema.Struct({ data: Schema.String }),
+            }),
+            HttpApiEndpoint.get("events", "/events", { success: HttpApiSchema.StreamSse({ data: Schema.String }) }),
+          ),
+        ),
+      ),
+    )
+    const directory = await mkdtemp(join(tmpdir(), "opencode-httpapi-codegen-"))
+    try {
+      await Promise.all(output.files.map((file) => Bun.write(join(directory, file.path), file.content)))
+      const generated = await import(`${join(directory, "index.ts")}?t=${crypto.randomUUID()}`)
+      const signal = new AbortController().signal
+      const iterable = {
+        async *[Symbol.asyncIterator]() {
+          yield "event"
+        },
+      }
+      const client = generated.OpenCode.make({
+        baseUrl: "https://example.com",
+        headers: { "x-default": "default", "x-ticket": "default" },
+        fetch: () => {
+          throw new Error("transport must not fetch")
+        },
+        transport: {
+          request: async (descriptor: unknown, options: { headers: Headers; signal: AbortSignal }) => {
+            expect(descriptor).toMatchObject({
+              operation: "update",
+              params: { sessionID: "a/b" },
+              query: { limit: 2 },
+              body: { title: "title" },
+            })
+            expect(options.signal).toBe(signal)
+            expect(options.headers.get("x-default")).toBe("default")
+            expect(options.headers.get("x-ticket")).toBe("override")
+            return { data: "updated" }
+          },
+          stream: (descriptor: { operation: string }, options: { headers: Headers; signal: AbortSignal }) => {
+            expect(descriptor.operation).toBe("events")
+            expect(options.signal).toBe(signal)
+            expect(options.headers.get("x-default")).toBe("default")
+            return iterable
+          },
+        },
+      })
+      expect(
+        await client.session.update(
+          { sessionID: "a/b", limit: 2, "x-ticket": "endpoint", title: "title" },
+          {
+            signal,
+            headers: { "x-ticket": "override" },
+          },
+        ),
+      ).toBe("updated")
+      expect(client.session.events({ signal })).toBe(iterable)
+    } finally {
+      await rm(directory, { recursive: true, force: true })
+    }
+  })
+
   test("executes an emitted Promise GET through fetch", async () => {
     const output = emitPromise(
       compileContract(

packages/protocol/src/rpc.ts

@@ -0,0 +1,117 @@
+export * as OpenCodeRpc from "./rpc.js"
+
+import { Predicate, Schema, SchemaAST, Stream } from "effect"
+import { HttpApi, HttpApiEndpoint, HttpApiGroup, HttpApiMiddleware, HttpApiSchema } from "effect/unstable/httpapi"
+import { Rpc, RpcGroup, RpcSchema } from "effect/unstable/rpc"
+import { ClientApi } from "./client.js"
+import { LocationQuery } from "./groups/location.js"
+
+export const omitEndpoints: ReadonlySet<string> = new Set(["pty.connect", "persistentPty.connect"])
+
+export const FileRead = Schema.Struct({ content: Schema.Uint8ArrayFromBase64, mime: Schema.String })
+export const FileReadParams = Schema.Struct({ path: Schema.String })
+
+type Part<K extends string, S extends Schema.Constraint> = [S] extends [never] ? {} : { readonly [P in K]: S["Type"] }
+
+export type Request<E extends HttpApiEndpoint.ConstraintRequest> = (E["identifier"] extends "fs.read"
+  ? { readonly params: typeof FileReadParams.Type }
+  : Part<"params", E["~Params"]>) &
+  Part<"query", E["~Query"]> &
+  Part<"payload", E["~Payload"]> &
+  Part<"headers", E["~Headers"]> & {
+    readonly location?: typeof LocationQuery.Type.location
+  }
+
+type Success<S extends Schema.Constraint> =
+  S["Type"] extends Stream.Stream<infer A, infer E>
+    ? RpcSchema.Stream<Schema.Codec<A, unknown>, Schema.Codec<E, unknown>>
+    : Schema.Codec<S["Type"], unknown>
+
+export type Endpoint<E extends HttpApiEndpoint.ConstraintRequest> = E extends HttpApiEndpoint.ConstraintRequest
+  ? E["identifier"] extends "pty.connect" | "persistentPty.connect"
+    ? never
+    : Rpc.Rpc<
+        E["identifier"],
+        Schema.Codec<Request<E>, unknown>,
+        E["identifier"] extends "fs.read" ? typeof FileRead : Success<E["~Success"]>,
+        Schema.Codec<E["~Error"]["Type"] | HttpApiMiddleware.Error<E["~Middleware"]>, unknown>
+      >
+  : never
+
+type Endpoints<G extends HttpApiGroup.Constraint> =
+  HttpApiGroup.Endpoints<G> extends infer E
+    ? E extends HttpApiEndpoint.ConstraintRequest
+      ? Endpoint<E>
+      : never
+    : never
+
+/** Convert HTTP stream declarations to RPC streams without serializing an SSE envelope. */
+export function successSchema(schema: Schema.Top): Schema.Top {
+  if (RpcSchema.isStreamSchema(schema)) return schema
+  if (!isHttpStream(schema)) return Schema.toCodecJson(schema)
+  if (schema._tag === "StreamUint8Array") {
+    return RpcSchema.Stream(Schema.Uint8ArrayFromBase64, Schema.Unknown)
+  }
+  if (schema.sseMode === "events") {
+    return RpcSchema.Stream(Schema.toCodecJson(schema.events), Schema.toCodecJson(schema.error))
+  }
+  // StreamSse({ data }) stores its original data codec in the event struct's data field.
+  const ast = SchemaAST.toType(schema.events.ast)
+  const data = SchemaAST.isObjects(ast) ? ast.propertySignatures.find((field) => field.name === "data") : undefined
+  if (!data) throw new Error("SSE data schema is missing its data field")
+  return RpcSchema.Stream(Schema.toCodecJson(Schema.make(data.type)), Schema.toCodecJson(schema.error))
+}
+
+function isHttpStream(schema: Schema.Top): schema is HttpApiSchema.StreamSchema {
+  return Predicate.hasProperty(schema, "~effect/httpapi/HttpApiSchema/Stream")
+}
+
+export function fromEndpoint<E extends HttpApiEndpoint.ConstraintRequest>(input: E): Endpoint<E> {
+  const endpoint = input as unknown as HttpApiEndpoint.Top
+  if (omitEndpoints.has(endpoint.identifier)) throw new Error(`Raw WebSocket endpoint: ${endpoint.identifier}`)
+  const payload = Array.from(endpoint.payload.values()).flatMap((entry) => entry.schemas)
+  const success = endpoint.success.size ? Array.from(endpoint.success) : [HttpApiSchema.NoContent]
+  const middleware = Array.from(endpoint.middlewares) as unknown as HttpApiMiddleware.AnyService[]
+  const errors = [...endpoint.error, ...middleware.flatMap((service) => Array.from(service.error))]
+  if (success.length > 1 && success.some(isHttpStream)) {
+    throw new Error(`Mixed streaming responses are not supported: ${endpoint.identifier}`)
+  }
+  const request = Schema.Struct({
+    ...(endpoint.identifier === "fs.read"
+      ? { params: FileReadParams }
+      : endpoint.params
+        ? { params: Schema.toCodecJson(Schema.toType(endpoint.params)) }
+        : {}),
+    ...(endpoint.query ? { query: Schema.toCodecJson(Schema.toType(endpoint.query)) } : {}),
+    ...(endpoint.headers ? { headers: Schema.toCodecJson(Schema.toType(endpoint.headers)) } : {}),
+    ...(payload.length ? { payload: Schema.toCodecJson(Schema.toType(Schema.Union(payload))) } : {}),
+    location: LocationQuery.fields.location,
+  })
+  return Rpc.make(endpoint.identifier, {
+    payload: request,
+    success:
+      endpoint.identifier === "fs.read"
+        ? FileRead
+        : success.length === 1
+          ? successSchema(success[0]!)
+          : Schema.Union(success.map(successSchema)),
+    error: Schema.toCodecJson(Schema.Union([...new Set(errors)])),
+  }) as unknown as Endpoint<E>
+}
+
+/** Server passes its concrete HttpApi so its middleware error schemas are retained. */
+export function makeGroup<Id extends string, G extends HttpApiGroup.Constraint>(
+  api: HttpApi.HttpApi<Id, G>,
+): RpcGroup.RpcGroup<Endpoints<G>> {
+  const groups = Object.values((api as unknown as HttpApi.Top).groups)
+  return RpcGroup.make(
+    ...groups.flatMap((group) =>
+      Object.values(group.endpoints)
+        .filter((endpoint) => !omitEndpoints.has(endpoint.identifier))
+        .map(fromEndpoint),
+    ),
+  ) as unknown as RpcGroup.RpcGroup<Endpoints<G>>
+}
+
+export type Rpcs = Endpoints<(typeof ClientApi.groups)[keyof typeof ClientApi.groups]>
+export const Group: RpcGroup.RpcGroup<Rpcs> = makeGroup(ClientApi)

packages/protocol/test/rpc.test.ts

@@ -0,0 +1,81 @@
+import { expect, test } from "bun:test"
+import { Schema } from "effect"
+import { Event } from "@opencode-ai/schema/event"
+import { HttpApi, HttpApiEndpoint, HttpApiGroup, HttpApiMiddleware, HttpApiSchema } from "effect/unstable/httpapi"
+import { RpcSchema } from "effect/unstable/rpc"
+import { ClientApi } from "../src/client.js"
+import { InvalidRequestError, SessionNotFoundError, UnauthorizedError } from "../src/errors.js"
+import { OpenCodeRpc } from "../src/rpc.js"
+
+test("every HTTP operation except the raw PTY sockets has a native RPC", () => {
+  const expected = Object.values(ClientApi.groups).flatMap((group) =>
+    Object.values(group.endpoints).map((endpoint) => endpoint.identifier),
+  )
+  expect([...OpenCodeRpc.Group.requests.keys()].sort()).toEqual(
+    expected.filter((name) => !OpenCodeRpc.omitEndpoints.has(name)).sort(),
+  )
+})
+
+test("request envelopes preserve decoded numeric queries and validate required params", () => {
+  const list = OpenCodeRpc.fromEndpoint(ClientApi.groups["server.session"].endpoints["session.list"])
+  expect(Schema.decodeUnknownSync(list.payloadSchema)({ query: { limit: 10 } })).toEqual({ query: { limit: 10 } })
+  expect(() => Schema.decodeUnknownSync(list.payloadSchema)({ query: { limit: "10" } })).toThrow()
+  expect(() => Schema.decodeUnknownSync(list.payloadSchema)({ query: { limit: -1 } })).toThrow()
+  const get = OpenCodeRpc.fromEndpoint(ClientApi.groups["server.session"].endpoints["session.get"])
+  expect(() => Schema.decodeUnknownSync(get.payloadSchema)({})).toThrow()
+})
+
+test("SSE streams retain their typed items, without SSE framing", () => {
+  const rpc = OpenCodeRpc.fromEndpoint(ClientApi.groups["server.session"].endpoints["session.log"])
+  expect(RpcSchema.isStreamSchema(rpc.successSchema)).toBe(true)
+  const item = { type: "log.synced" as const, aggregateID: "ses_test", seq: Event.Seq.make(1) }
+  expect(Schema.decodeUnknownSync(rpc.successSchema.success)(item)).toEqual(item)
+  expect(() => Schema.decodeUnknownSync(rpc.successSchema.success)({ type: "not-an-event" })).toThrow()
+  const events = OpenCodeRpc.fromEndpoint(ClientApi.groups["server.event"].endpoints["event.subscribe"])
+  expect(
+    Schema.decodeUnknownSync(events.successSchema.success)({ id: "evt_connected", type: "server.connected", data: {} }),
+  ).toEqual({ id: Event.ID.make("evt_connected"), type: "server.connected", data: {} })
+})
+
+test("binary file reads have explicit path and base64 JSON codecs", () => {
+  const rpc = OpenCodeRpc.fromEndpoint(ClientApi.groups["server.fs"].endpoints["fs.read"])
+  expect(Schema.decodeUnknownSync(rpc.payloadSchema)({ params: { path: "a.bin" }, query: {} })).toEqual({
+    params: { path: "a.bin" },
+    query: {},
+  })
+  const value = { content: new Uint8Array([0, 255, 128]), mime: "application/octet-stream" }
+  const encoded = Schema.encodeSync(rpc.successSchema)(value)
+  expect(encoded).toEqual({ content: "AP+A", mime: "application/octet-stream" })
+  expect(Schema.decodeUnknownSync(rpc.successSchema)(JSON.parse(JSON.stringify(encoded)))).toEqual(value)
+})
+
+test("NoContent remains void and endpoint plus middleware errors remain typed", () => {
+  const rpc = OpenCodeRpc.fromEndpoint(HttpApiEndpoint.delete("test.remove", "/test"))
+  expect(Schema.encodeSync(rpc.successSchema)(undefined)).toBeNull()
+  expect(Schema.decodeUnknownSync(rpc.successSchema)(null)).toBeUndefined()
+  const get = OpenCodeRpc.fromEndpoint(ClientApi.groups["server.session"].endpoints["session.get"])
+  for (const error of [
+    new InvalidRequestError({ message: "invalid" }),
+    new UnauthorizedError({ message: "unauthorized" }),
+    new SessionNotFoundError({ sessionID: "ses_test", message: "missing" }),
+  ]) {
+    expect(Schema.decodeUnknownSync(get.errorSchema)(Schema.encodeSync(get.errorSchema)(error))).toEqual(error)
+  }
+})
+
+test("makeGroup retains concrete server middleware errors and stream errors", () => {
+  class TestError extends Schema.TaggedError<TestError>()("TestError", { message: Schema.String }) {}
+  class Middleware extends HttpApiMiddleware.Service<Middleware>()("test/rpc", { error: TestError }) {}
+  const api = HttpApi.make("test").add(
+    HttpApiGroup.make("test").add(
+      HttpApiEndpoint.get("test.stream", "/test", {
+        success: HttpApiSchema.StreamSse({ data: Schema.Number, error: TestError }),
+      }).middleware(Middleware),
+    ),
+  )
+  const rpc = OpenCodeRpc.makeGroup(api).requests.get("test.stream")!
+  const error = new TestError({ message: "failed" })
+  expect(Schema.decodeUnknownSync(rpc.errorSchema)(Schema.encodeSync(rpc.errorSchema)(error))).toEqual(error)
+  expect(Schema.decodeUnknownSync(rpc.successSchema.error)({ _tag: "TestError", message: "failed" })).toEqual(error)
+  expect(Schema.decodeUnknownSync(rpc.successSchema.success)(42)).toBe(42)
+})

packages/server/script/benchmark-rpc.ts

@@ -0,0 +1,90 @@
+import fs from "node:fs/promises"
+import os from "node:os"
+import path from "node:path"
+import { ClientApi } from "@opencode-ai/protocol/client"
+import { OpenCodeRpc } from "@opencode-ai/protocol/rpc"
+import { AbsolutePath } from "@opencode-ai/schema/schema"
+import { Effect, Layer } from "effect"
+import { FetchHttpClient, HttpClient, HttpClientRequest, HttpServer } from "effect/unstable/http"
+import { HttpApiClient } from "effect/unstable/httpapi"
+import { RpcClient, RpcSerialization } from "effect/unstable/rpc"
+import { Socket } from "effect/unstable/socket"
+import { ServerProcess } from "../src/process"
+
+// A loopback transport microbenchmark, not an end-to-end desktop performance claim.
+// Both clients decode the same schemas and read the same 50-session page.
+await Effect.gen(function* () {
+  const directory = yield* Effect.acquireRelease(
+    Effect.promise(() => fs.mkdtemp(path.join(os.tmpdir(), "opencode-rpc-bench-"))),
+    (directory) => Effect.promise(() => fs.rm(directory, { recursive: true, force: true })),
+  )
+  const password = crypto.randomUUID()
+  const server = yield* ServerProcess.start<never, never>({
+    hostname: "127.0.0.1",
+    port: 0,
+    password,
+    database: { path: ":memory:" },
+    config: { directory: path.join(directory, "config"), project: false },
+    models: { fetch: false },
+    fs: { filewatcher: false, fff: false },
+  })
+  const url = HttpServer.formatAddress(server.address)
+  const http = yield* HttpApiClient.make(ClientApi, {
+    baseUrl: url,
+    transformClient: (client) =>
+      HttpClient.mapRequest(
+        client,
+        HttpClientRequest.setHeader("authorization", `Basic ${btoa(`opencode:${password}`)}`),
+      ),
+  }).pipe(Effect.provide(FetchHttpClient.layer))
+  const socket = new URL("/api/rpc", url)
+  socket.protocol = "ws:"
+  socket.searchParams.set("auth_token", btoa(`opencode:${password}`))
+  const protocol = yield* Layer.build(
+    RpcClient.layerProtocolSocket({ retryTransientErrors: false }).pipe(
+      Layer.provide(RpcSerialization.layerJson),
+      Layer.provide(Socket.layerWebSocket(socket.href).pipe(Layer.provide(Socket.layerWebSocketConstructorGlobal))),
+    ),
+  )
+  const rpc = yield* RpcClient.make(OpenCodeRpc.Group).pipe(Effect.provideContext(protocol))
+  yield* Effect.forEach(
+    Array.from({ length: 50 }, (_, index) => index),
+    (index) =>
+      rpc["session.create"]({
+        payload: { title: `Benchmark ${index}`, location: { directory: AbsolutePath.make(directory) } },
+      }),
+  )
+  const calls: ReadonlyArray<{ transport: string; call: Effect.Effect<unknown, unknown> }> = [
+    { transport: "HTTP", call: http["server.session"]["session.list"]({ query: {} }) },
+    { transport: "RPC", call: rpc["session.list"]({ query: {} }) },
+  ]
+  for (const concurrency of [1, 16]) {
+    for (const { transport, call } of calls) {
+      yield* Effect.forEach(Array.from({ length: 30 }), () => call, { concurrency })
+      const start = performance.now()
+      const samples = yield* Effect.forEach(
+        Array.from({ length: 300 }),
+        () =>
+          Effect.gen(function* () {
+            const start = performance.now()
+            yield* call
+            return performance.now() - start
+          }),
+        { concurrency },
+      )
+      const total = performance.now() - start
+      samples.sort((a, b) => a - b)
+      console.log(
+        JSON.stringify({
+          transport,
+          concurrency,
+          requests: samples.length,
+          elapsedMs: Math.round(total),
+          requestsPerSecond: Math.round((samples.length / total) * 1000),
+          p50Ms: Number(samples[Math.floor(samples.length * 0.5)]!.toFixed(2)),
+          p95Ms: Number(samples[Math.floor(samples.length * 0.95)]!.toFixed(2)),
+        }),
+      )
+    }
+  }
+}).pipe(Effect.timeout("60 seconds"), Effect.scoped, Effect.runPromise)

packages/server/src/event-feed.ts

@@ -22,6 +22,7 @@ export type Error = SubscriberOverflowError | EncodingError
 
 export interface Interface {
   readonly subscribe: Effect.Effect<Stream.Stream<string, Error>, never, Scope.Scope>
+  readonly subscribeEvents: Effect.Effect<Stream.Stream<OpenCodeEvent, Error>, never, Scope.Scope>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/server/EventFeed") {}
@@ -37,6 +38,7 @@ export const make = Effect.fn("EventFeed.make")(function* (
   const capacity = options?.capacity ?? SubscriberCapacity
   const render = options?.encode ?? frame
   const subscribers = new Set<Queue.Queue<string, Error>>()
+  const events = new Set<Queue.Queue<OpenCodeEvent, Error>>()
 
   const fail = (error: Error) =>
     Effect.sync(() => {
@@ -47,6 +49,11 @@ export const make = Effect.fn("EventFeed.make")(function* (
 
   const publish = Effect.fnUntraced(function* (event: Event.Payload) {
     if (!isOpenCodeEvent(event)) return
+    for (const subscriber of events) {
+      if (Queue.offerUnsafe(subscriber, event)) continue
+      events.delete(subscriber)
+      Queue.failCauseUnsafe(subscriber, Cause.fail(new SubscriberOverflowError({ capacity })))
+    }
     if (subscribers.size === 0) return
     const encoded = yield* Effect.try({
       try: () => render(event),
@@ -72,6 +79,10 @@ export const make = Effect.fn("EventFeed.make")(function* (
   yield* Effect.addFinalizer(() => unsubscribe)
 
   return Service.of({
+    subscribeEvents: Effect.acquireRelease(
+      Queue.dropping<OpenCodeEvent, Error>(capacity).pipe(Effect.tap((queue) => Effect.sync(() => events.add(queue)))),
+      (queue) => Effect.sync(() => events.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid),
+    ).pipe(Effect.map(Stream.fromQueue)),
     subscribe: Effect.acquireRelease(
       Queue.dropping<string, Error>(capacity).pipe(Effect.tap((queue) => Effect.sync(() => subscribers.add(queue)))),
       (queue) =>

packages/server/src/handlers/fs.ts

@@ -6,17 +6,19 @@ import { HttpApiBuilder } from "effect/unstable/httpapi"
 import { Api } from "../api"
 import { response } from "../location"
 
+export const readFile = Effect.fn("Server.readFile")(function* (path: RelativePath) {
+  const fs = yield* FileSystem.Service
+  return yield* fs.read({ path })
+})
+
 export const FileSystemHandler = HttpApiBuilder.group(Api, "server.fs", (handlers) =>
   Effect.gen(function* () {
     return handlers
       .handleRaw("fs.read", (ctx) =>
         Effect.gen(function* () {
-          const fs = yield* FileSystem.Service
-          const file = yield* fs.read({
-            path: RelativePath.make(
-              decodeURIComponent(new URL(ctx.request.url, "http://localhost").pathname.slice(13)),
-            ),
-          })
+          const file = yield* readFile(
+            RelativePath.make(decodeURIComponent(new URL(ctx.request.url, "http://localhost").pathname.slice(13))),
+          )
           return HttpServerResponse.uint8Array(file.content, { contentType: file.mime })
         }),
       )

packages/server/src/routes.ts

@@ -36,6 +36,8 @@ import { Context, Effect, Layer, Option } from "effect"
 import { Api } from "./api"
 import { ServerAuth } from "./auth"
 import { handlers } from "./handlers"
+import { rpcRoutes } from "./rpc"
+import { EventFeed } from "./event-feed"
 import { authorizationLayer } from "./middleware/authorization"
 import { schemaErrorLayer } from "./middleware/schema-error"
 import { PtyEnvironment } from "./pty-environment"
@@ -147,7 +149,8 @@ function makeRoutes<AuthError, AuthServices>(
         ),
         ServerInfo.layer(serviceURLs, options.app),
       )
-      const api = HttpApiBuilder.layer(Api, { openapiPath: "/openapi.json" }).pipe(
+      const api = Layer.merge(HttpApiBuilder.layer(Api, { openapiPath: "/openapi.json" }), rpcRoutes).pipe(
+        Layer.provide(EventFeed.layer.pipe(Layer.provide(services))),
         Layer.provide(handlers.pipe(Layer.provide(services))),
         Layer.provide(formLocationLayer),
         Layer.provide(sessionLocationLayer),

packages/server/src/rpc.ts

@@ -0,0 +1,166 @@
+import { Event } from "@opencode-ai/schema/event"
+import { RelativePath } from "@opencode-ai/core/schema"
+import { Context, Effect, Layer, Scope, Stream } from "effect"
+import { Headers, HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
+import { HttpApiGroup, HttpApiEndpoint } from "effect/unstable/httpapi"
+import { Rpc, RpcGroup, RpcSchema, RpcSerialization, RpcServer } from "effect/unstable/rpc"
+import { OpenCodeRpc } from "@opencode-ai/protocol/rpc"
+import { Api } from "./api"
+import { ServerAuth } from "./auth"
+import { CorsConfig, isAllowedRequestOrigin } from "./cors"
+import { EventFeed } from "./event-feed"
+import type { handlers } from "./handlers"
+import { readFile } from "./handlers/fs"
+import { LocationMiddleware } from "./location"
+import { Authorization, authorizedRequest } from "./middleware/authorization"
+import { FormLocationMiddleware } from "./middleware/form-location"
+import { SessionLocationMiddleware } from "./middleware/session-location"
+import { SchemaErrorMiddleware } from "./middleware/schema-error"
+
+type Services =
+  | Layer.Success<typeof handlers>
+  | ServerAuth.Config
+  | EventFeed.Service
+  | LocationMiddleware
+  | FormLocationMiddleware
+  | SessionLocationMiddleware
+  | Authorization
+  | SchemaErrorMiddleware
+
+type Input = {
+  readonly params?: Readonly<Record<string, string>>
+  readonly query?: { readonly location?: { readonly directory?: string; readonly workspace?: string } }
+  readonly payload?: unknown
+  readonly headers?: Readonly<Record<string, string | undefined>>
+  readonly location?: { readonly directory?: string; readonly workspace?: string }
+}
+
+type Middleware = (
+  effect: Effect.Effect<unknown, unknown, unknown>,
+  options: { readonly group: HttpApiGroup.Top; readonly endpoint: HttpApiEndpoint.Top },
+) => Effect.Effect<unknown, unknown, unknown>
+
+type Handler = {
+  readonly endpoint: HttpApiEndpoint.Top
+  readonly uninterruptible: boolean
+  readonly handler: (
+    input: Input & {
+      readonly request: HttpServerRequest.HttpServerRequest
+      readonly group: HttpApiGroup.Top
+      readonly endpoint: HttpApiEndpoint.Top
+    },
+  ) => Effect.Effect<unknown, unknown, unknown>
+}
+
+export const rpcRoutes = HttpRouter.use((router) =>
+  Effect.gen(function* () {
+    const services = yield* Effect.context<Services>()
+    const config = yield* ServerAuth.Config
+    const feed = yield* EventFeed.Service
+    const group = OpenCodeRpc.makeGroup(Api)
+    // HttpApiBuilder stores its decoded handlers beside the built routes. Keep this
+    // dependency on Effect's handler registry here, rather than duplicating handlers
+    // or routing RPC calls through HTTP serialization and parsing.
+    const entries = (Object.values(Api.groups) as unknown as ReadonlyArray<HttpApiGroup.Top>).flatMap((definition) => {
+      const implementation = services.mapUnsafe.get(definition.key) as {
+        readonly handlers: ReadonlyMap<string, Handler>
+      }
+      return Array.from(implementation.handlers.values(), (handler) => ({ definition, ...handler }))
+    })
+    yield* router.add(
+      "GET",
+      "/api/rpc",
+      Effect.gen(function* () {
+        const request = yield* HttpServerRequest.HttpServerRequest
+        if (!(yield* authorizedRequest(request, config)))
+          return HttpServerResponse.empty({
+            status: 401,
+            headers: { "www-authenticate": 'Basic realm="Secure Area"' },
+          })
+        const cors = yield* CorsConfig
+        if (!isAllowedRequestOrigin(request.headers.origin, request.headers.host, cors))
+          return HttpServerResponse.empty({ status: 403 })
+
+        // Handler resources belong to an RPC request, not the WebSocket upgrade.
+        const context = Context.merge(services, yield* Effect.context<never>()).pipe(Context.omit(Scope.Scope))
+        const implementations = Object.fromEntries(
+          entries.flatMap((entry) => {
+            const rpc = group.requests.get(entry.endpoint.identifier)
+            if (!rpc) return []
+            const streaming = RpcSchema.isStreamSchema(rpc.successSchema)
+            const invoke = (input: Input, options: { readonly headers: Headers.Headers }) => {
+              const url = new URL(request.url, "http://localhost")
+              url.pathname = entry.endpoint.path
+              const location = input.location ?? input.query?.location
+              if (location) {
+                if (location.directory) url.searchParams.set("location[directory]", location.directory)
+                if (location.workspace) url.searchParams.set("location[workspace]", location.workspace)
+                else url.searchParams.delete("location[workspace]")
+              }
+              // RPC metadata carries contextual headers (global forms and PTY tickets);
+              // the authenticated upgrade remains authoritative for its own headers.
+              const headers = Headers.merge(
+                Headers.merge(Headers.fromInput(input.headers), options.headers),
+                request.headers,
+              )
+              const current = request.modify({
+                url: `${url.pathname}${url.search}`,
+                headers: location ? Headers.remove(headers, "x-opencode-workspace") : headers,
+              })
+              const run = Effect.gen(function* () {
+                if (rpc._tag === "event.subscribe") {
+                  const live = yield* feed.subscribeEvents
+                  return Stream.make({ id: Event.ID.create(), type: "server.connected" as const, data: {} }).pipe(
+                    Stream.concat(live),
+                    Stream.orDie,
+                  )
+                }
+                if (rpc._tag === "fs.read") return yield* readFile(RelativePath.make(input.params?.path ?? ""))
+                const result = yield* entry.handler({
+                  ...input,
+                  request: current,
+                  group: entry.definition,
+                  endpoint: entry.endpoint,
+                })
+                // Streams are consumed after middleware returns; retain the location
+                // services selected for this call, not the connection's last location.
+                if (Stream.isStream(result)) return Stream.provideContext(result, yield* Effect.context<never>())
+                return result
+              }) as Effect.Effect<unknown, unknown, unknown>
+              const wrapped = Array.from(entry.endpoint.middlewares).reduce((effect, key) => {
+                // Authentication belongs to the upgrade, not client-supplied frame headers.
+                if (key.key === Authorization.key) return effect
+                const middleware = Context.getUnsafe(context, key) as Middleware
+                if (typeof middleware !== "function") throw new Error(`Unsupported RPC middleware: ${key.key}`)
+                return middleware(effect, { group: entry.definition, endpoint: entry.endpoint })
+              }, run)
+              const effect = (entry.uninterruptible ? Effect.uninterruptible(wrapped) : wrapped).pipe(
+                Effect.provideService(HttpServerRequest.HttpServerRequest, current),
+                Effect.provideService(HttpRouter.RouteContext, {
+                  params: input.params ?? {},
+                  route: HttpRouter.route(
+                    entry.endpoint.method,
+                    entry.endpoint.path as HttpRouter.PathInput,
+                    HttpServerResponse.empty(),
+                  ),
+                }),
+                Effect.provideContext(context as Context.Context<unknown>),
+              )
+              return streaming
+                ? Stream.unwrap(effect as Effect.Effect<Stream.Stream<unknown, unknown>, unknown>)
+                : effect
+            }
+            return [[rpc._tag, invoke]]
+          }),
+        )
+        const runtime = group as unknown as RpcGroup.RpcGroup<Rpc.Any>
+        const implementation = yield* runtime.toHandlers(implementations as unknown as RpcGroup.HandlersFrom<Rpc.Any>)
+        const websocket = yield* RpcServer.toHttpEffectWebsocket(runtime, { disableFatalDefects: true }).pipe(
+          Effect.provideContext(implementation),
+          Effect.provide(RpcSerialization.layerJson),
+        )
+        return yield* websocket
+      }),
+    )
+  }),
+)

packages/server/test/event-feed.test.ts

@@ -39,6 +39,40 @@ function makeSource() {
 }
 
 describe("EventFeed", () => {
+  it.effect("delivers typed RPC events without SSE encoding and filters internal events", () =>
+    Effect.gen(function* () {
+      const source = makeSource()
+      const feed = yield* EventFeed.make(source.observe, {
+        capacity: 1,
+        encode: () => {
+          throw new Error("RPC-only subscribers must not encode SSE")
+        },
+      })
+      const stream = yield* feed.subscribeEvents
+      yield* source.publish(internal("one"))
+      yield* source.publish(internal("two"))
+      const payload = event("rpc")
+      yield* source.publish(payload)
+      expect(yield* stream.pipe(Stream.take(1), Stream.runCollect)).toEqual([payload])
+    }),
+  )
+
+  it.effect("bounds RPC subscriber lag independently of HTTP subscribers", () =>
+    Effect.gen(function* () {
+      const source = makeSource()
+      const feed = yield* EventFeed.make(source.observe, { capacity: 1 })
+      const slow = yield* feed.subscribeEvents
+      yield* source.publish(event("one"))
+      const http = yield* feed.subscribe
+      const payload = event("two")
+      yield* source.publish(payload)
+      const exit = yield* slow.pipe(Stream.runCollect, Effect.exit)
+      expect(Exit.isFailure(exit)).toBeTrue()
+      expect(Option.getOrUndefined(Exit.findErrorOption(exit))).toBeInstanceOf(EventFeed.SubscriberOverflowError)
+      expect(yield* http.pipe(Stream.take(1), Stream.runCollect)).toEqual([EventFeed.frame(payload)])
+    }),
+  )
+
   test("preserves the public SSE frame encoding", () => {
     const payload = event("wire")
     expect(EventFeed.frame(payload)).toBe(`data: ${JSON.stringify(payload)}\n\n`)

packages/server/test/rpc-desktop.test.ts

@@ -0,0 +1,70 @@
+import { expect } from "bun:test"
+import { Effect } from "effect"
+import { HttpServer } from "effect/unstable/http"
+import { OpenCodeRpc } from "../../client/src/promise/rpc"
+import { it } from "../../core/test/lib/effect"
+import { tmpdir } from "../../core/test/fixture/tmpdir"
+import { ServerProcess } from "../src/process"
+
+it.live(
+  "serves the desktop Promise client over native RPC, including location and ticket headers",
+  () =>
+    Effect.gen(function* () {
+      const tmp = yield* Effect.acquireRelease(
+        Effect.promise(() => tmpdir("desktop-rpc-")),
+        (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
+      )
+      const server = yield* ServerProcess.start<never, never>({
+        hostname: "127.0.0.1",
+        port: 0,
+        password: "test-password",
+        database: { path: ":memory:" },
+        config: { directory: `${tmp.path}/config`, project: false },
+        models: { fetch: false },
+        fs: { filewatcher: false, fff: false },
+      })
+      const api = yield* Effect.acquireRelease(
+        Effect.sync(() =>
+          OpenCodeRpc.make({
+            baseUrl: HttpServer.formatAddress(server.address),
+            headers: { authorization: `Basic ${btoa("opencode:test-password")}` },
+          }),
+        ),
+        (api) => Effect.promise(() => api.dispose()),
+      )
+      const headers = { "x-opencode-directory": encodeURIComponent(tmp.path) }
+      expect((yield* Effect.promise(() => api.location.get(undefined, { headers }))).directory).toBe(tmp.path)
+
+      const events = api.event.subscribe()[Symbol.asyncIterator]()
+      expect(yield* Effect.promise(() => events.next())).toMatchObject({ value: { type: "server.connected" } })
+      const created = yield* Effect.promise(() =>
+        api.session.create({ title: "Desktop RPC", location: { directory: tmp.path } }),
+      )
+      expect(typeof created.time.created).toBe("number")
+      expect((yield* Effect.promise(() => api.session.list({ limit: 1 }))).data[0]?.id).toBe(created.id)
+      const event = yield* Effect.promise(async () => {
+        for (let item = await events.next(); !item.done; item = await events.next()) {
+          if (item.value.type === "session.created") return item.value
+        }
+      })
+      expect(event).toMatchObject({ type: "session.created", data: { sessionID: created.id } })
+      yield* Effect.promise(() => events.return!())
+      expect(
+        yield* Effect.promise(() => api.session.rename({ sessionID: created.id, title: "Renamed" })),
+      ).toBeUndefined()
+      expect((yield* Effect.promise(() => api.session.get({ sessionID: created.id }))).title).toBe("Renamed")
+
+      yield* Effect.promise(() => Bun.write(`${tmp.path}/file #%.txt`, "Desktop bytes"))
+      expect(
+        yield* Effect.promise(() => api.file.read({ path: "file #%.txt", location: { directory: tmp.path } })),
+      ).toEqual(new TextEncoder().encode("Desktop bytes"))
+      const ticket = yield* Effect.promise(() =>
+        api.pty.connect
+          .token({ ptyID: "pty_missing" }, { headers: { ...headers, "x-opencode-ticket": "1" } })
+          .catch((error) => error),
+      )
+      expect(ticket).toMatchObject({ _tag: "PtyNotFoundError" })
+      expect((yield* Effect.promise(() => api.health.get())).healthy).toBe(true)
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)

packages/server/test/rpc.test.ts

@@ -0,0 +1,331 @@
+import { expect } from "bun:test"
+import fs from "node:fs/promises"
+import path from "node:path"
+import { NodeSocket } from "@effect/platform-node"
+import { OpenCodeRpc } from "@opencode-ai/protocol/rpc"
+import { SessionNotFoundError } from "@opencode-ai/protocol/errors"
+import { AbsolutePath } from "@opencode-ai/schema/schema"
+import { Pty } from "@opencode-ai/schema/pty"
+import { Session } from "@opencode-ai/schema/session"
+import { Cause, Deferred, Effect, Exit, Fiber, Layer, Queue, Schema, Scope, Stream } from "effect"
+import { HttpServer, HttpServerRequest } from "effect/unstable/http"
+import { Rpc, RpcClient, RpcGroup, RpcSerialization } from "effect/unstable/rpc"
+import { Socket } from "effect/unstable/socket"
+import { tmpdir } from "../../core/test/fixture/tmpdir"
+import { it } from "../../core/test/lib/effect"
+import { ServerProcess } from "../src/process"
+
+const authorization = `Basic ${btoa("opencode:secret")}`
+
+const fixture = Effect.fn(function* <R extends Rpc.Any>(
+  group: RpcGroup.RpcGroup<R>,
+  transform?: ServerProcess.Transform,
+) {
+  const tmp = yield* Effect.acquireRelease(
+    Effect.promise(() => tmpdir("opencode-rpc-")),
+    (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
+  )
+  const global = path.join(tmp.path, "config")
+  const directories = [path.join(tmp.path, "first"), path.join(tmp.path, "second")]
+  yield* Effect.promise(() => Promise.all([global, ...directories].map((dir) => fs.mkdir(dir))))
+  const server = yield* ServerProcess.start<never, never>(
+    {
+      hostname: "127.0.0.1",
+      port: 0,
+      password: "secret",
+      app: { version: "test-version" },
+      database: { path: ":memory:" },
+      events: { persist: true },
+      config: { directory: global, project: false },
+      models: { fetch: false },
+      fs: { filewatcher: false, fff: false },
+    },
+    undefined,
+    transform,
+  )
+  const url = HttpServer.formatAddress(server.address)
+  const sockets: WebSocket[] = []
+  const protocol = yield* Layer.build(
+    RpcClient.layerProtocolSocket({ retryTransientErrors: false }).pipe(
+      Layer.provide(RpcSerialization.layerJson),
+      Layer.provide(
+        Socket.layerWebSocket(new URL("/api/rpc", url).href.replace("http:", "ws:")).pipe(
+          Layer.provide(
+            Layer.succeed(Socket.WebSocketConstructor, (url) => {
+              const socket = new NodeSocket.NodeWS.WebSocket(url, {
+                headers: { authorization, origin: "http://localhost:3000" },
+              }) as unknown as WebSocket
+              sockets.push(socket)
+              return socket
+            }),
+          ),
+        ),
+      ),
+    ),
+  )
+  const rpc = yield* RpcClient.make(group).pipe(Effect.provideContext(protocol))
+  const request = (pathname: string, init?: RequestInit) =>
+    Effect.promise(() =>
+      fetch(new URL(pathname, url), {
+        ...init,
+        headers: { authorization, "content-type": "application/json", ...init?.headers },
+      }),
+    )
+  return {
+    rpc,
+    request,
+    url,
+    sockets,
+    first: { directory: AbsolutePath.make(directories[0]!) },
+    second: { directory: AbsolutePath.make(directories[1]!) },
+  }
+})
+
+it.live(
+  "multiplexes concurrent RPC calls and shares session state with HTTP",
+  () =>
+    Effect.gen(function* () {
+      const { rpc, request, first, sockets } = yield* fixture(OpenCodeRpc.Group)
+      const sessions = yield* Effect.all(
+        Array.from({ length: 8 }, (_, index) =>
+          rpc["session.create"]({ payload: { title: `parallel-${index}`, location: first } }),
+        ),
+        { concurrency: "unbounded" },
+      )
+      expect(new Set(sessions.map((session) => session.data.id)).size).toBe(8)
+      expect(sessions.map((session) => session.data.title)).toEqual(
+        Array.from({ length: 8 }, (_, index) => `parallel-${index}`),
+      )
+      const sessionID = sessions[0]!.data.id
+      const http = yield* request(`/api/session/${sessionID}`)
+      expect(http.status).toBe(200)
+      expect(
+        Schema.decodeUnknownSync(Schema.Struct({ data: Session.Info }))(yield* Effect.promise(() => http.json())),
+      ).toEqual(yield* rpc["session.get"]({ params: { sessionID } }))
+
+      expect(
+        yield* rpc["session.rename"]({ params: { sessionID }, payload: { title: "renamed by RPC" } }),
+      ).toBeUndefined()
+      const renamed = yield* request(`/api/session/${sessionID}`)
+      expect(yield* Effect.promise(() => renamed.json())).toMatchObject({ data: { title: "renamed by RPC" } })
+
+      expect(
+        (yield* request(`/api/session/${sessionID}/rename`, {
+          method: "POST",
+          body: JSON.stringify({ title: "renamed by HTTP" }),
+        })).status,
+      ).toBe(204)
+      expect((yield* rpc["session.get"]({ params: { sessionID } })).data.title).toBe("renamed by HTTP")
+      expect(yield* rpc["session.remove"]({ params: { sessionID } })).toBeUndefined()
+      const missing = yield* rpc["session.get"]({ params: { sessionID } }).pipe(
+        Effect.catchTag("SessionNotFoundError", Effect.succeed),
+      )
+      expect(missing).toBeInstanceOf(SessionNotFoundError)
+      expect(missing).toMatchObject({ _tag: "SessionNotFoundError", sessionID })
+      const absent = yield* request(`/api/session/${sessionID}`)
+      expect(absent.status).toBe(404)
+      expect(yield* Effect.promise(() => absent.json())).toMatchObject({ _tag: "SessionNotFoundError", sessionID })
+      expect(sockets).toHaveLength(1)
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)
+
+it.live(
+  "reads binary files and MIME types at two locations over one RPC connection",
+  () =>
+    Effect.gen(function* () {
+      const { rpc, request, first, second, sockets } = yield* fixture(OpenCodeRpc.Group)
+      const filename = "space # percent% question? plus+.png"
+      const firstBytes = new Uint8Array([137, 80, 78, 71, 13, 10, 26, 10, 0, 255, 128])
+      const secondBytes = new Uint8Array([137, 80, 78, 71, 13, 10, 26, 10, 127, 1, 254])
+      yield* Effect.promise(() =>
+        Promise.all([
+          Bun.write(path.join(first.directory, filename), firstBytes),
+          Bun.write(path.join(second.directory, filename), secondBytes),
+        ]),
+      )
+      const files = yield* Effect.all(
+        [first, second].map((location) => rpc["fs.read"]({ params: { path: filename }, query: {}, location })),
+        { concurrency: "unbounded" },
+      )
+      expect(files.map((file) => file.content)).toEqual([firstBytes, secondBytes])
+      expect(files.map((file) => file.mime)).toEqual(["image/png", "image/png"])
+      const query = new URLSearchParams({ "location[directory]": first.directory })
+      const http = yield* request(`/api/fs/read/${encodeURIComponent(filename)}?${query}`)
+      expect(http.status).toBe(200)
+      expect(http.headers.get("content-type")).toBe(files[0]!.mime)
+      expect(Array.from(new Uint8Array(yield* Effect.promise(() => http.arrayBuffer())))).toEqual(
+        Array.from(files[0]!.content),
+      )
+      const lists = yield* Effect.all(
+        [first, second].map((location) => rpc["fs.list"]({ query: {}, location })),
+        { concurrency: "unbounded" },
+      )
+      expect(lists.map((list) => list.location.directory)).toEqual([first.directory, second.directory])
+      expect(sockets).toHaveLength(1)
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)
+
+it.live(
+  "streams the connected event and later domain events alongside unary RPC calls",
+  () =>
+    Effect.gen(function* () {
+      const { rpc, first, sockets } = yield* fixture(OpenCodeRpc.Group)
+      yield* Effect.gen(function* () {
+        const events = yield* rpc["event.subscribe"]({}, { asQueue: true })
+        expect(yield* Queue.take(events)).toMatchObject({ type: "server.connected", data: {} })
+        const created = yield* rpc["session.create"]({ payload: { title: "observed", location: first } })
+        const received = yield* Stream.fromQueue(events).pipe(
+          Stream.filter((event) => event.type === "session.created"),
+          Stream.take(1),
+          Stream.runCollect,
+        )
+        expect(received).toMatchObject([{ type: "session.created", data: { sessionID: created.data.id } }])
+      }).pipe(Effect.scoped)
+      expect(yield* rpc["health.get"]({})).toMatchObject({ healthy: true, version: "test-version" })
+      expect(sockets).toHaveLength(1)
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)
+
+it.live(
+  "cancels event subscriptions without accumulating upgrade finalizers",
+  () =>
+    Effect.gen(function* () {
+      const captured = yield* Deferred.make<Scope.Scope>()
+      const { rpc, sockets } = yield* fixture(OpenCodeRpc.Group, (app) =>
+        Effect.gen(function* () {
+          const request = yield* HttpServerRequest.HttpServerRequest
+          if (request.url === "/api/rpc") {
+            const scope = yield* Scope.Scope
+            yield* Deferred.succeed(captured, scope)
+          }
+          return yield* app
+        }),
+      )
+      yield* rpc["health.get"]({})
+      const scope = yield* Deferred.await(captured)
+      const finalizerCount = () => {
+        const state = scope.state
+        if (state._tag !== "Open") throw new Error("WebSocket upgrade scope must remain open")
+        return state.finalizers.size
+      }
+      const baseline = finalizerCount()
+      for (let index = 0; index < 5; index++) {
+        yield* Effect.gen(function* () {
+          const events = yield* rpc["event.subscribe"]({}, { asQueue: true })
+          expect(yield* Queue.take(events)).toMatchObject({ type: "server.connected" })
+          // The subscription belongs to its RPC scope even while it is active.
+          expect(finalizerCount()).toBe(baseline)
+        }).pipe(Effect.scoped)
+        expect(yield* rpc["health.get"]({})).toMatchObject({ healthy: true, version: "test-version" })
+        expect(finalizerCount()).toBe(baseline)
+      }
+      expect(sockets).toHaveLength(1)
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)
+
+it.live(
+  "replays session logs and cancels a live stream without closing the RPC connection",
+  () =>
+    Effect.gen(function* () {
+      const { rpc, first, sockets } = yield* fixture(OpenCodeRpc.Group)
+      const session = yield* rpc["session.create"]({ payload: { title: "before replay", location: first } })
+      const params = { sessionID: session.data.id }
+      yield* rpc["session.rename"]({ params, payload: { title: "in replay" } })
+      const replay = yield* rpc["session.log"]({ params, query: { follow: false } }).pipe(Stream.runCollect)
+      expect(replay.map((event) => event.type)).toEqual(["session.created", "session.renamed", "log.synced"])
+      expect(replay[1]).toMatchObject({ data: { sessionID: params.sessionID, title: "in replay" } })
+      const watermark = replay.find((event) => event.type === "log.synced")!
+      expect(watermark).toMatchObject({ type: "log.synced", aggregateID: params.sessionID })
+
+      const synced = yield* Deferred.make<void>()
+      const renamed = yield* Deferred.make<void>()
+      const follow = yield* rpc["session.log"]({ params, query: { after: watermark.seq, follow: true } }).pipe(
+        Stream.runForEach((event) =>
+          event.type === "log.synced"
+            ? Deferred.succeed(synced, undefined)
+            : event.type === "session.renamed" && event.data.title === "after replay"
+              ? Deferred.succeed(renamed, undefined)
+              : Effect.void,
+        ),
+        Effect.forkScoped,
+      )
+      yield* Deferred.await(synced)
+      yield* rpc["session.rename"]({ params, payload: { title: "after replay" } })
+      yield* Deferred.await(renamed)
+      yield* Fiber.interrupt(follow)
+      expect((yield* rpc["session.get"]({ params })).data.title).toBe("after replay")
+      expect(yield* rpc["health.get"]({})).toMatchObject({ healthy: true })
+      expect(sockets).toHaveLength(1)
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)
+
+it.live(
+  "validates malformed RPC payloads on the server without poisoning the connection",
+  () =>
+    Effect.gen(function* () {
+      // The permissive client codec sends invalid data instead of rejecting it before transport.
+      const { rpc, request, first, sockets } = yield* fixture(
+        RpcGroup.make(
+          Rpc.make("session.create", { payload: Schema.Unknown, success: Schema.Unknown, error: Schema.Unknown }),
+          Rpc.make("health.get", { payload: Schema.Unknown, success: Schema.Unknown, error: Schema.Unknown }),
+        ),
+      )
+      const invalid = yield* rpc["session.create"]({ payload: { title: 42, location: first } }).pipe(Effect.exit)
+      expect(Exit.isFailure(invalid)).toBe(true)
+      if (Exit.isFailure(invalid)) expect(Cause.pretty(invalid.cause)).toContain("title")
+      const http = yield* request("/api/session", {
+        method: "POST",
+        body: JSON.stringify({ title: 42, location: first }),
+      })
+      expect(http.status).toBe(400)
+      const list = yield* request("/api/session")
+      expect(yield* Effect.promise(() => list.json())).toMatchObject({ data: [] })
+      expect(yield* rpc["health.get"]({})).toMatchObject({ healthy: true })
+      expect(sockets).toHaveLength(1)
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)
+
+it.live(
+  "rejects missing credentials and untrusted browser origins before upgrading RPC",
+  () =>
+    Effect.gen(function* () {
+      const { rpc, url } = yield* fixture(OpenCodeRpc.Group)
+      const denied = yield* Effect.all(
+        [
+          new Headers(),
+          new Headers({ authorization: `Basic ${btoa("opencode:wrong")}` }),
+          new Headers({ authorization, origin: "https://untrusted.example" }),
+          new Headers({ authorization, origin: "null" }),
+        ].map((headers) => Effect.promise(() => fetch(new URL("/api/rpc", url), { headers }))),
+        { concurrency: "unbounded" },
+      )
+      expect(denied.map((response) => response.status)).toEqual([401, 401, 403, 403])
+      expect(denied.every((response) => !response.headers.has("sec-websocket-accept"))).toBe(true)
+      expect(yield* rpc["health.get"]({})).toMatchObject({ healthy: true })
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)
+
+it.live(
+  "forwards declared PTY ticket headers to the shared business handler",
+  () =>
+    Effect.gen(function* () {
+      const { rpc, first } = yield* fixture(OpenCodeRpc.Group)
+      const input = { params: { ptyID: Pty.ID.make("pty_missing") }, query: {}, location: first }
+      const denied = yield* rpc["pty.connectToken"]({ ...input, headers: {} }).pipe(
+        Effect.catchTag("ForbiddenError", Effect.succeed),
+      )
+      expect(denied).toMatchObject({ _tag: "ForbiddenError" })
+      const missing = yield* rpc["pty.connectToken"]({ ...input, headers: { "x-opencode-ticket": "1" } }).pipe(
+        Effect.catchTag("PtyNotFoundError", Effect.succeed),
+      )
+      expect(missing).toMatchObject({ _tag: "PtyNotFoundError", ptyID: input.params.ptyID })
+    }).pipe(Effect.timeout("20 seconds")),
+  30_000,
+)