mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-16 01:19:19 -04:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5b9892f0f7 | |||
| 58deb52dcd |
@@ -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,
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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))),
|
||||
|
||||
@@ -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))),
|
||||
|
||||
@@ -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))),
|
||||
|
||||
@@ -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))),
|
||||
|
||||
@@ -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))),
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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: {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user