feat(server): expose the HTTP API over WebSocket RPC

packages/client/README.md

@@ -6,6 +6,7 @@ Private generation target for clients derived directly from OpenCode's authorita
 
 - `@opencode-ai/client`: zero-Effect Promise client using `fetch`.
 - `@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 +26,36 @@ yield *
   })
 yield * client.sessions.prompt({ sessionID, prompt: Prompt.make({ text: "Hello" }) })
 ```
+
+## WebSocket RPC
+
+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; the apps still use HTTP until explicitly switched.

packages/client/package.json

@@ -23,6 +23,7 @@
     "./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/test/import-boundaries.test.ts

@@ -28,6 +28,12 @@ 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 promiseService = await bundleInputs("@opencode-ai/client/service", "bun")
 
     expect(within(promiseService, effect)).toEqual([])

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/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,161 @@
+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) => {
+              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]")
+              }
+              const headers = Headers.merge(Headers.fromInput(input.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.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,
+)