fix(core): fail rpc.internal as a typed error on every path

packages/core/src/rpc.ts

@@ -117,11 +117,29 @@ const layer = Layer.effect(
       const parsed = yield* parse(method.input, input).pipe(
         Effect.mapError((error) => failure("rpc.invalid_input", errorMessage(error, "Invalid RPC input"))),
       )
+      // Handler defects and undeclared errors become the same typed rpc.internal failure the HTTP handler
+      // exposes, so an in-process plugin caller can recover exactly like a remote one.
+      const internal = (error: unknown, message = "RPC call failed") =>
+        Effect.logError("rpc handler failed", { rpc: rpcID, method: name, error }).pipe(
+          Effect.andThen(Effect.fail(failure("rpc.internal", message))),
+        )
       const result = yield* Effect.suspend(() => {
         // The heterogeneous registry erases handlers after their selected schema validates input.
         const execution: Effect.Effect<unknown, unknown> = Reflect.apply(handler, undefined, [parsed, callContext])
         return execution
-      }).pipe(Effect.catch((error) => encodeError(method, error)))
+      }).pipe(
+        Effect.catch((error) => {
+          if (!(error instanceof DeclaredError)) return internal(error)
+          const declared = method.errors && Object.hasOwn(method.errors, error.type) && method.errors[error.type]
+          if (!declared) return internal(error, `Undeclared RPC error: ${error.type}`)
+          return encode(declared, error.data).pipe(
+            Effect.catch((cause) => Effect.die(cause)),
+            Effect.flatMap((data) => Effect.fail(failure(error.type, error.message, data))),
+          )
+        }),
+        // Also covers declared error data that fails its own schema: a handler bug, not a caller error.
+        Effect.catchDefect((defect) => internal(defect)),
+      )
       return yield* encode(method.output, result).pipe(
         Effect.mapError((error) => failure("rpc.invalid_output", errorMessage(error, "Invalid RPC output"))),
       )
@@ -215,17 +233,6 @@ function encode(schema: Tool.ValueSchema, value: unknown): Effect.Effect<unknown
   return Schema.isSchema(schema) ? Schema.encodeUnknownEffect(schema)(value) : parse(schema, value)
 }
 
