From b4630bdf79ba3ca97d073389044b07faaa7cb207 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 2 Sep 2026 16:39:48 -0400 Subject: [PATCH 1/2] test(core): restore plugin rpc tests --- packages/core/test/plugin/rpc-effect.test.ts | 106 +++++ packages/core/test/plugin/rpc-promise.test.ts | 290 ++++++++++++++ packages/core/test/rpc.test.ts | 378 ++++++++++++++++++ packages/server/test/rpc.test.ts | 360 +++++++++++++++++ 4 files changed, 1134 insertions(+) create mode 100644 packages/core/test/plugin/rpc-effect.test.ts create mode 100644 packages/core/test/plugin/rpc-promise.test.ts create mode 100644 packages/core/test/rpc.test.ts create mode 100644 packages/server/test/rpc.test.ts diff --git a/packages/core/test/plugin/rpc-effect.test.ts b/packages/core/test/plugin/rpc-effect.test.ts new file mode 100644 index 000000000000..d5899168384d --- /dev/null +++ b/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") + }), +) diff --git a/packages/core/test/plugin/rpc-promise.test.ts b/packages/core/test/plugin/rpc-promise.test.ts new file mode 100644 index 000000000000..e48ac38b98d5 --- /dev/null +++ b/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.toThrow("handler defect") + 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() + const cancelled = Promise.withResolvers() + const signals = new Map() + await ctx.rpc.register(service, { + wait: async (input, call) => { + signals.set(input, call.signal) + if (input === "complete") return input + started.resolve() + await new Promise((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() + 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() + const release = Promise.withResolvers() + const second = Promise.withResolvers() + const third = Promise.withResolvers() + 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>> + idle: AsyncIterator> + nativeIdle: AsyncIterator + }>() + 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() + const secondSeen = Promise.withResolvers() + const nextSeen = Promise.withResolvers() + 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) + }), + ) +}) diff --git a/packages/core/test/rpc.test.ts b/packages/core/test/rpc.test.ts new file mode 100644 index 000000000000..32812547614b --- /dev/null +++ b/packages/core/test/rpc.test.ts @@ -0,0 +1,378 @@ +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 { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, 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("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() + const release = yield* Deferred.make() + 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() + const stopped = yield* Deferred.make() + 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 + }), + ) +}) diff --git a/packages/server/test/rpc.test.ts b/packages/server/test/rpc.test.ts new file mode 100644 index 000000000000..f4399823f960 --- /dev/null +++ b/packages/server/test/rpc.test.ts @@ -0,0 +1,360 @@ +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 + +const authorization = `Basic ${btoa("opencode:secret")}` + +const fixture = Effect.fn(function* (plugins: readonly Parameters[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; 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 }, + 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("handler defect")), + 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, + }) + }), + ) + 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: "handler defect", + }) + 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() + const stopped = yield* Deferred.make() + const promiseStarted = Promise.withResolvers() + const promiseStopped = Promise.withResolvers() + 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((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() + const server = yield* fixture([ + define({ + id: "updates-implementer", + effect: (ctx) => + Effect.gen(function* () { + const registration = yield* ctx.rpc.register(Updates, { + emit: (input): Effect.Effect => + 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, +) From 3c8308f45f6968788781184b476887f25b2daeef Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 2 Sep 2026 16:47:36 -0400 Subject: [PATCH 2/2] fix(core): fail rpc.internal as a typed error on every path --- packages/core/src/rpc.ts | 31 ++++++---- packages/core/test/plugin/rpc-promise.test.ts | 2 +- packages/core/test/rpc.test.ts | 62 ++++++++++++++++++- packages/server/src/handlers/rpc.ts | 12 ++-- packages/server/test/rpc.test.ts | 15 ++++- 5 files changed, 99 insertions(+), 23 deletions(-) diff --git a/packages/core/src/rpc.ts b/packages/core/src/rpc.ts index 1f306f61066b..d72033e8afb0 100644 --- a/packages/core/src/rpc.ts +++ b/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 = 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 { - 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 { if (!method.errors || !Object.hasOwn(method.errors, error.type)) return Effect.fail(error) return read(method.errors[error.type], error.data).pipe( diff --git a/packages/core/test/plugin/rpc-promise.test.ts b/packages/core/test/plugin/rpc-promise.test.ts index e48ac38b98d5..b42161a4171c 100644 --- a/packages/core/test/plugin/rpc-promise.test.ts +++ b/packages/core/test/plugin/rpc-promise.test.ts @@ -75,7 +75,7 @@ describe("Promise plugin RPC", () => { message: "thrown failure", data: { attempts: 2 }, }) - await expect(client.defect()).rejects.toThrow("handler defect") + 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() diff --git a/packages/core/test/rpc.test.ts b/packages/core/test/rpc.test.ts index 32812547614b..b87ff439fc25 100644 --- a/packages/core/test/rpc.test.ts +++ b/packages/core/test/rpc.test.ts @@ -7,7 +7,7 @@ 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 { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, Schema, Scope, Stream } from "effect" +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" @@ -227,6 +227,66 @@ describe("Rpc", () => { }), ) + 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 diff --git a/packages/server/src/handlers/rpc.ts b/packages/server/src/handlers/rpc.ts index efb0661215d0..5c136d65fef0 100644 --- a/packages/server/src/handlers/rpc.ts +++ b/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" }))), ), ), ), diff --git a/packages/server/test/rpc.test.ts b/packages/server/test/rpc.test.ts index f4399823f960..27b18c602852 100644 --- a/packages/server/test/rpc.test.ts +++ b/packages/server/test/rpc.test.ts @@ -80,6 +80,7 @@ it.live("dispatches RPC wrappers with query, header and default locations and ge 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: {}, @@ -95,7 +96,9 @@ it.live("dispatches RPC wrappers with query, header and default locations and ge empty: () => Effect.succeed(undefined), fail: (_input, context) => Effect.fail(context.error("rejected", "handler failed", { reason: "declared" })), - defect: () => Effect.die(new Error("handler defect")), + 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), @@ -166,12 +169,20 @@ it.live("dispatches RPC wrappers with query, header and default locations and ge }) }), ) + // 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: "handler defect", + 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)