From 58deb52dcd048ef6d2de6287c640041ea862f3ff Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Sat, 15 Aug 2026 18:55:44 -0400 Subject: [PATCH] feat(plugin): select event subscriptions --- packages/core/src/config/plugin/agent.ts | 2 +- packages/core/src/config/plugin/command.ts | 2 +- packages/core/src/config/plugin/policy.ts | 3 +- packages/core/src/config/plugin/provider.ts | 3 +- packages/core/src/config/plugin/reference.ts | 3 +- packages/core/src/config/plugin/skill.ts | 3 +- packages/core/src/config/plugin/websearch.ts | 3 +- packages/core/src/plugin/host.ts | 8 +++- packages/core/test/plugin.test.ts | 43 +++++++++++++++++--- packages/core/test/plugin/promise.test.ts | 25 +++++++++++- packages/plugin/src/effect/event.ts | 14 ++++++- packages/plugin/src/promise/adapter.ts | 11 +++-- packages/plugin/src/promise/event.ts | 13 +++++- packages/plugin/test/event-types.test.ts | 26 ++++++++++++ packages/www/content/docs/build/plugins.mdx | 8 ++++ 15 files changed, 141 insertions(+), 26 deletions(-) create mode 100644 packages/plugin/test/event-types.test.ts diff --git a/packages/core/src/config/plugin/agent.ts b/packages/core/src/config/plugin/agent.ts index c14f96bdd3e..0a157758186 100644 --- a/packages/core/src/config/plugin/agent.ts +++ b/packages/core/src/config/plugin/agent.ts @@ -82,7 +82,7 @@ export const Plugin = define({ .pipe( Stream.filterEffect((update) => Effect.map(config.entries(), (entries) => isAgentSource(entries, update.path))), ) - const configUpdates = ctx.event.subscribe().pipe(Stream.filter((event) => event.type === "config.updated")) + const configUpdates = ctx.event.subscribe("config.updated") yield* Stream.merge(sourceChanges, configUpdates).pipe( Stream.debounce("100 millis"), Stream.runForEach(() => reload), diff --git a/packages/core/src/config/plugin/command.ts b/packages/core/src/config/plugin/command.ts index 458d08d5f23..1ff6061877e 100644 --- a/packages/core/src/config/plugin/command.ts +++ b/packages/core/src/config/plugin/command.ts @@ -43,7 +43,7 @@ export const Plugin = define({ Effect.map(config.entries(), (entries) => isCommandSource(entries, update.path)), ), ) - const configUpdates = ctx.event.subscribe().pipe(Stream.filter((event) => event.type === "config.updated")) + const configUpdates = ctx.event.subscribe("config.updated") yield* Stream.merge(sourceChanges, configUpdates).pipe( Stream.debounce("100 millis"), Stream.runForEach(() => reload), diff --git a/packages/core/src/config/plugin/policy.ts b/packages/core/src/config/plugin/policy.ts index 7509e2b9b92..a97e7c6e44f 100644 --- a/packages/core/src/config/plugin/policy.ts +++ b/packages/core/src/config/plugin/policy.ts @@ -22,8 +22,7 @@ export const Plugin = define({ if (policy?.effect === "deny") catalog.provider.remove(record.provider.id) } }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* ctx.event.subscribe("config.updated").pipe( Stream.runForEach(() => config.entries().pipe( Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))), diff --git a/packages/core/src/config/plugin/provider.ts b/packages/core/src/config/plugin/provider.ts index 6d89cb6b7a7..e5b0da8759a 100644 --- a/packages/core/src/config/plugin/provider.ts +++ b/packages/core/src/config/plugin/provider.ts @@ -97,8 +97,7 @@ export const Plugin = define({ } } }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* ctx.event.subscribe("config.updated").pipe( Stream.runForEach(() => config.entries().pipe( Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))), diff --git a/packages/core/src/config/plugin/reference.ts b/packages/core/src/config/plugin/reference.ts index 9b8b444962b..9965762ce55 100644 --- a/packages/core/src/config/plugin/reference.ts +++ b/packages/core/src/config/plugin/reference.ts @@ -49,8 +49,7 @@ export const Plugin = define({ } for (const [name, source] of entries) draft.add(name, source) }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* ctx.event.subscribe("config.updated").pipe( Stream.runForEach(() => config.entries().pipe( Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))), diff --git a/packages/core/src/config/plugin/skill.ts b/packages/core/src/config/plugin/skill.ts index d263fbc420b..cbfa7b08989 100644 --- a/packages/core/src/config/plugin/skill.ts +++ b/packages/core/src/config/plugin/skill.ts @@ -180,8 +180,7 @@ export const Plugin = define({ yield* ctx.skill.transform((draft) => { for (const skill of loaded.skills) draft.add(skill) }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* ctx.event.subscribe("config.updated").pipe( Stream.runForEach(() => config.entries().pipe( Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))), diff --git a/packages/core/src/config/plugin/websearch.ts b/packages/core/src/config/plugin/websearch.ts index 8bfef386653..b56bef76475 100644 --- a/packages/core/src/config/plugin/websearch.ts +++ b/packages/core/src/config/plugin/websearch.ts @@ -14,8 +14,7 @@ export const Plugin = define({ if (selection === false) websearch.default.set(false) if (selection) websearch.default.set(selection.provider) }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* ctx.event.subscribe("config.updated").pipe( Stream.runForEach(() => config.entries().pipe( Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))), diff --git a/packages/core/src/plugin/host.ts b/packages/core/src/plugin/host.ts index d431d2233f1..7be18cd117d 100644 --- a/packages/core/src/plugin/host.ts +++ b/packages/core/src/plugin/host.ts @@ -59,6 +59,12 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p ref.directory === location.directory && ref.workspaceID === location.workspaceID const response = (effect: Effect.Effect) => effect.pipe(Effect.map((data) => ({ location: locationInfo(), data }))) + const subscribe: Plugin.Context["event"]["subscribe"] = (type?: EventManifest.ServerEvent["type"]) => { + if (type === undefined) return bus.subscribe().pipe(Stream.filter(EventManifest.isServer)) + const definition = EventManifest.Server.get(type) + if (!definition) return Stream.fail(new Error(`Unknown plugin event type: ${type}`)) + return bus.subscribe(definition).pipe(Stream.filter(EventManifest.isServer)) + } return { app, @@ -180,7 +186,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p }), }, event: { - subscribe: () => bus.subscribe().pipe(Stream.filter(EventManifest.isServer)), + subscribe, }, integration: { list: () => response(integration.list()), diff --git a/packages/core/test/plugin.test.ts b/packages/core/test/plugin.test.ts index 83a8dbc01cf..c1d99e588b0 100644 --- a/packages/core/test/plugin.test.ts +++ b/packages/core/test/plugin.test.ts @@ -20,24 +20,55 @@ class Secret extends Context.Service()("@opencode/test/PluginSec const versioned = (plugin: EffectPlugin.Plugin, version = "1") => ({ ...plugin, version }) describe("Plugin", () => { - it.live("exposes public events through the plugin context", () => + it.live("selects one public event type through the plugin context", () => Effect.gen(function* () { const plugins = yield* Plugin.Service const bus = yield* Bus.Service const host = yield* PluginHost.make(plugins) - const received = yield* host.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), - Stream.runHead, - Effect.forkScoped({ startImmediately: true }), - ) + const received = yield* host.event + .subscribe("config.updated") + .pipe(Stream.runHead, Effect.forkScoped({ startImmediately: true })) yield* Effect.sleep("10 millis") + yield* bus.publish(Plugin.Event.Updated, {}) yield* bus.publish(ConfigSchema.Event.Updated, {}) expect((yield* Fiber.join(received)).valueOrUndefined?.type).toBe("config.updated") }), ) + it.live("exposes all public events through a wildcard plugin subscription", () => + Effect.gen(function* () { + const plugins = yield* Plugin.Service + const bus = yield* Bus.Service + const host = yield* PluginHost.make(plugins) + const received = yield* host.event + .subscribe() + .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped({ startImmediately: true })) + yield* Effect.sleep("10 millis") + + yield* bus.publish(Plugin.Event.Updated, {}) + yield* bus.publish(ConfigSchema.Event.Updated, {}) + + expect(Array.from(yield* Fiber.join(received), (event) => event.type)).toEqual([ + "plugin.updated", + "config.updated", + ]) + }), + ) + + it.effect("rejects unknown runtime plugin event types", () => + Effect.gen(function* () { + const plugins = yield* Plugin.Service + const host = yield* PluginHost.make(plugins) + const subscribe = host.event.subscribe as unknown as (type: string) => Stream.Stream + + const failure = yield* subscribe("unknown.event").pipe(Stream.runDrain, Effect.flip) + + expect(failure.message).toBe("Unknown plugin event type: unknown.event") + }), + ) + it.effect("replaces plugins by ID and version", () => Effect.gen(function* () { const plugins = yield* Plugin.Service diff --git a/packages/core/test/plugin/promise.test.ts b/packages/core/test/plugin/promise.test.ts index 056b823d6aa..e7132ed90f7 100644 --- a/packages/core/test/plugin/promise.test.ts +++ b/packages/core/test/plugin/promise.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import { Message, SystemPart } from "@opencode-ai/ai" -import { DateTime, Effect, Schema } from "effect" +import { DateTime, Effect, Schema, Stream } from "effect" import { Agent } from "@opencode-ai/core/agent" import { Catalog } from "@opencode-ai/core/catalog" import { Model } from "@opencode-ai/core/model" @@ -18,6 +18,7 @@ import { Provider } from "@opencode-ai/core/provider" import { Project } from "@opencode-ai/core/project" import { AbsolutePath } from "@opencode-ai/core/schema" import { define } from "@opencode-ai/plugin/promise/plugin" +import { Plugin as EffectPlugin } from "@opencode-ai/plugin/effect" import { Money } from "@opencode-ai/schema/money" import type { SessionHooks } from "@opencode-ai/plugin/effect/session" import { testEffect } from "../lib/effect" @@ -27,6 +28,28 @@ import { host as testHost } from "./host" const it = testEffect(PluginTestLayer) describe("fromPromise", () => { + it.effect("forwards a selected event type", () => + Effect.gen(function* () { + let selected: string | undefined + const subscribe: EffectPlugin.Context["event"]["subscribe"] = (type?) => { + selected = type + return Stream.empty + } + const host = testHost({ event: { subscribe } }) + + yield* PluginPromise.fromPromise( + define({ + id: "promise-event-subscribe", + setup: (ctx) => { + ctx.event.subscribe("config.updated") + }, + }), + ).effect(host) + + expect(selected).toBe("config.updated") + }), + ) + it.effect("adapts session creation through the protocol schema", () => Effect.gen(function* () { let seen: unknown diff --git a/packages/plugin/src/effect/event.ts b/packages/plugin/src/effect/event.ts index 283d4109f05..8c3c3ad40a4 100644 --- a/packages/plugin/src/effect/event.ts +++ b/packages/plugin/src/effect/event.ts @@ -1,3 +1,15 @@ import type { EventApi } from "@opencode-ai/client/effect/api" +import type { OpenCodeEvent } from "@opencode-ai/client/effect" +import type { Stream } from "effect" -export interface EventDomain extends Pick, "subscribe"> {} +export type PluginEvent = Exclude +export type PluginEventType = PluginEvent["type"] + +export interface EventSubscribe { + (): Stream.Stream + (type: PluginEventType): Stream.Stream +} + +export interface EventDomain extends Omit, "subscribe"> { + readonly subscribe: EventSubscribe +} diff --git a/packages/plugin/src/promise/adapter.ts b/packages/plugin/src/promise/adapter.ts index 9ed8a6d7192..eb2528120ca 100644 --- a/packages/plugin/src/promise/adapter.ts +++ b/packages/plugin/src/promise/adapter.ts @@ -2,6 +2,7 @@ import { Tool } from "@opencode-ai/schema/tool" import { Effect, Schema, SchemaAST, Scope, Stream } from "effect" import { HttpApiEndpoint, HttpApiSchema } from "effect/unstable/httpapi" import { define } from "../effect/plugin.js" +import type { PluginEventType } from "./event.js" import type { Context, Plugin } from "./plugin.js" import type { Info } from "./tool.js" @@ -149,13 +150,15 @@ export function fromPromise(plugin: Plugin) { reload: () => run(host.command.reload()), }, event: { - subscribe: () => - Stream.toAsyncIterable( - host.event.subscribe().pipe( + subscribe: (type?: PluginEventType) => { + const events = type === undefined ? host.event.subscribe() : host.event.subscribe(type) + return Stream.toAsyncIterable( + events.pipe( Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)), Stream.map((event) => event as unknown as PromiseEvent), ), - ), + ) + }, }, integration: { list: adaptApiMethod(IntegrationEndpoints["integration.list"], host.integration.list), diff --git a/packages/plugin/src/promise/event.ts b/packages/plugin/src/promise/event.ts index 344f5d6f30a..37d62ed5de0 100644 --- a/packages/plugin/src/promise/event.ts +++ b/packages/plugin/src/promise/event.ts @@ -1,3 +1,14 @@ +import type { OpenCodeEvent } from "@opencode-ai/client" import type { EventApi } from "@opencode-ai/client/promise/api" -export interface EventDomain extends Pick {} +export type PluginEvent = Exclude +export type PluginEventType = PluginEvent["type"] + +export interface EventSubscribe { + (): AsyncIterable + (type: PluginEventType): AsyncIterable +} + +export interface EventDomain extends Omit { + readonly subscribe: EventSubscribe +} diff --git a/packages/plugin/test/event-types.test.ts b/packages/plugin/test/event-types.test.ts new file mode 100644 index 00000000000..13a1cca2d18 --- /dev/null +++ b/packages/plugin/test/event-types.test.ts @@ -0,0 +1,26 @@ +import { expect, test } from "bun:test" +import type { Context as EffectContext } from "../src/effect/plugin.js" +import type { Context as PromiseContext } from "../src/promise/plugin.js" + +function effectSubscriptions(ctx: EffectContext) { + ctx.event.subscribe() + ctx.event.subscribe("config.updated") + // @ts-expect-error server.connected is a network-only marker + ctx.event.subscribe("server.connected") + // @ts-expect-error plugin subscriptions select at most one event type + ctx.event.subscribe(["config.updated"]) +} + +function promiseSubscriptions(ctx: PromiseContext) { + ctx.event.subscribe() + ctx.event.subscribe("config.updated") + // @ts-expect-error server.connected is a network-only marker + ctx.event.subscribe("server.connected") + // @ts-expect-error plugin subscriptions select at most one event type + ctx.event.subscribe(["config.updated"]) +} + +test("event subscription types support wildcard and one public event", () => { + expect(effectSubscriptions).toBeFunction() + expect(promiseSubscriptions).toBeFunction() +}) diff --git a/packages/www/content/docs/build/plugins.mdx b/packages/www/content/docs/build/plugins.mdx index cdfe3cb3598..233e57786e5 100644 --- a/packages/www/content/docs/build/plugins.mdx +++ b/packages/www/content/docs/build/plugins.mdx @@ -184,6 +184,14 @@ and plugin options. | `ctx.event` | `subscribe` to the current public server event stream | | `ctx.options` | Readonly options from the matching config object | +Event subscriptions can receive every plugin-visible public event, or select +one event type: + +```ts +ctx.event.subscribe() +ctx.event.subscribe("config.updated") +``` + ### Transform hooks Transform hooks let a plugin modify how OpenCode is configured. Use them to add