-function encodeError(method: Rpc.Method, error: unknown): Effect.Effect<never, Rpc.Failure> {
-  if (!(error instanceof DeclaredError)) return Effect.die(error)
-  if (!method.errors || !Object.hasOwn(method.errors, error.type)) {
-    return Effect.die(new Error(`Undeclared RPC error: ${error.type}`))
-  }
-  return encode(method.errors[error.type], error.data).pipe(
-    Effect.catch((cause) => Effect.die(cause)),
-    Effect.flatMap((data) => Effect.fail(failure(error.type, error.message, data))),
-  )
-}
-
 function decodeError(method: Rpc.Method, error: Rpc.Failure): Effect.Effect<never, Rpc.Failure> {
   if (!method.errors || !Object.hasOwn(method.errors, error.type)) return Effect.fail(error)
   return read(method.errors[error.type], error.data).pipe(

packages/core/test/plugin/rpc-effect.test.ts

@@ -0,0 +1,106 @@
+import { expect } from "bun:test"
+import { Plugin } from "@opencode-ai/core/plugin"
+import { Rpc } from "@opencode-ai/core/rpc"
+import { Bus } from "@opencode-ai/core/bus"
+import { Location } from "@opencode-ai/core/location"
+import { PluginTestLayer } from "./fixture"
+import { Effect, Exit, Schema } from "effect"
+import { testEffect } from "../lib/effect"
+
+const it = testEffect(PluginTestLayer)
+const Echo = Rpc.define({
+  id: "shared-echo",
+  methods: {
+    echo: { input: Schema.String, output: Schema.String },
+    fail: {
+      input: Schema.String,
+      output: Schema.String,
+      errors: { missing: Schema.Struct({ attempts: Schema.FiniteFromString }) },
+    },
+  },
+  events: { updated: { schema: Schema.Struct({ text: Schema.String }) } },
+})
+
+it.effect("Effect plugins register, call, and publish RPCs independently of plugin identity", () =>
+  Effect.gen(function* () {
+    const plugins = yield* Plugin.Service
+    const rpc = yield* Rpc.Service
+    const bus = yield* Bus.Service
+    const location = yield* Location.Service
+    const events: string[] = []
+    const unsubscribe = yield* bus.listen((event) =>
+      Effect.sync(() => {
+        if (event.type !== "rpc.shared-echo.updated") return
+        expect(event.location).toEqual({ directory: location.directory })
+        if (typeof event.data === "object" && event.data && "text" in event.data && typeof event.data.text === "string")
+          events.push(event.data.text)
+      }),
+    )
+    yield* plugins.activate([
+      {
+        id: "implementer",
+        revision: "1",
+        effect: (ctx) =>
+          Effect.gen(function* () {
+            const registration = yield* ctx.rpc.register(Echo, {
+              echo: (value) => Effect.succeed(`${value}!`),
+              fail: (value, context) => Effect.fail(context.error("missing", "Missing", { attempts: Number(value) })),
+            })
+            yield* registration.events.emit("updated", { text: "ready" })
+          }).pipe(Effect.orDie),
+      },
+      {
+        id: "consumer",
+        revision: "1",
+        effect: (ctx) =>
+          Effect.gen(function* () {
+            expect(yield* ctx.rpc(Echo).echo("hello")).toBe("hello!")
+            expect(yield* ctx.rpc(Echo).fail("2").pipe(Effect.flip)).toEqual({
+              type: "missing",
+              message: "Missing",
+              data: { attempts: 2 },
+            })
+          }).pipe(Effect.orDie),
+      },
+    ])
+    expect(events).toEqual(["ready"])
+    expect(yield* rpc.client(Echo).echo("hello")).toBe("hello!")
+    yield* plugins.activate([])
+    expect(Exit.isFailure(yield* rpc.client(Echo).echo("hello").pipe(Effect.exit))).toBe(true)
+    yield* unsubscribe
+  }),
+)
+
+it.effect("failed plugin setup removes RPC overrides and restores the previous implementation", () =>
+  Effect.gen(function* () {
+    const plugins = yield* Plugin.Service
+    const rpc = yield* Rpc.Service
+    yield* plugins.activate([
+      {
+        id: "implementer",
+        revision: "1",
+        effect: (ctx) =>
+          ctx.rpc
+            .register(Echo, {
+              echo: () => Effect.succeed("original"),
+              fail: (_input, context) => Effect.fail(context.error("missing", "Missing", { attempts: 1 })),
+            })
+            .pipe(Effect.asVoid, Effect.orDie),
+      },
+    ])
+    yield* plugins.activate([
+      {
+        id: "implementer",
+        revision: "2",
+        effect: (ctx) =>
+          ctx.rpc
+            .register(Echo, {
+              echo: () => Effect.succeed("replacement"),
+              fail: (_input, context) => Effect.fail(context.error("missing", "Missing", { attempts: 1 })),
+            })
+            .pipe(Effect.andThen(Effect.die(new Error("setup failed"))), Effect.orDie),
+      },
+    ])
+    expect(yield* rpc.client(Echo).echo("hello")).toBe("original")
+  }),
+)

packages/core/test/plugin/rpc-promise.test.ts

@@ -0,0 +1,290 @@
+import { describe, expect } from "bun:test"
+import { Plugin } from "@opencode-ai/core/plugin"
+import { PluginPromise } from "@opencode-ai/core/plugin/promise"
+import { define } from "@opencode-ai/plugin/promise/plugin"
+import type { RpcEventPayload } from "@opencode-ai/plugin/promise/rpc"
+import { Rpc } from "@opencode-ai/plugin/rpc"
+import { Effect, Logger } from "effect"
+import { z } from "zod"
+import { testEffect } from "../lib/effect"
+import { PluginTestLayer } from "./fixture"
+
+const it = testEffect(PluginTestLayer)
+
+describe("Promise plugin RPC", () => {
+  it.live("adapts calls, schema transforms, failures, and registration disposal", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      const service = Rpc.define({
+        id: "promise-rpc-calls",
+        methods: {
+          standard: { input: z.string().transform(Number), output: z.number().transform(String) },
+          ping: { input: z.undefined(), output: z.null() },
+          errorShapedOutput: {
+            input: z.undefined(),
+            output: z.object({ type: z.string(), message: z.string(), data: z.object({ value: z.number() }) }),
+          },
+          returned: {
+            input: z.undefined(),
+            output: z.null(),
+            errors: { rejected: z.object({ attempts: z.string().transform(Number) }) },
+          },
+          thrown: {
+            input: z.undefined(),
+            output: z.null(),
+            errors: { rejected: z.object({ attempts: z.string().transform(Number) }) },
+          },
+          defect: { input: z.undefined(), output: z.null() },
+        },
+        events: {},
+      })
+      const adapted = PluginPromise.fromPromise(
+        define({
+          id: "promise-rpc-calls-plugin",
+          setup: async (ctx) => {
+            const registration = await ctx.rpc.register(service, {
+              standard: async (input) => {
+                expect(input).toBe(42)
+                return input + 1
+              },
+              ping: async () => null,
+              errorShapedOutput: async () => ({ type: "ordinary", message: "Success", data: { value: 1 } }),
+              returned: async (_input, context) => context.error("rejected", "returned failure", { attempts: "1" }),
+              thrown: async (_input, context) => {
+                throw context.error("rejected", "thrown failure", { attempts: "2" })
+              },
+              defect: async () => {
+                throw new Error("handler defect")
+              },
+            })
+            const client = ctx.rpc(service)
+            expect(await client.standard("42")).toBe("43")
+            expect(await client.ping()).toBeNull()
+            expect(await client.errorShapedOutput()).toEqual({
+              type: "ordinary",
+              message: "Success",
+              data: { value: 1 },
+            })
+            await expect(client.returned()).rejects.toEqual({
+              type: "rejected",
+              message: "returned failure",
+              data: { attempts: 1 },
+            })
+            await expect(client.thrown()).rejects.toEqual({
+              type: "rejected",
+              message: "thrown failure",
+              data: { attempts: 2 },
+            })
+            await expect(client.defect()).rejects.toEqual({ type: "rpc.internal", message: "RPC call failed" })
+            await registration.dispose()
+            await registration.dispose()
+            await expect(client.ping()).rejects.toBeDefined()
+          },
+        }),
+      )
+
+      yield* plugins.activate([{ ...adapted, revision: "1" }])
+      expect(yield* plugins.list()).toMatchObject([{ id: adapted.id, state: { status: "active" } }])
+    }),
+  )
+
+  it.live("cancels only the selected call and passes its AbortSignal to Promise handlers", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      const service = Rpc.define({
+        id: "promise-rpc-cancel",
+        methods: { wait: { input: z.string(), output: z.string() } },
+        events: {},
+      })
+      const adapted = PluginPromise.fromPromise(
+        define({
+          id: "promise-rpc-cancel-plugin",
+          setup: async (ctx) => {
+            const started = Promise.withResolvers<void>()
+            const cancelled = Promise.withResolvers<void>()
+            const signals = new Map<string, AbortSignal>()
+            await ctx.rpc.register(service, {
+              wait: async (input, call) => {
+                signals.set(input, call.signal)
+                if (input === "complete") return input
+                started.resolve()
+                await new Promise<void>((resolve) => {
+                  call.signal.addEventListener(
+                    "abort",
+                    () => {
+                      cancelled.resolve()
+                      resolve()
+                    },
+                    { once: true },
+                  )
+                })
+                return input
+              },
+            })
+            const client = ctx.rpc(service)
+            const controller = new AbortController()
+            const pending = client.wait("cancel", { signal: controller.signal })
+            const rejected = pending.then(
+              () => false,
+              () => true,
+            )
+            await started.promise
+            expect(await client.wait("complete")).toBe("complete")
+            controller.abort()
+            expect(await rejected).toBe(true)
+            await cancelled.promise
+            expect(signals.get("cancel")?.aborted).toBe(true)
+            expect(signals.get("complete")?.aborted).toBe(false)
+            expect(await client.wait("complete")).toBe("complete")
+          },
+        }),
+      )
+
+      yield* plugins.activate([{ ...adapted, revision: "1" }])
+      expect(yield* plugins.list()).toMatchObject([{ id: adapted.id, state: { status: "active" } }])
+    }),
+  )
+
+  it.live("awaits async callbacks and logs failures without stopping other plugin listeners", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      const service = Rpc.define({
+        id: "promise-rpc-async-listeners",
+        methods: {},
+        events: { updated: { schema: z.object({ value: z.number() }) } },
+      })
+      const error = new Error("Expected async plugin callback failure")
+      const reported = Promise.withResolvers<void>()
+      const logger = Logger.make((entry) => {
+        if (Array.isArray(entry.message) && entry.message.includes(error)) reported.resolve()
+      })
+      const adapted = PluginPromise.fromPromise(
+        define({
+          id: "promise-rpc-async-listeners-plugin",
+          setup: async (ctx) => {
+            const registration = await ctx.rpc.register(service, {})
+            const client = ctx.rpc(service)
+            const started = Promise.withResolvers<void>()
+            const release = Promise.withResolvers<void>()
+            const second = Promise.withResolvers<void>()
+            const third = Promise.withResolvers<void>()
+            const failed: number[] = []
+            const healthy: number[] = []
+            client.events.on("updated", async (event) => {
+              failed.push(event.data.value)
+              started.resolve()
+              await release.promise
+              throw error
+            })
+            client.events.on("updated", (event) => {
+              healthy.push(event.data.value)
+              if (event.data.value === 2) second.resolve()
+              if (event.data.value === 3) third.resolve()
+            })
+            await registration.events.emit("updated", { value: 1 })
+            await started.promise
+            await registration.events.emit("updated", { value: 2 })
+            await second.promise
+            expect(failed).toEqual([1])
+            release.resolve()
+            await reported.promise
+            await registration.events.emit("updated", { value: 3 })
+            await third.promise
+            expect(failed).toEqual([1])
+            expect(healthy).toEqual([1, 2, 3])
+          },
+        }),
+      )
+      yield* plugins
+        .activate([{ ...adapted, revision: "1" }])
+        .pipe(Effect.provideService(Logger.CurrentLoggers, new Set([logger])))
+      expect(yield* plugins.list()).toMatchObject([{ id: adapted.id, state: { status: "active" } }])
+      yield* plugins.activate([])
+    }),
+  )
+
+  it.live("isolates event listeners and closes pending and idle iterators on plugin unload", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      const service = Rpc.define({
+        id: "promise-rpc-events",
+        methods: {},
+        events: {
+          counted: { schema: z.object({ count: z.number() }).transform(({ count }) => ({ text: String(count) })) },
+        },
+      })
+      const subscriptions = Promise.withResolvers<{
+        pending: Promise<IteratorResult<RpcEventPayload<typeof service, "counted">>>
+        idle: AsyncIterator<RpcEventPayload<typeof service, "counted">>
+        nativeIdle: AsyncIterator<unknown>
+      }>()
+      const adapted = PluginPromise.fromPromise(
+        define({
+          id: "promise-rpc-events-plugin",
+          setup: async (ctx) => {
+            const registration = await ctx.rpc.register(service, {})
+            const client = ctx.rpc(service)
+            const first: string[] = []
+            const second: string[] = []
+            const firstSeen = Promise.withResolvers<void>()
+            const secondSeen = Promise.withResolvers<void>()
+            const nextSeen = Promise.withResolvers<void>()
+            const unsubscribe = client.events.on("counted", (event) => {
+              first.push(event.data.text)
+              firstSeen.resolve()
+            })
+            client.events.on("counted", (event) => {
+              second.push(event.data.text)
+              if (event.data.text === "1") secondSeen.resolve()
+              if (event.data.text === "2") nextSeen.resolve()
+            })
+            const controller = new AbortController()
+            const iterator = client.events.subscribe("counted", { signal: controller.signal })[Symbol.asyncIterator]()
+            const next = iterator.next()
+            const idle = client.events.subscribe("counted")[Symbol.asyncIterator]()
+            const idleNext = idle.next()
+            const nativeController = new AbortController()
+            const native = ctx.event.subscribe({ signal: nativeController.signal })[Symbol.asyncIterator]()
+            const nativeNext = native.next()
+            const nativeIdle = ctx.event.subscribe()[Symbol.asyncIterator]()
+            const nativeIdleNext = nativeIdle.next()
+            await registration.events.emit("counted", { count: 1 })
+            await Promise.all([firstSeen.promise, secondSeen.promise])
+            const event = (await next).value
+            expect(event.type).toBe("rpc.promise-rpc-events.counted")
+            expect(event.data).toEqual({ text: "1" })
+            expect(typeof event.location.directory).toBe("string")
+            expect((await idleNext).value.data).toEqual({ text: "1" })
+            expect((await nativeNext).value.type).toBe("rpc.promise-rpc-events.counted")
+            expect((await nativeIdleNext).value.type).toBe("rpc.promise-rpc-events.counted")
+            nativeController.abort()
+            expect((await native.next()).done).toBe(true)
+            unsubscribe()
+            unsubscribe()
+            controller.abort()
+            expect((await iterator.next()).done).toBe(true)
+            await registration.events.emit("counted", { count: 2 })
+            await nextSeen.promise
+            expect(first).toEqual(["1"])
+            expect(second).toEqual(["1", "2"])
+            const aborted = client.events.subscribe("counted", { signal: controller.signal })[Symbol.asyncIterator]()
+            expect((await aborted.next()).done).toBe(true)
+            subscriptions.resolve({
+              pending: client.events.subscribe("counted")[Symbol.asyncIterator]().next(),
+              idle,
+              nativeIdle,
+            })
+          },
+        }),
+      )
+
+      yield* plugins.activate([{ ...adapted, revision: "1" }])
+      expect(yield* plugins.list()).toMatchObject([{ id: adapted.id, state: { status: "active" } }])
+      const active = yield* Effect.promise(() => subscriptions.promise)
+      yield* plugins.activate([])
+      expect((yield* Effect.promise(() => active.pending)).done).toBe(true)
+      expect((yield* Effect.promise(() => active.idle.next())).done).toBe(true)
+      expect((yield* Effect.promise(() => active.nativeIdle.next())).done).toBe(true)
+    }),
+  )
+})

