Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290export * as Rpc from "./rpc.js"export { define } from "@opencode/schema/rpc"export type { Definition, EventPayload, Failure } from "@opencode/schema/rpc"
import type { RpcClient, RpcDomain, RpcHandlers } from "@opencode/plugin/effect/rpc"import type { Rpc } from "@opencode/schema/rpc"import { Event } from "@opencode/schema/event"import type { Tool } from "@opencode/schema/tool"import type { StandardSchemaV1 } from "@standard-schema/spec"import { makeLocationNode } from "@opencode/util/effect/app-node"import { Context, Deferred, Effect, JsonSchema, Layer, Schema, SchemaRepresentation, Stream } from "effect"import { Bus } from "./bus.js"import { Location } from "./location.js"import { optional, statics } from "./schema.js"
export interface Interface { readonly register: RpcDomain["register"] readonly client: <D extends Rpc.Definition>(definition: D) => RpcClient<D, Rpc.SystemError, never, unknown> readonly call: (rpcID: string, method: string, input: unknown) => Effect.Effect<unknown, Rpc.Failure> readonly close: Effect.Effect<void>}
export class Service extends Context.Service<Service, Interface>()("@opencode/Rpc") {}
class DeclaredError extends Error { constructor( readonly type: string, message: string, readonly data?: unknown, ) { super(message) }}
const layer = Layer.effect( Service, Effect.gen(function* () { const bus = yield* Bus.Service const location = yield* Location.Service const ref = Location.Ref.make({ directory: location.directory, workspaceID: location.workspaceID }) const closed = yield* Deferred.make<void>() const callContext = { error: (type: string, message: string, data?: unknown) => new DeclaredError(type, message, data), } const registrations = new Map< string, Array<{ readonly definition: Rpc.Definition readonly handlers: Readonly<Record<string, Function>> }> >() const definitions = new WeakMap< Rpc.Definition, ReadonlyMap<string, { readonly event: Rpc.EventDefinition; readonly definition: Event.Definition }> >() const eventsFor = (definition: Rpc.Definition) => { const existing = definitions.get(definition) if (existing) return existing const events = new Map( Object.entries(definition.events).map(([name, event]) => [ name, { event, definition: eventDefinition(definition, name) }, ]), ) definitions.set(definition, events) return events }
const register = Effect.fn("Rpc.register")(function* <const D extends Rpc.Definition>( definition: D, handlers: RpcHandlers<NoInfer<D>>, ) { const entry = { definition, handlers } const dispose = Effect.sync(() => { const remaining = (registrations.get(definition.id) ?? []).filter((candidate) => candidate !== entry) if (remaining.length === 0) { registrations.delete(definition.id) return } registrations.set(definition.id, remaining) }) yield* Effect.acquireRelease( Effect.sync(() => registrations.set(definition.id, [...(registrations.get(definition.id) ?? []), entry])), () => dispose, )
const events = eventsFor(definition) return { dispose, events: { emit: Effect.fn("Rpc.emit")(function* (...args: Rpc.EventInput<D>) { const registered = events.get(args[0]) if (!registered) return yield* Effect.fail(new Error(`Unknown RPC event: ${definition.id}.${args[0]}`)) const event = registered.event // SAFETY: The public event-schema contract guarantees an object encoded/output type. // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion const data = (yield* encode(event.schema, args[1])) as Readonly<Record<string, unknown>> return yield* bus .publish(registered.definition, data, { location: Location.Ref.make({ directory: ref.directory, workspaceID: ref.workspaceID }), }) .pipe(Effect.asVoid) }), }, } })
const call = Effect.fn("Rpc.call")(function* (rpcID: string, name: string, input: unknown) { if (yield* Deferred.isDone(closed)) return yield* Effect.fail(failure("rpc.unavailable", `RPC is unavailable: ${rpcID}`)) const entry = registrations.get(rpcID)?.at(-1) if (!entry) return yield* Effect.fail(failure("rpc.unavailable", `RPC is unavailable: ${rpcID}`)) if (!Object.hasOwn(entry.definition.methods, name) || !Object.hasOwn(entry.handlers, name)) return yield* Effect.fail(failure("rpc.method_not_found", `Unknown RPC method: ${rpcID}.${name}`)) const method = entry.definition.methods[name] const handler = entry.handlers[name] const parsed = yield* parse(method.input, input).pipe( Effect.mapError((error) => failure("rpc.invalid_input", errorMessage(error, "Invalid RPC input"))), ) const result = yield* Effect.suspend(() => { // The heterogeneous registry erases handlers after their selected schema validates input. const execution: Effect.Effect<unknown, unknown> = handler(parsed, callContext) return execution }).pipe( Effect.catch((error) => encodeError(method, error)), // Normalize handler bugs here so direct callers can recover just like HTTP callers. Effect.catchDefect((defect) => Effect.logError("rpc handler failed", { rpc: rpcID, method: name, defect }).pipe( Effect.andThen(Effect.fail(failure("rpc.internal", "RPC call failed"))), ), ), // A long-lived RPC must release its location when the routing cache invalidates it. Effect.raceFirst( Deferred.await(closed).pipe( Effect.andThen(Effect.fail(failure("rpc.unavailable", `RPC is unavailable: ${rpcID}`))), ), ), ) return yield* encode(method.output, result).pipe( Effect.mapError((error) => failure("rpc.invalid_output", errorMessage(error, "Invalid RPC output"))), ) })
const client = <D extends Rpc.Definition>(definition: D): RpcClient<D, Rpc.SystemError, never, unknown> => { const events = eventsFor(definition) const methods = Object.fromEntries( Object.entries(definition.methods).map(([name, method]) => [ name, (input: unknown) => call(definition.id, name, input).pipe( Effect.catch((error) => decodeError(method, error)), Effect.flatMap((value) => read(method.output, value).pipe(Effect.catch((cause) => Effect.die(cause)))), ), ]), ) // SAFETY: Every runtime key comes from this definition, and each method delegates through its corresponding schema. // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion return { ...methods, events: { subscribe: <Name extends keyof D["events"] & string>(name: Name) => { const registered = events.get(name) if (!registered) return Stream.fail(new Error(`Unknown RPC event: ${definition.id}.${name}`)) return bus.subscribe(registered.definition).pipe( Stream.provideService(Location.Service, location), Stream.mapEffect((payload) => logicalEvent(definition, name, payload, ref)), ) }, }, } as RpcClient<D, Rpc.SystemError, never, unknown> }
return Service.of({ register, call, client, close: Deferred.succeed(closed, undefined).pipe(Effect.asVoid) }) }),)
export const node = makeLocationNode({ service: Service, layer, deps: [Bus.node, Location.node] })
const fields = { id: Event.ID, created: Schema.Finite, metadata: optional(Schema.Record(Schema.String, Schema.Unknown)), location: optional(Location.Ref),}const EventData = Schema.Record(Schema.String, Schema.Unknown)const jsonSchemas = new WeakMap<JsonSchema.JsonSchema, Schema.Codec<unknown>>()
function eventType<const D extends Rpc.Definition, const Name extends keyof D["events"] & string>( definition: D, name: Name,): `rpc.${D["id"]}.${Name}` { return `rpc.${definition.id}.${name}`}
function eventDefinition(definition: Rpc.Definition, name: string): Event.Definition { const type = eventType(definition, name) const data = EventData return Schema.Struct({ ...fields, type: Schema.Literal(type), data }).pipe( statics(() => ({ type, durability: "ephemeral" as const, durable: undefined, data })), ) satisfies Event.EphemeralDefinition<string, typeof data>}
function parse(schema: Tool.ValueSchema, value: unknown): Effect.Effect<unknown, unknown> { if (Schema.isSchema(schema)) return Schema.decodeUnknownEffect(schema)(value) if (isStandardSchema(schema)) { return Effect.gen(function* () { const parsed = yield* Effect.try({ try: () => schema["~standard"].validate(value), catch: (cause) => cause }) const result = parsed instanceof Promise ? yield* Effect.tryPromise({ try: () => parsed, catch: (cause) => cause }) : parsed if (result.issues) return yield* Effect.fail(new Error(result.issues.map((issue) => issue.message).join("\n"))) return result.value }) } return Effect.try({ try: () => { const existing = jsonSchemas.get(schema) if (existing) return existing const codec = Schema.make<Schema.Codec<unknown>>( SchemaRepresentation.fromJsonSchemaDocument(JsonSchema.fromSchemaDraft2020_12(schema)).ast, ) jsonSchemas.set(schema, codec) return codec }, catch: (cause) => cause, }).pipe(Effect.flatMap((codec) => Schema.decodeUnknownEffect(codec)(value)))}
function encode(schema: Tool.ValueSchema, value: unknown): Effect.Effect<unknown, 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( Effect.catch((cause) => Effect.die(cause)), Effect.flatMap((data) => Effect.fail(failure(error.type, error.message, data))), )}
function failure(type: string, message: string, data?: unknown): Rpc.Failure { return data === undefined ? { type, message } : { type, message, data }}
function errorMessage(error: unknown, fallback: string) { if (error instanceof Error) return error.message if (typeof error === "string") return error return fallback}
function isStandardSchema(schema: Tool.ValueSchema): schema is Extract<Tool.ValueSchema, StandardSchemaV1> { return "~standard" in schema}
function read(schema: Tool.ValueSchema, value: unknown): Effect.Effect<unknown, unknown> { // Standard Schema results were already parsed by the publisher; don't apply transforms twice. return Schema.isSchema(schema) ? Schema.decodeUnknownEffect(schema)(value) : Effect.succeed(value)}
const logicalEvent = Effect.fn("Rpc.logicalEvent")(function* < D extends Rpc.Definition, Name extends keyof D["events"] & string,>( definition: D, name: Name, payload: Event.Payload, ref: Location.Ref,): Effect.fn.Return<Rpc.EventPayload<D, Name>, unknown> { const event = definition.events[name] const data = yield* read(event.schema, payload.data) // SAFETY: The private Bus definition owns the envelope and location. // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion return { ...payload, type: eventType(definition, name), data, location: Location.Ref.make({ directory: ref.directory, workspaceID: ref.workspaceID }), } as Rpc.EventPayload<D, Name>})