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")
+  }),
+)