Compare commits

..

2 Commits

Author SHA1 Message Date
Dax Raad 5b9892f0f7 test(plugin): type selected event callback 2026-08-15 18:59:17 -04:00
Dax Raad 58deb52dcd feat(plugin): select event subscriptions 2026-08-15 18:55:44 -04:00
16 changed files with 146 additions and 283 deletions
+2 -47
View File
@@ -27,7 +27,6 @@ const requestedTarget = process.argv.find((arg) => arg.startsWith("--target="))?
const skipInstall = process.argv.includes("--skip-install")
const skipWebUi = process.argv.includes("--skip-web-ui")
const solidPlugin = createSolidTransformPlugin()
const releaseAssets = new Map<string, Promise<Map<string, string>>>()
const allTargets: {
os: string
@@ -162,13 +161,7 @@ async function compileExecutable(item: (typeof allTargets)[number]) {
if (!release) return
const platform = item.os === "win32" ? "windows" : item.os
const name = [
"bun",
platform,
item.arch === "arm64" ? "aarch64" : item.arch,
item.abi,
item.avx2 === false ? "baseline" : undefined,
]
const name = ["bun", platform, item.arch, item.abi, item.avx2 === false ? "baseline" : undefined]
.filter(Boolean)
.join("-")
const cache = path.join(outdir, ".bun", release)
@@ -177,13 +170,7 @@ async function compileExecutable(item: (typeof allTargets)[number]) {
await mkdir(cache, { recursive: true })
const archive = path.join(cache, `${name}.zip`)
const assets = await compileReleaseAssets(release)
const url = assets.get(`${name}.zip`)
if (!url) throw new Error(`Bun release ${release} does not include ${name}.zip`)
const token = process.env.GH_TOKEN ?? process.env.GITHUB_TOKEN
const response = await fetch(url, {
headers: { Accept: "application/octet-stream", ...(token ? { Authorization: `Bearer ${token}` } : {}) },
})
const response = await fetch(`https://github.com/oven-sh/bun/releases/download/${release}/${name}.zip`)
if (!response.ok) throw new Error(`Failed to download ${name} from Bun release ${release}: ${response.status}`)
await Bun.write(archive, response)
await $`unzip -oq ${archive} -d ${cache}`
@@ -191,38 +178,6 @@ async function compileExecutable(item: (typeof allTargets)[number]) {
return executable
}
function compileReleaseAssets(release: string) {
const existing = releaseAssets.get(release)
if (existing) return existing
const pending = fetch(`https://api.github.com/repos/oven-sh/bun/releases/tags/${release}?cache=${Date.now()}`)
.then(async (response) => {
if (!response.ok) throw new Error(`Failed to resolve Bun release ${release}: ${response.status}`)
const data: unknown = await response.json()
if (typeof data !== "object" || data === null || !("assets" in data) || !Array.isArray(data.assets)) {
throw new Error(`Bun release ${release} returned invalid metadata`)
}
return new Map(
data.assets
.filter(
(asset): asset is { name: string; url: string } =>
typeof asset === "object" &&
asset !== null &&
"name" in asset &&
typeof asset.name === "string" &&
"url" in asset &&
typeof asset.url === "string",
)
.map((asset) => [asset.name, asset.url]),
)
})
.catch((error) => {
releaseAssets.delete(release)
throw error
})
releaseAssets.set(release, pending)
return pending
}
function targetName(item: (typeof allTargets)[number]) {
return [
binary,
+1 -1
View File
@@ -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),
+1 -1
View File
@@ -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),
+1 -2
View File
@@ -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))),
+1 -2
View File
@@ -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))),
+1 -2
View File
@@ -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))),
+1 -2
View File
@@ -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))),
+1 -2
View File
@@ -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))),
+7 -1
View File
@@ -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 = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
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()),
+37 -6
View File
@@ -20,24 +20,55 @@ class Secret extends Context.Service<Secret, string>()("@opencode/test/PluginSec
const versioned = <R>(plugin: EffectPlugin.Plugin<R>, 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<never, Error>
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
+25 -167
View File
@@ -1,8 +1,7 @@
import { describe, expect } from "bun:test"
import { Message, SystemPart } from "@opencode-ai/ai"
import { DateTime, Deferred, Effect, Schema, Stream } from "effect"
import { DateTime, Effect, Schema, Stream } from "effect"
import { Agent } from "@opencode-ai/core/agent"
import { Bus } from "@opencode-ai/core/bus"
import { Catalog } from "@opencode-ai/core/catalog"
import { Model } from "@opencode-ai/core/model"
import { Location } from "@opencode-ai/core/location"
@@ -19,7 +18,8 @@ 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 { Config as ConfigSchema } from "@opencode-ai/schema/config"
import { Plugin as EffectPlugin } from "@opencode-ai/plugin/effect"
import type { PluginEventType } from "@opencode-ai/plugin/effect/event"
import { Money } from "@opencode-ai/schema/money"
import type { SessionHooks } from "@opencode-ai/plugin/effect/session"
import { testEffect } from "../lib/effect"
@@ -29,6 +29,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?: PluginEventType) => {
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
@@ -439,170 +461,6 @@ describe("fromPromise", () => {
}),
)
it.effect("closes a pending event iterator with the plugin scope", () =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
let finalized = 0
let iterator: AsyncIterator<unknown> | undefined
let pending: Promise<IteratorResult<unknown>> | undefined
const host = testHost({
event: {
subscribe: () =>
Stream.fromEffect(Deferred.succeed(started, undefined)).pipe(
Stream.flatMap(() => Stream.never),
Stream.ensuring(Effect.sync(() => finalized++)),
),
},
})
yield* Effect.scoped(
PluginPromise.fromPromise(
define({
id: "promise-event-pending",
setup: async (ctx) => {
iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
pending = iterator.next()
await Effect.runPromise(Deferred.await(started))
},
}),
).effect(host),
)
expect(finalized).toBe(1)
if (!pending || !iterator) yield* Effect.die("event iterator was not initialized")
expect(yield* Effect.promise(() => pending)).toEqual({ done: true, value: undefined })
expect(yield* Effect.promise(() => iterator.next())).toEqual({ done: true, value: undefined })
}),
)
it.live("closes event iterators on break, completion, and failure", () =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const plugins = yield* Plugin.Service
const closed: string[] = []
const broke = yield* Deferred.make<void>()
const promisePlugin = define({
id: "promise-event-terminal",
setup: async (ctx) => {
void (async () => {
for await (const _event of ctx.event.subscribe()) break
await Effect.runPromise(Deferred.succeed(broke, undefined))
})()
},
})
yield* plugins.activate([{ ...PluginPromise.fromPromise(promisePlugin), version: "1" }])
yield* Effect.sleep("10 millis")
yield* bus.publish(ConfigSchema.Event.Updated, {})
yield* Deferred.await(broke)
yield* Effect.scoped(
PluginPromise.fromPromise(
define({
id: "promise-event-complete",
setup: async (ctx) => {
const iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
expect(await iterator.next()).toEqual({ done: true, value: undefined })
expect(await iterator.next()).toEqual({ done: true, value: undefined })
},
}),
).effect(
testHost({
event: {
subscribe: () => Stream.empty.pipe(Stream.ensuring(Effect.sync(() => closed.push("complete")))),
},
}),
),
)
yield* Effect.scoped(
PluginPromise.fromPromise(
define({
id: "promise-event-failure",
setup: async (ctx) => {
const iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
await expect(iterator.next()).rejects.toThrow("event failure")
expect(await iterator.next()).toEqual({ done: true, value: undefined })
},
}),
).effect(
testHost({
event: {
subscribe: () =>
Stream.fail(new Error("event failure")).pipe(
Stream.ensuring(Effect.sync(() => closed.push("failure"))),
),
},
}),
),
)
expect(closed).toEqual(["complete", "failure"])
}),
)
it.effect("closes every event iterator when the plugin scope closes", () =>
Effect.gen(function* () {
const ready = yield* Deferred.make<void>()
let started = 0
let finalized = 0
const host = testHost({
event: {
subscribe: () =>
Stream.fromEffect(
Effect.sync(() => ++started).pipe(
Effect.tap((count) => (count === 3 ? Deferred.succeed(ready, undefined) : Effect.void)),
),
).pipe(
Stream.flatMap(() => Stream.never),
Stream.ensuring(Effect.sync(() => finalized++)),
),
},
})
yield* Effect.scoped(
PluginPromise.fromPromise(
define({
id: "promise-event-multiple",
setup: async (ctx) => {
const events = ctx.event.subscribe()
void events[Symbol.asyncIterator]().next()
void events[Symbol.asyncIterator]().next()
void ctx.event.subscribe()[Symbol.asyncIterator]().next()
await Effect.runPromise(Deferred.await(ready))
},
}),
).effect(host),
)
expect(finalized).toBe(3)
}),
)
it.effect("closes a Promise event iterator when the plugin is replaced", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
let iterator: AsyncIterator<unknown> | undefined
let pending: Promise<IteratorResult<unknown>> | undefined
const previous = PluginPromise.fromPromise(
define({
id: "promise-event-replacement",
setup: async (ctx) => {
iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
pending = iterator.next()
},
}),
)
yield* plugins.activate([{ ...previous, version: "1" }])
yield* plugins.activate([{ id: previous.id, version: "2", effect: () => Effect.void }])
if (!pending || !iterator) yield* Effect.die("event iterator was not initialized")
expect(yield* Effect.promise(() => pending)).toEqual({ done: true, value: undefined })
expect(yield* Effect.promise(() => iterator.next())).toEqual({ done: true, value: undefined })
}),
)
it.effect("constructs plain Promise tool definitions in the host", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
+13 -1
View File
@@ -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<EventApi<unknown>, "subscribe"> {}
export type PluginEvent = Exclude<OpenCodeEvent, { readonly type: "server.connected" }>
export type PluginEventType = PluginEvent["type"]
export interface EventSubscribe {
(): Stream.Stream<PluginEvent, unknown>
(type: PluginEventType): Stream.Stream<PluginEvent, unknown>
}
export interface EventDomain extends Omit<EventApi<unknown>, "subscribe"> {
readonly subscribe: EventSubscribe
}
+9 -48
View File
@@ -1,7 +1,8 @@
import { Tool } from "@opencode-ai/schema/tool"
import { Cause, Effect, Exit, Queue, Schema, SchemaAST, Scope, Stream } from "effect"
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,54 +150,14 @@ export function fromPromise(plugin: Plugin) {
reload: () => run(host.command.reload()),
},
event: {
subscribe: () => {
const events = host.event.subscribe().pipe(
Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)),
Stream.map((event) => event as unknown as PromiseEvent),
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),
),
)
return {
[Symbol.asyncIterator]() {
const child = Scope.forkUnsafe(scope)
const done = { done: true, value: undefined } as const
let terminal = false
let closing: Promise<IteratorResult<PromiseEvent>> | undefined
const queue = Effect.gen(function* () {
const queue = yield* Stream.toQueue(events, { capacity: "unbounded" })
// Finalizers are LIFO: mark terminal before queue shutdown wakes a pending next().
yield* Scope.addFinalizer(
child,
Effect.sync(() => (terminal = true)),
)
return queue
}).pipe(Scope.provide(child), Effect.runPromiseWith(context))
const iterator = {
next: () => {
if (terminal) return closing ?? Promise.resolve(done)
return queue
.then((queue) => Effect.runPromiseWith(context)(Queue.take(queue)))
.then(
(value) => (terminal ? (closing ?? done) : { done: false as const, value }),
async (error) => {
if (terminal) return closing ?? done
await iterator.return()
if (Cause.isDone(error)) return done
throw error
},
)
},
return: () => {
if (closing) return closing
terminal = true
closing = Effect.runPromiseWith(context)(Scope.close(child, Exit.void)).then(
() => done,
() => done,
)
return closing
},
}
return iterator
},
}
},
},
integration: {
+12 -1
View File
@@ -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<EventApi, "subscribe"> {}
export type PluginEvent = Exclude<OpenCodeEvent, { readonly type: "server.connected" }>
export type PluginEventType = PluginEvent["type"]
export interface EventSubscribe {
(): AsyncIterable<PluginEvent>
(type: PluginEventType): AsyncIterable<PluginEvent>
}
export interface EventDomain extends Omit<EventApi, "subscribe"> {
readonly subscribe: EventSubscribe
}
+26
View File
@@ -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()
})
+8
View File
@@ -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