packages/core/test/rpc.test.ts

@@ -0,0 +1,438 @@
+import { describe, expect } from "bun:test"
+import { Bus } from "@opencode-ai/core/bus"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { Location } from "@opencode-ai/core/location"
+import { Rpc } from "@opencode-ai/core/rpc"
+import { AbsolutePath } from "@opencode-ai/core/schema"
+import { Workspace } from "@opencode-ai/core/workspace"
+import type { Event } from "@opencode-ai/schema/event"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Context, Deferred, Effect, Exit, Fiber, Layer, Logger, Schema, Scope, Stream } from "effect"
+import { z } from "zod"
+import { location } from "./fixture/location"
+import { testEffect } from "./lib/effect"
+
+const ref = Location.Ref.make({ directory: AbsolutePath.make("/rpc-project") })
+const it = testEffect(
+  AppNodeBuilder.build(LayerNode.group([Rpc.node, Bus.node, Location.node]), [
+    Location.node.replace(Layer.succeed(Location.Service, location(ref))),
+  ]),
+)
+const Echo = Rpc.define({
+  id: "test.rpc",
+  methods: { echo: { input: z.string(), output: z.string() } },
+  events: { updated: { schema: z.object({ text: z.string() }) } },
+})
+
+describe("Rpc", () => {
+  it.effect("creates handles before registration and resolves on every execution", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const client = rpc.client(Echo)
+      const request = client.echo("hello")
+      expect(yield* request.pipe(Effect.flip)).toEqual({
+        type: "rpc.unavailable",
+        message: "RPC is unavailable: test.rpc",
+      })
+
+      yield* rpc.register(Echo, { echo: (value) => Effect.succeed(value) })
+      expect(yield* request).toBe("hello")
+      yield* rpc.register(Echo, { echo: (value) => Effect.succeed(`${value}!`) })
+      expect(yield* request).toBe("hello!")
+      expect(yield* rpc.call(Echo.id, "missing", "hello").pipe(Effect.flip)).toEqual({
+        type: "rpc.method_not_found",
+        message: "Unknown RPC method: test.rpc.missing",
+      })
+      expect(yield* rpc.call(Echo.id, "toString", "hello").pipe(Effect.flip)).toEqual({
+        type: "rpc.method_not_found",
+        message: "Unknown RPC method: test.rpc.toString",
+      })
+    }),
+  )
+
+  it.effect("uses the latest whole registration and reveals previous implementations on disposal", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const client = rpc.client(Echo)
+      const first = yield* rpc.register(Echo, { echo: () => Effect.succeed("first") })
+      const second = yield* rpc.register(Echo, { echo: () => Effect.succeed("second") })
+      const third = yield* rpc.register(Echo, { echo: () => Effect.succeed("third") })
+      expect(yield* client.echo("hello")).toBe("third")
+      yield* second.dispose
+      expect(yield* client.echo("hello")).toBe("third")
+      yield* third.dispose
+      expect(yield* client.echo("hello")).toBe("first")
+      yield* third.dispose
+      expect(yield* client.echo("hello")).toBe("first")
+      yield* first.dispose
+      expect(Exit.isFailure(yield* client.echo("hello").pipe(Effect.exit))).toBe(true)
+    }),
+  )
+
+  it.effect("removes registrations when their owning scope closes", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      yield* rpc.register(Echo, { echo: () => Effect.succeed("original") })
+      const scope = yield* Scope.make()
+      yield* rpc.register(Echo, { echo: () => Effect.succeed("override") }).pipe(Scope.provide(scope))
+      expect(yield* rpc.client(Echo).echo("hello")).toBe("override")
+      yield* Scope.close(scope, Exit.void)
+      expect(yield* rpc.client(Echo).echo("hello")).toBe("original")
+    }),
+  )
+
+  it.effect("validates inputs before running handlers and validates returned results", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const received: string[] = []
+      yield* rpc.register(Echo, {
+        echo: (value) =>
+          Effect.sync(() => {
+            received.push(value)
+            return value
+          }),
+      })
+      expect(Exit.isFailure(yield* rpc.call(Echo.id, "echo", 42).pipe(Effect.exit))).toBe(true)
+      expect(received).toEqual([])
+
+      const Checked = Rpc.define({
+        id: "checked",
+        methods: { echo: { input: z.string(), output: z.string().min(3) } },
+        events: {},
+      })
+      yield* rpc.register(Checked, { echo: () => Effect.succeed("a") })
+      expect(Exit.isFailure(yield* rpc.client(Checked).echo("hello").pipe(Effect.exit))).toBe(true)
+    }),
+  )
+
+  it.effect("leaves local transport values to the declared schema", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const Identity = Rpc.define({
+        id: "identity",
+        methods: { echo: { input: Schema.Unknown, output: Schema.Unknown } },
+        events: {},
+      })
+      yield* rpc.register(Identity, { echo: Effect.succeed })
+      const value = new Date(0)
+      expect(yield* rpc.client(Identity).echo(value)).toBe(value)
+    }),
+  )
+
+  it.effect("applies Standard Schema transforms once for inputs, outputs, and events", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const counts = { input: 0, output: 0, event: 0 }
+      const Transformed = Rpc.define({
+        id: "transformed",
+        methods: {
+          count: {
+            input: z.string().transform((value) => {
+              counts.input++
+              return Number(value)
+            }),
+            output: z.number().transform((value) => {
+              counts.output++
+              return String(value)
+            }),
+          },
+        },
+        events: {
+          counted: {
+            schema: z.object({ count: z.number() }).transform(({ count }) => {
+              counts.event++
+              return { text: String(count) }
+            }),
+          },
+        },
+      })
+      const registration = yield* rpc.register(Transformed, { count: (value) => Effect.succeed(value + 1) })
+      const client = rpc.client(Transformed)
+      expect(yield* client.count("41")).toBe("42")
+      const events = yield* client.events
+        .subscribe("counted")
+        .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+      yield* registration.events.emit("counted", { count: 42 })
+      expect((yield* Fiber.join(events))[0].data).toEqual({ text: "42" })
+      expect(counts).toEqual({ input: 1, output: 1, event: 1 })
+    }),
+  )
+
+  it.effect("keeps encoded dispatch and decoded local results consistent for Effect codecs", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const Codec = Rpc.define({
+        id: "codec",
+        methods: { count: { input: Schema.FiniteFromString, output: Schema.FiniteFromString } },
+        events: { counted: { schema: Schema.Struct({ count: Schema.FiniteFromString }) } },
+      })
+      const registration = yield* rpc.register(Codec, { count: (value) => Effect.succeed(value + 1) })
+      expect(yield* rpc.call(Codec.id, "count", "41")).toBe("42")
+      expect(yield* rpc.client(Codec).count("41")).toBe(42)
+      const events = yield* rpc
+        .client(Codec)
+        .events.subscribe("counted")
+        .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+      yield* registration.events.emit("counted", { count: 42 })
+      expect((yield* Fiber.join(events))[0].data).toEqual({ count: 42 })
+    }),
+  )
+
+  it.effect("validates declared error data and decodes it for local clients", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const Failing = Rpc.define({
+        id: "failing",
+        methods: {
+          standard: {
+            input: z.undefined(),
+            output: z.string(),
+            errors: { missing: z.object({ attempts: z.string().transform(Number) }) },
+          },
+          effect: {
+            input: Schema.Undefined,
+            output: Schema.String,
+            errors: { invalid: Schema.Struct({ count: Schema.FiniteFromString }) },
+          },
+        },
+        events: {},
+      })
+      yield* rpc.register(Failing, {
+        standard: (_input, context) => Effect.fail(context.error("missing", "Missing", { attempts: "2" })),
+        effect: (_input, context) => Effect.fail(context.error("invalid", "Invalid", { count: 3 })),
+      })
+
+      expect(yield* rpc.call(Failing.id, "standard", undefined).pipe(Effect.flip)).toEqual({
+        type: "missing",
+        message: "Missing",
+        data: { attempts: 2 },
+      })
+      expect(yield* rpc.client(Failing).standard().pipe(Effect.flip)).toEqual({
+        type: "missing",
+        message: "Missing",
+        data: { attempts: 2 },
+      })
+      expect(yield* rpc.call(Failing.id, "effect", undefined).pipe(Effect.flip)).toEqual({
+        type: "invalid",
+        message: "Invalid",
+        data: { count: "3" },
+      })
+      expect(yield* rpc.client(Failing).effect().pipe(Effect.flip)).toEqual({
+        type: "invalid",
+        message: "Invalid",
+        data: { count: 3 },
+      })
+    }),
+  )
+
+  it.effect("fails in-process callers with a typed rpc.internal error for defects and undeclared errors", () => {
+    const logged: unknown[] = []
+    const logger = Logger.make((entry) => {
+      if (entry.logLevel === "Error") logged.push(entry.message)
+    })
+    return Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const Broken = Rpc.define({
+        id: "broken",
+        methods: {
+          dies: { input: z.undefined(), output: z.string() },
+          throws: { input: z.undefined(), output: z.string() },
+          raw: { input: z.undefined(), output: z.string() },
+          undeclared: { input: z.undefined(), output: z.string(), errors: { known: z.object({}) } },
+        },
+        events: {},
+      })
+      yield* rpc.register(Broken, {
+        dies: () => Effect.die(new Error("handler defect")),
+        throws: () => {
+          throw new Error("handler threw")
+        },
+        // @ts-expect-error the Promise adapter fails with raw thrown values that the types otherwise forbid
+        raw: () => Effect.fail(new Error("raw failure")),
+        // @ts-expect-error undeclared error names are rejected statically; the runtime contract is under test
+        undeclared: (_input, context) => Effect.fail(context.error("unknown", "Unknown")),
+      })
+      const client = rpc.client(Broken)
+      const internal = { type: "rpc.internal" as const, message: "RPC call failed" }
+
+      // Every path yields the same typed failure the HTTP handler already exposes, so callers can recover.
+      expect(yield* client.dies().pipe(Effect.flip)).toEqual(internal)
+      expect(yield* client.throws().pipe(Effect.flip)).toEqual(internal)
+      expect(yield* client.raw().pipe(Effect.flip)).toEqual(internal)
+      expect(yield* rpc.call(Broken.id, "dies", undefined).pipe(Effect.flip)).toEqual(internal)
+      expect(yield* client.undeclared().pipe(Effect.flip)).toEqual({
+        type: "rpc.internal",
+        message: "Undeclared RPC error: unknown",
+      })
+      expect(
+        yield* client.dies().pipe(
+          Effect.catchIf(
+            (error) => error.type === "rpc.internal",
+            (error) => Effect.succeed(`recovered:${error.type}`),
+          ),
+        ),
+      ).toBe("recovered:rpc.internal")
+      // Each conversion logs its cause once so the detail is not lost.
+      expect(logged).toHaveLength(6)
+      expect(logged[0]).toEqual([
+        "rpc handler failed",
+        { rpc: "broken", method: "dies", error: new Error("handler defect") },
+      ])
+      expect(logged[2]).toEqual([
+        "rpc handler failed",
+        { rpc: "broken", method: "raw", error: new Error("raw failure") },
+      ])
+    }).pipe(Effect.provideService(Logger.CurrentLoggers, new Set([logger])))
+  })
+
+  it.effect("keeps other event consumers running after one subscription ends", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const registration = yield* rpc.register(Echo, { echo: (value) => Effect.succeed(value) })
+      const client = rpc.client(Echo)
+      const first = yield* client.events.subscribe("updated").pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      const second = yield* client.events
+        .subscribe("updated")
+        .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+      yield* registration.events.emit("updated", { text: "first" })
+      const received = yield* Fiber.join(first)
+      expect(received.map((event) => event.data.text)).toEqual(["first"])
+      Reflect.set(received[0].location, "directory", "/consumer-mutated")
+      yield* registration.events.emit("updated", { text: "second" })
+      expect((yield* Fiber.join(second)).map((event) => event.data.text)).toEqual(["first", "second"])
+    }),
+  )
+
+  it.effect("validates plain JSON Schema inputs and outputs without type inference", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const Raw = Rpc.define({
+        id: "raw",
+        methods: { count: { input: { type: "integer", minimum: 0 }, output: { type: "integer", minimum: 1 } } },
+        events: {
+          counted: {
+            schema: {
+              type: "object",
+              properties: { count: { type: "integer", minimum: 1 } },
+              required: ["count"],
+              additionalProperties: false,
+            },
+          },
+        },
+      })
+      const registration = yield* rpc.register(Raw, { count: (value) => Effect.succeed(value) })
+      expect(yield* rpc.call(Raw.id, "count", 42)).toBe(42)
+      expect(Exit.isFailure(yield* rpc.call(Raw.id, "count", "42").pipe(Effect.exit))).toBe(true)
+      expect(Exit.isFailure(yield* rpc.call(Raw.id, "count", 0).pipe(Effect.exit))).toBe(true)
+      expect(Exit.isFailure(yield* registration.events.emit("counted", { count: 0 }).pipe(Effect.exit))).toBe(true)
+    }),
+  )
+
+  it.effect("supports methods with no input and no returned value", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const Empty = Rpc.define({
+        id: "empty",
+        methods: { ping: { input: z.undefined(), output: z.undefined() } },
+        events: {},
+      })
+      yield* rpc.register(Empty, { ping: () => Effect.undefined })
+      expect(yield* rpc.client(Empty).ping()).toBeUndefined()
+    }),
+  )
+
+  it.effect("keeps in-flight calls on their original implementation after removal", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const started = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      const registration = yield* rpc.register(Echo, {
+        echo: (value) =>
+          Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release)), Effect.as(value)),
+      })
+      const call = yield* rpc.client(Echo).echo("original").pipe(Effect.forkScoped)
+      yield* Deferred.await(started)
+      yield* registration.dispose
+      yield* rpc.register(Echo, { echo: () => Effect.succeed("replacement") })
+      expect(yield* rpc.client(Echo).echo("hello")).toBe("replacement")
+      yield* Deferred.succeed(release, undefined)
+      expect(yield* Fiber.join(call)).toBe("original")
+    }),
+  )
+
+  it.effect("interrupts the running Effect handler when its call is cancelled", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const started = yield* Deferred.make<void>()
+      const stopped = yield* Deferred.make<void>()
+      yield* rpc.register(Echo, {
+        echo: () =>
+          Deferred.succeed(started, undefined).pipe(
+            Effect.andThen(Effect.never),
+            Effect.onInterrupt(() => Deferred.succeed(stopped, undefined)),
+          ),
+      })
+      const call = yield* rpc.client(Echo).echo("hello").pipe(Effect.forkScoped)
+      yield* Deferred.await(started)
+      yield* Fiber.interrupt(call)
+      yield* Deferred.await(stopped)
+    }),
+  )
+
+  it.effect("isolates registrations and subscriptions while publishing location-tagged events on the shared bus", () =>
+    Effect.gen(function* () {
+      const rpc = yield* Rpc.Service
+      const bus = yield* Bus.Service
+      const otherRef = Location.Ref.make({ directory: ref.directory, workspaceID: Workspace.ID.make("wrk_other") })
+      const otherContext = yield* Layer.build(
+        LayerNode.compile(Rpc.node, {
+          replacements: [
+            Bus.node.replace(Layer.succeed(Bus.Service, bus)),
+            Location.node.replace(Layer.succeed(Location.Service, location(otherRef))),
+          ],
+        }).pipe(Layer.fresh),
+      )
+      const other = Context.get(otherContext, Rpc.Service)
+      const first = yield* rpc.register(Echo, { echo: () => Effect.succeed("first") })
+      expect(Exit.isFailure(yield* other.client(Echo).echo("hello").pipe(Effect.exit))).toBe(true)
+      const second = yield* other.register(Echo, { echo: () => Effect.succeed("second") })
+      expect(yield* rpc.client(Echo).echo("hello")).toBe("first")
+      expect(yield* other.client(Echo).echo("hello")).toBe("second")
+
+      const all: Event.Payload[] = []
+      const unsubscribe = yield* bus.listen((event) =>
+        Effect.sync(() => {
+          all.push(event)
+        }),
+      )
+      const localEvents = yield* rpc
+        .client(Echo)
+        .events.subscribe("updated")
+        .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      const otherEvents = yield* other
+        .client(Echo)
+        .events.subscribe("updated")
+        .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+      yield* second.events.emit("updated", { text: "second" })
+      yield* first.events
+        .emit("updated", { text: "first" })
+        .pipe(Effect.provideService(Location.Service, location(otherRef)))
+      expect((yield* Fiber.join(localEvents))[0]).toMatchObject({
+        type: "rpc.test.rpc.updated",
+        data: { text: "first" },
+        location: ref,
+      })
+      expect((yield* Fiber.join(otherEvents))[0]).toMatchObject({
+        type: "rpc.test.rpc.updated",
+        data: { text: "second" },
+        location: otherRef,
+      })
+      expect(all.map((event) => event.location)).toEqual([otherRef, ref])
+      yield* unsubscribe
+    }),
+  )
+})

packages/server/src/handlers/rpc.ts

@@ -14,7 +14,7 @@ export const RpcHandler = HttpApiBuilder.group(Api, "server.rpc", (handlers) =>
       return output === undefined ? {} : { output }
     }).pipe(
       Effect.mapError((error) =>
-        error.type === "rpc.invalid_output"
+        error.type === "rpc.invalid_output" || error.type === "rpc.internal"
           ? new RpcInternalError({ type: error.type, message: error.message })
           : new RpcError({
               type: error.type,
@@ -22,12 +22,10 @@ export const RpcHandler = HttpApiBuilder.group(Api, "server.rpc", (handlers) =>
               ...(error.data === undefined ? {} : { data: error.data }),
             }),
       ),
-      Effect.catchDefect((error) =>
-        Effect.fail(
-          new RpcInternalError({
-            type: "rpc.internal",
-            message: error instanceof Error ? error.message : "RPC call failed",
-          }),
+      // Core already converts handler defects; this net catches anything else without echoing its text.
+      Effect.catchDefect((defect) =>
+        Effect.logError("rpc call failed", { rpc: params.rpcID, method: params.method, defect }).pipe(
+          Effect.andThen(Effect.fail(new RpcInternalError({ type: "rpc.internal", message: "RPC call failed" }))),
         ),
       ),
     ),

packages/server/test/rpc.test.ts

@@ -0,0 +1,371 @@
+import { expect } from "bun:test"
+import { mkdir } from "node:fs/promises"
+import path from "node:path"
+import { Location } from "@opencode-ai/core/location"
+import { LocationServiceMap } from "@opencode-ai/core/location-services"
+import { Plugin } from "@opencode-ai/core/plugin"
+import { SdkPlugins } from "@opencode-ai/core/plugin/sdk"
+import { define } from "@opencode-ai/plugin/effect/plugin"
+import { fromPromise } from "@opencode-ai/plugin/promise/adapter"
+import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
+import { Rpc } from "@opencode-ai/schema/rpc"
+import { AbsolutePath } from "@opencode-ai/schema/schema"
+import { Context, Deferred, Effect, Fiber, Layer, Schema, Stream } from "effect"
+import { HttpEffect, HttpRouter, HttpServer } from "effect/unstable/http"
+import { tmpdirScoped } from "../../core/test/fixture/tmpdir"
+import { it } from "../../core/test/lib/effect"
+import { createRoutes } from "../src/routes"
+
+type RpcEvent = Extract<OpenCodeEvent, { type: `rpc.${string}` }>
+
+const authorization = `Basic ${btoa("opencode:secret")}`
+
+const fixture = Effect.fn(function* (plugins: readonly Parameters<SdkPlugins.Interface["register"]>[0][]) {
+  const tmp = yield* tmpdirScoped("opencode-rpc-server-")
+  const first = path.join(tmp.path, "first")
+  const second = path.join(tmp.path, "second")
+  const config = path.join(tmp.path, "config")
+  yield* Effect.promise(() => Promise.all([first, second, config].map((directory) => mkdir(directory))))
+  const context = yield* Layer.build(
+    createRoutes({
+      password: "secret",
+      database: { path: ":memory:" },
+      config: { directory: config, project: false, content: "{}" },
+      fs: { filewatcher: false },
+    }).pipe(Layer.provide(HttpServer.layerServices)),
+  )
+  const sdk = Context.get(context, SdkPlugins.Service)
+  yield* Effect.forEach(plugins, (plugin) => sdk.register(plugin))
+  const locations = Context.get(context, LocationServiceMap.Service)
+  const handler = Context.get(context, HttpRouter.HttpRouter).asHttpEffect().pipe(HttpEffect.toWebHandlerWith(context))
+  return {
+    first,
+    second,
+    handler,
+    boot: (directory: string) =>
+      Plugin.awaitActivation.pipe(
+        Effect.provide(locations.get(Location.Ref.make({ directory: AbsolutePath.make(directory) }))),
+      ),
+    call: (
+      route: string,
+      body: unknown = {},
+      options: { directory?: string; headers?: Record<string, string>; signal?: AbortSignal } = {},
+    ) =>
+      Effect.promise(() => {
+        const url = new URL(`/api/rpc/${route}`, "http://opencode.local")
+        if (options.directory) url.searchParams.set("location[directory]", options.directory)
+        return handler(
+          new Request(url, {
+            method: "POST",
+            headers: { authorization, "content-type": "application/json", ...options.headers },
+            body: JSON.stringify(body),
+            signal: options.signal,
+          }),
+        )
+      }),
+  }
+})
+
+it.live("dispatches RPC wrappers with query, header and default locations and generic failures", () =>
+  Effect.gen(function* () {
+    const Echo = Rpc.define({
+      id: "transport.echo",
+      methods: {
+        echo: { input: Schema.String, output: Schema.String },
+        json: { input: Schema.Json, output: Schema.Json },
+        empty: { input: Schema.Undefined, output: Schema.Undefined },
+        fail: {
+          input: Schema.Undefined,
+          output: Schema.String,
+          errors: { rejected: Schema.Struct({ reason: Schema.String }) },
+        },
+        defect: { input: Schema.Undefined, output: Schema.String },
+        undeclared: { input: Schema.Undefined, output: Schema.String },
+        invalid: { input: Schema.Undefined, output: { type: "string" } },
+      },
+      events: {},
+    })
+    const server = yield* fixture([
+      define({
+        id: "transport-implementer",
+        effect: (ctx) =>
+          Effect.gen(function* () {
+            yield* ctx.rpc.register(Echo, {
+              echo: (input) => Effect.succeed(`${ctx.location.directory}:${input}`),
+              json: (input) => Effect.succeed(input),
+              empty: () => Effect.succeed(undefined),
+              fail: (_input, context) =>
+                Effect.fail(context.error("rejected", "handler failed", { reason: "declared" })),
+              defect: () => Effect.die(new Error("secret handler detail")),
+              // @ts-expect-error undeclared error names are rejected statically; the runtime contract is under test
+              undeclared: (_input, context) => Effect.fail(context.error("surprise", "undeclared failure")),
+              invalid: () => Effect.succeed(123),
+            })
+          }).pipe(Effect.orDie),
+      }),
+    ])
+    yield* server.boot(server.first)
+    yield* server.boot(server.second)
+    yield* server.boot(process.cwd())
+    const selected = yield* server.call(
+      "transport.echo/echo",
+      { input: "selected" },
+      {
+        directory: server.first,
+        headers: { "x-opencode-directory": encodeURIComponent(server.second) },
+      },
+    )
+    expect(selected.status).toBe(200)
+    expect(yield* Effect.promise(() => selected.json())).toEqual({ output: `${server.first}:selected` })
+    const header = yield* server.call(
+      "transport.echo/echo",
+      { input: "header" },
+      {
+        headers: { "x-opencode-directory": encodeURIComponent(server.second) },
+      },
+    )
+    expect(yield* Effect.promise(() => header.json())).toEqual({ output: `${server.second}:header` })
+    const fallback = yield* server.call("transport.echo/echo", { input: "default" })
+    expect(yield* Effect.promise(() => fallback.json())).toEqual({ output: `${process.cwd()}:default` })
+    const empty = yield* server.call("transport.echo/empty")
+    expect(empty.status).toBe(200)
+    expect(yield* Effect.promise(() => empty.json())).toEqual({})
+    yield* Effect.forEach([null, false, 42, ["array"], { location: "ordinary input" }], (input) =>
+      Effect.gen(function* () {
+        const response = yield* server.call("transport.echo/json", { input })
+        expect(response.status).toBe(200)
+        expect(yield* Effect.promise(() => response.json())).toEqual({ output: input })
+      }),
+    )
+    const denied = yield* server.call("transport.echo/empty", {}, { headers: { authorization: "" } })
+    expect(denied.status).toBe(401)
+    yield* Effect.forEach(
+      [
+        {
+          route: "missing/echo",
+          body: {},
+          error: { type: "rpc.unavailable", message: "RPC is unavailable: missing" },
+        },
+        {
+          route: "transport.echo/missing",
+          body: {},
+          error: { type: "rpc.method_not_found", message: "Unknown RPC method: transport.echo.missing" },
+        },
+        {
+          route: "transport.echo/fail",
+          body: {},
+          error: { type: "rejected", message: "handler failed", data: { reason: "declared" } },
+        },
+        { route: "transport.echo/echo", body: { input: 123 }, error: { type: "rpc.invalid_input" } },
+      ],
+      (item) =>
+        Effect.gen(function* () {
+          const response = yield* server.call(item.route, item.body)
+          expect(response.status).toBe(400)
+          expect(yield* Effect.promise(() => response.json())).toMatchObject({
+            _tag: "RpcError",
+            message: expect.any(String),
+            ...item.error,
+          })
+        }),
+    )
+    // Handler exception text never reaches HTTP clients; the fixed message matches the in-process contract.
+    const defect = yield* server.call("transport.echo/defect")
+    expect(defect.status).toBe(500)
+    expect(yield* Effect.promise(() => defect.json())).toEqual({
+      _tag: "RpcInternalError",
+      type: "rpc.internal",
+      message: "RPC call failed",
+    })
+    const undeclared = yield* server.call("transport.echo/undeclared")
+    expect(undeclared.status).toBe(500)
+    expect(yield* Effect.promise(() => undeclared.json())).toEqual({
+      _tag: "RpcInternalError",
+      type: "rpc.internal",
+      message: "Undeclared RPC error: surprise",
+    })
+    const invalid = yield* server.call("transport.echo/invalid")
+    expect(invalid.status).toBe(500)
+    expect(yield* Effect.promise(() => invalid.json())).toMatchObject({
+      _tag: "RpcInternalError",
+      type: "rpc.invalid_output",
+      message: expect.any(String),
+    })
+    const malformed = yield* server.call("transport.echo/echo", "not a wrapper")
+    expect(malformed.status).toBe(400)
+    expect(yield* Effect.promise(() => malformed.json())).toMatchObject({
+      _tag: "InvalidRequestError",
+      message: expect.any(String),
+    })
+  }),
+)
+
+it.live("request cancellation interrupts Effect RPC handlers and signals Promise RPC handlers", () =>
+  Effect.gen(function* () {
+    const started = yield* Deferred.make<void>()
+    const stopped = yield* Deferred.make<void>()
+    const promiseStarted = Promise.withResolvers<void>()
+    const promiseStopped = Promise.withResolvers<void>()
+    const Blocking = Rpc.define({
+      id: "blocking",
+      methods: { wait: { input: Schema.Undefined, output: Schema.Undefined } },
+      events: {},
+    })
+    const PromiseBlocking = Rpc.define({
+      id: "promise-blocking",
+      methods: { wait: { input: { type: "null" }, output: { type: "null" } } },
+      events: {},
+    })
+    const server = yield* fixture([
+      define({
+        id: "effect-blocking",
+        effect: (ctx) =>
+          ctx.rpc
+            .register(Blocking, {
+              wait: () =>
+                Deferred.succeed(started, undefined).pipe(
+                  Effect.andThen(Effect.never),
+                  Effect.ensuring(Deferred.succeed(stopped, undefined)),
+                ),
+            })
+            .pipe(Effect.asVoid, Effect.orDie),
+      }),
+      fromPromise({
+        id: "promise-blocking",
+        async setup(ctx) {
+          await ctx.rpc.register(PromiseBlocking, {
+            wait: (_input, call) =>
+              new Promise<null>((resolve) => {
+                promiseStarted.resolve()
+                call.signal.addEventListener(
+                  "abort",
+                  () => {
+                    promiseStopped.resolve()
+                    resolve(null)
+                  },
+                  { once: true },
+                )
+              }),
+          })
+        },
+      }),
+    ])
+    yield* server.boot(server.first)
+    const controller = new AbortController()
+    const pending = yield* server
+      .call(
+        "blocking/wait",
+        {},
+        {
+          directory: server.first,
+          signal: controller.signal,
+        },
+      )
+      .pipe(Effect.forkScoped)
+    yield* Deferred.await(started)
+    controller.abort()
+    yield* Deferred.await(stopped)
+    expect((yield* Fiber.join(pending)).status).not.toBe(400)
+    const promiseController = new AbortController()
+    const promisePending = yield* server
+      .call(
+        "promise-blocking/wait",
+        { input: null },
+        {
+          directory: server.first,
+          signal: promiseController.signal,
+        },
+      )
+      .pipe(Effect.forkScoped)
+    yield* Effect.promise(() => promiseStarted.promise)
+    promiseController.abort()
+    yield* Effect.promise(() => promiseStopped.promise)
+    expect((yield* Fiber.join(promisePending)).status).not.toBe(400)
+  }),
+)
+
+it.live(
+  "public SSE and generic native plugin subscriptions receive RPC events across locations",
+  () =>
+    Effect.gen(function* () {
+      const Updates = Rpc.define({
+        id: "updates",
+        methods: { emit: { input: Schema.String, output: Schema.Undefined } },
+        events: { updated: { schema: Schema.Struct({ text: Schema.String }) } },
+      })
+      const received: RpcEvent[] = []
+      const observed = yield* Deferred.make<void>()
+      const server = yield* fixture([
+        define({
+          id: "updates-implementer",
+          effect: (ctx) =>
+            Effect.gen(function* () {
+              const registration = yield* ctx.rpc.register(Updates, {
+                emit: (input): Effect.Effect<undefined> =>
+                  registration.events.emit("updated", { text: input }).pipe(Effect.as(undefined), Effect.orDie),
+              })
+            }).pipe(Effect.orDie),
+        }),
+        define({
+          id: "native-observer",
+          effect: (ctx) =>
+            Effect.gen(function* () {
+              // One observer instance should see both locations, just like the public native stream.
+              if (path.basename(ctx.location.directory) !== "first") return
+              yield* ctx.event.subscribe().pipe(
+                Stream.filter((event): event is RpcEvent => event.type === "rpc.updates.updated"),
+                Stream.take(2),
+                Stream.runForEach((event) => Effect.sync(() => received.push(event))),
+                Effect.andThen(Deferred.succeed(observed, undefined)),
+                Effect.forkScoped({ startImmediately: true }),
+              )
+            }).pipe(Effect.orDie),
+        }),
+      ])
+      yield* server.boot(server.first)
+      yield* server.boot(server.second)
+      const response = yield* Effect.promise(() =>
+        server.handler(
+          new Request("http://opencode.local/api/event", {
+            headers: { authorization, "x-opencode-directory": encodeURIComponent(server.first) },
+          }),
+        ),
+      )
+      expect(response.status).toBe(200)
+      if (!response.body) throw new Error("Expected an SSE body")
+      const reader = response.body.pipeThrough(new TextDecoderStream()).getReader()
+      yield* Effect.addFinalizer(() => Effect.promise(() => reader.cancel()))
+      expect((yield* Effect.promise(() => reader.read())).value).toContain('"type":"server.connected"')
+      const first = yield* server.call("updates/emit", { input: "first" }, { directory: server.first })
+      const second = yield* server.call("updates/emit", { input: "second" }, { directory: server.second })
+      expect(first.status).toBe(200)
+      expect(second.status).toBe(200)
+      const events: RpcEvent[] = []
+      while (events.length < 2) {
+        const chunk = yield* Effect.promise(() => reader.read())
+        if (chunk.done) throw new Error("Event stream closed before RPC events arrived")
+        events.push(
+          ...chunk.value
+            .split("\n\n")
+            .filter((frame) => frame.startsWith("data: "))
+            .map((frame) => Schema.decodeUnknownSync(Schema.fromJsonString(OpenCodeEvent))(frame.slice(6)))
+            .filter((event): event is RpcEvent => event.type === "rpc.updates.updated"),
+        )
+      }
+      yield* Deferred.await(observed)
+      expect(events).toMatchObject([
+        {
+          type: "rpc.updates.updated",
+          location: { directory: server.first },
+          data: { text: "first" },
+        },
+        {
+          type: "rpc.updates.updated",
+          location: { directory: server.second },
+          data: { text: "second" },
+        },
+      ])
+      expect(received).toEqual(events)
+    }),
+  15_000,
+)