diff --git a/packages/cli/test/run/noninteractive.test.ts b/packages/cli/test/run/noninteractive.test.ts index 24d56c4d27ad..10a679dea7b0 100644 --- a/packages/cli/test/run/noninteractive.test.ts +++ b/packages/cli/test/run/noninteractive.test.ts @@ -239,7 +239,7 @@ async function run(input: { yield value } })() - spyOn(sdk.event, "subscribe").mockImplementation(() => stream) + spyOn(sdk.event, "subscribe").mockImplementation((() => stream) as typeof sdk.event.subscribe) spyOn(sdk.permission, "list").mockImplementation(() => ok([]) as never) spyOn(sdk.session.form, "list").mockImplementation( (request) => ok(input.pendingForms?.filter((item) => item.sessionID === request.sessionID) ?? []) as never, diff --git a/packages/client/src/effect/api.ts b/packages/client/src/effect/api.ts index 0683ead1c96a..e6f0e9990022 100644 --- a/packages/client/src/effect/api.ts +++ b/packages/client/src/effect/api.ts @@ -1,6 +1,7 @@ import type { WebsearchApi } from "./api/api.js" export type { RpcApi, RpcClient } from "./rpc.js" +export type { SelectedEvent } from "../shared-events.js" export type * from "./api/api.js" diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index fadf62bae76d..975fd3b0e8be 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -1891,8 +1891,109 @@ export interface RpcApi { readonly call: RpcCallOperation } +export type EventSubscribeInput = { + readonly types?: + | ReadonlyArray< + | "location.shutdown" + | "models-dev.refreshed" + | "credential.updated" + | "credential.switched" + | "integration.updated" + | "provider.updated" + | "model.updated" + | "agent.updated" + | "session.created" + | "session.agent.selected" + | "session.model.selected" + | "session.moved" + | "session.renamed" + | "session.metadata.updated" + | "session.permissions" + | "session.viewed" + | "session.archived" + | "session.unarchived" + | "session.usage.updated" + | "session.deleted" + | "session.forked" + | "session.inbox.delivered" + | "session.inbox.enqueued" + | "session.inbox.cancelled" + | "session.inbox.delivery.changed" + | "session.execution.started" + | "session.execution.succeeded" + | "session.execution.failed" + | "session.execution.interrupted" + | "session.instructions.updated" + | "session.synthetic" + | "session.skill.activated" + | "session.shell.started" + | "session.shell.ended" + | "session.step.started" + | "session.step.streamed" + | "session.step.ended" + | "session.step.failed" + | "session.text.started" + | "session.text.delta" + | "session.text.ended" + | "session.reasoning.started" + | "session.reasoning.delta" + | "session.reasoning.ended" + | "session.tool.input.started" + | "session.tool.input.delta" + | "session.tool.input.ended" + | "session.tool.called" + | "session.tool.progress" + | "session.tool.success" + | "session.tool.failed" + | "session.retry.scheduled" + | "session.compaction.started" + | "session.compaction.delta" + | "session.compaction.ended" + | "session.compaction.failed" + | "session.revert.staged" + | "session.revert.cleared" + | "session.revert.committed" + | "filesystem.changed" + | "reference.updated" + | "permission.asked" + | "permission.replied" + | "plugin.updated" + | "project.updated" + | "worktree.updated" + | "worktree.resolved" + | "command.updated" + | "config.updated" + | "skill.updated" + | "pty.created" + | "pty.updated" + | "pty.exited" + | "pty.deleted" + | "persistent-pty.added" + | "persistent-pty.removed" + | "shell.created" + | "shell.exited" + | "shell.deleted" + | "form.created" + | "form.replied" + | "form.cancelled" + | "websearch.updated" + | "session.status" + | "session.idle" + | "tui.prompt.append" + | "tui.command.execute" + | "tui.toast.show" + | "tui.session.select" + | "installation.updated" + | "installation.update-available" + | "vcs.branch.updated" + | "mcp.status.changed" + | "mcp.resources.changed" + | `${"rpc."}${string}` + > + | undefined +} export type EventSubscribeOutput = OpenCodeEvent -export type EventSubscribeOperation = () => Stream.Stream +export type EventSubscribeOperation = (input?: EventSubscribeInput) => Stream.Stream export interface EventApi { readonly subscribe: EventSubscribeOperation diff --git a/packages/client/src/effect/client.ts b/packages/client/src/effect/client.ts index c757acc80279..4aa01eacb9a6 100644 --- a/packages/client/src/effect/client.ts +++ b/packages/client/src/effect/client.ts @@ -4,9 +4,12 @@ import { Cause, Context, Effect, Stream } from "effect" import { HttpClient, HttpClientRequest } from "effect/http" import { SharedEvents } from "../shared-events.js" import { ClientError, OpenCode } from "./generated/index.js" +import type { EventSubscribeInput } from "./api.js" import { RpcClientRuntime } from "./rpc.js" import type { RpcCallOptions } from "../promise/rpc.js" +type EventType = NonNullable[number] + const CurrentHeaders = Context.Reference("@opencode/client/effect/rpc/headers", { defaultValue: () => undefined, }) @@ -38,8 +41,8 @@ export const make = Effect.fn("OpenCode.make")(function* (options?: { readonly b context, ), ) - const subscribe = () => - Stream.fromAsyncIterable(shared.subscribe(), (error) => error).pipe( + const subscribe = (input?: { readonly types?: ReadonlyArray }) => + Stream.fromAsyncIterable(shared.subscribe({ types: input?.types }), (error) => error).pipe( Stream.catch((error) => Stream.failCause(error instanceof EventFailure ? error.cause : Cause.fail(new ClientError({ cause: error }))), ), diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index e548c7f55818..4659c10b17cd 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -205,6 +205,7 @@ import type { SkillListOutput, RpcCallInput, RpcCallOutput, + EventSubscribeInput, EventSubscribeOutput, PtyListInput, PtyListOutput, @@ -1326,10 +1327,10 @@ const EndpointRpcCall = (raw: RawClient["server.rpc"]) => (input: RpcCallInput) const adaptGroupRpc = (raw: RawClient["server.rpc"]) => ({ call: EndpointRpcCall(raw) }) -const EndpointEventSubscribe = (raw: RawClient["server.event"]) => () => +const EndpointEventSubscribe = (raw: RawClient["server.event"]) => (input?: EventSubscribeInput) => preserveStream()( Stream.unwrap( - raw["event.subscribe"]({}).pipe( + raw["event.subscribe"]({ query: { types: input?.["types"] } }).pipe( Effect.mapError(mapClientError), Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))), ), diff --git a/packages/client/src/promise/api.ts b/packages/client/src/promise/api.ts index ed9b4bd9570a..2c202ac3133a 100644 --- a/packages/client/src/promise/api.ts +++ b/packages/client/src/promise/api.ts @@ -3,6 +3,7 @@ import type { OpenCode } from "./client.js" type Client = ReturnType export type { RpcApi, RpcCallOptions, RpcClient, RpcEventPayload } from "./rpc.js" +export type { SelectedEvent, SubscribeOptions } from "../shared-events.js" export type AgentApi = Client["agent"] export type CommandApi = Client["command"] diff --git a/packages/client/src/promise/client.ts b/packages/client/src/promise/client.ts index 0db239d9cc58..cbcab3dcdf5c 100644 --- a/packages/client/src/promise/client.ts +++ b/packages/client/src/promise/client.ts @@ -9,7 +9,7 @@ export type { ClientOptions, RequestOptions } from "./generated/client.js" export function make(options: ClientOptions) { const raw = OpenCode.make(options) - const events = SharedEvents.make((signal, onActivity) => raw.event.subscribe({ signal, onActivity })) + const events = SharedEvents.make((signal, onActivity) => raw.event.subscribe(undefined, { signal, onActivity })) return { ...raw, rpc: Object.assign(makeRpc(raw, events), raw.rpc), diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index cc87f73a9f13..f0dbfa0c6a92 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -203,6 +203,7 @@ import type { SkillListOutput, RpcCallInput, RpcCallOutput, + EventSubscribeInput, EventSubscribeOutput, PtyListInput, PtyListOutput, @@ -1811,9 +1812,16 @@ export function make(options: ClientOptions) { ), }, event: { - subscribe: (requestOptions?: RequestOptions): AsyncIterable => + subscribe: (input?: EventSubscribeInput, requestOptions?: RequestOptions): AsyncIterable => sse( - { method: "GET", path: `/api/event`, successStatus: 200, declaredStatuses: [400, 401], empty: false }, + { + method: "GET", + path: `/api/event`, + query: { types: input?.["types"] }, + successStatus: 200, + declaredStatuses: [400, 401], + empty: false, + }, requestOptions, ), }, diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 40d8c4e38c3d..f544bf8cfda7 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -3987,6 +3987,108 @@ export type RpcCallInput = { export type RpcCallOutput = RpcOutput +export type EventSubscribeInput = { + readonly types?: + | ReadonlyArray< + | "location.shutdown" + | "models-dev.refreshed" + | "credential.updated" + | "credential.switched" + | "integration.updated" + | "provider.updated" + | "model.updated" + | "agent.updated" + | "session.created" + | "session.agent.selected" + | "session.model.selected" + | "session.moved" + | "session.renamed" + | "session.metadata.updated" + | "session.permissions" + | "session.viewed" + | "session.archived" + | "session.unarchived" + | "session.usage.updated" + | "session.deleted" + | "session.forked" + | "session.inbox.delivered" + | "session.inbox.enqueued" + | "session.inbox.cancelled" + | "session.inbox.delivery.changed" + | "session.execution.started" + | "session.execution.succeeded" + | "session.execution.failed" + | "session.execution.interrupted" + | "session.instructions.updated" + | "session.synthetic" + | "session.skill.activated" + | "session.shell.started" + | "session.shell.ended" + | "session.step.started" + | "session.step.streamed" + | "session.step.ended" + | "session.step.failed" + | "session.text.started" + | "session.text.delta" + | "session.text.ended" + | "session.reasoning.started" + | "session.reasoning.delta" + | "session.reasoning.ended" + | "session.tool.input.started" + | "session.tool.input.delta" + | "session.tool.input.ended" + | "session.tool.called" + | "session.tool.progress" + | "session.tool.success" + | "session.tool.failed" + | "session.retry.scheduled" + | "session.compaction.started" + | "session.compaction.delta" + | "session.compaction.ended" + | "session.compaction.failed" + | "session.revert.staged" + | "session.revert.cleared" + | "session.revert.committed" + | "filesystem.changed" + | "reference.updated" + | "permission.asked" + | "permission.replied" + | "plugin.updated" + | "project.updated" + | "worktree.updated" + | "worktree.resolved" + | "command.updated" + | "config.updated" + | "skill.updated" + | "pty.created" + | "pty.updated" + | "pty.exited" + | "pty.deleted" + | "persistent-pty.added" + | "persistent-pty.removed" + | "shell.created" + | "shell.exited" + | "shell.deleted" + | "form.created" + | "form.replied" + | "form.cancelled" + | "websearch.updated" + | "session.status" + | "session.idle" + | "tui.prompt.append" + | "tui.command.execute" + | "tui.toast.show" + | "tui.session.select" + | "installation.updated" + | "installation.update-available" + | "vcs.branch.updated" + | "mcp.status.changed" + | "mcp.resources.changed" + | `${"rpc."}${string}` + > + | undefined +} + export type EventSubscribeOutput = V2Event export type PtyListInput = { readonly location?: { readonly directory?: string | undefined } | undefined } diff --git a/packages/client/src/shared-events.ts b/packages/client/src/shared-events.ts index f8e4ce919ed1..50097a3376a6 100644 --- a/packages/client/src/shared-events.ts +++ b/packages/client/src/shared-events.ts @@ -1,6 +1,14 @@ export * as SharedEvents from "./shared-events.js" -export type SubscribeOptions = { +/** The members of event union `A` whose type matches one of `Type`, including `rpc.*` events. */ +export type SelectedEvent = A extends unknown + ? [Extract] extends [never] + ? never + : A + : never + +export type SubscribeOptions = { + readonly types?: ReadonlyArray readonly signal?: AbortSignal /** Reports transport activity on the shared stream, including keepalive frames that carry no event. */ readonly onActivity?: () => void @@ -58,14 +66,21 @@ export function make( } return { - subscribe(options?: SubscribeOptions): AsyncIterable { + subscribe = Exclude>( + options?: SubscribeOptions, + ): AsyncIterable> { return { [Symbol.asyncIterator]() { - const pending: ReturnType>>[] = [] - const queued: A[] = [] + // server.connected always comes through so a filtered subscriber still sees reconnects and can resync. + type Item = A & SelectedEvent + const pending: ReturnType>>[] = [] + const queued: Item[] = [] let started = false let completion: Completion | undefined let connection: Connection | undefined + const types = options?.types?.length ? new Set(options.types) : undefined + const selected = (value: A): value is Item => + !types || value.type === "server.connected" || types.has(value.type) function finish(result: Completion, discard = true) { completion = result @@ -89,6 +104,7 @@ export function make( }, push(value) { if (completion) return + if (!selected(value)) return const request = pending.shift() if (request) { request.resolve({ done: false, value }) @@ -116,7 +132,7 @@ export function make( } return { - next(): Promise> { + next(): Promise> { const value = queued.shift() if (value) return Promise.resolve({ done: false, value }) if (completion) { @@ -127,7 +143,7 @@ export function make( abort() return Promise.resolve({ done: true, value: undefined }) } - const request = Promise.withResolvers>() + const request = Promise.withResolvers>() pending.push(request) if (!started) { started = true @@ -136,7 +152,7 @@ export function make( } return request.promise }, - return(): Promise> { + return(): Promise> { finish({}) return Promise.resolve({ done: true, value: undefined }) }, diff --git a/packages/client/test/shared-events.test.ts b/packages/client/test/shared-events.test.ts index bf3370be5e86..6c9540ce81a2 100644 --- a/packages/client/test/shared-events.test.ts +++ b/packages/client/test/shared-events.test.ts @@ -119,6 +119,20 @@ test("multiple consumers share one source and receive live native and RPC events await events.connections[0].closed }) +test("a types filter delivers only the requested types plus server.connected", async () => { + const events = source() + const shared = SharedEvents.make(events.connect) + const filtered = shared.subscribe({ types: ["session.updated"] })[Symbol.asyncIterator]() + const connected = filtered.next() + for (const type of ["server.connected", "permission.asked", "rpc.example.updated", "session.updated"]) + events.connections[0].push({ type }) + + expect(await connected).toEqual({ done: false, value: { type: "server.connected" } }) + expect(await filtered.next()).toEqual({ done: false, value: { type: "session.updated" } }) + await filtered.return!() + await events.connections[0].closed +}) + test("a slow consumer does not stall other subscribers", async () => { const events = source() const shared = SharedEvents.make(events.connect) diff --git a/packages/core/src/bus.ts b/packages/core/src/bus.ts index 189729362403..7b37335eaae3 100644 --- a/packages/core/src/bus.ts +++ b/packages/core/src/bus.ts @@ -198,7 +198,7 @@ export function configured(options?: Options) { const pubsub = { live: yield* PubSub.unbounded(), durable: new Map>>(), - typed: new Map>(), + typed: new Map>>(), } const projectors = new Map() const listeners = new Array() @@ -265,15 +265,6 @@ export function configured(options?: Options) { } }) - const getOrCreate = (definition: Event.Definition) => - Effect.gen(function* () { - const existing = pubsub.typed.get(definition.type) - if (existing) return existing - const created = yield* PubSub.unbounded() - pubsub.typed.set(definition.type, created) - return created - }) - yield* Effect.addFinalizer(() => Effect.gen(function* () { yield* PubSub.shutdown(pubsub.live) @@ -282,7 +273,11 @@ export function configured(options?: Options) { (pubsubs) => Effect.forEach(pubsubs, PubSub.shutdown, { discard: true }), { discard: true }, ) - yield* Effect.forEach(pubsub.typed.values(), PubSub.shutdown, { discard: true }) + yield* Effect.forEach( + pubsub.typed.values(), + (subscribers) => Effect.forEach(subscribers, PubSub.shutdown, { discard: true }), + { discard: true }, + ) }), ) @@ -505,9 +500,11 @@ export function configured(options?: Options) { (listener) => (isolateListeners ? observe(event, listener) : listener(event)), { discard: true }, ) - const typed = pubsub.typed.get(event.type) - if (typed) yield* PubSub.publish(typed, event) + // Type subscribers resume inline and may publish follow-up events, so live subscribers get this one first. yield* PubSub.publish(pubsub.live, event) + const typed = pubsub.typed.get(event.type) + if (typed) + yield* Effect.forEach(typed, (subscriber) => PubSub.publish(subscriber, event), { discard: true }) }) } @@ -758,11 +755,30 @@ export function configured(options?: Options) { ): Stream.Stream> function subscribe(input?: Event.Definition | readonly Event.Definition[]): Stream.Stream { if (input === undefined) return streamLive() - if (isDefinition(input)) { - return local(Stream.unwrap(getOrCreate(input).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub))))) - } - const types = new Set(input.map((definition) => definition.type)) - return streamLive().pipe(Stream.filter((event) => types.has(event.type))) + const types = new Set((isDefinition(input) ? [input] : input).map((definition) => definition.type)) + // One PubSub per subscription, registered under each type, keeps publish order across types. + return local( + Stream.unwrap( + Effect.acquireRelease( + PubSub.unbounded().pipe( + Effect.tap((subscriber) => + Effect.sync(() => { + for (const type of types) + pubsub.typed.set(type, (pubsub.typed.get(type) ?? new Set()).add(subscriber)) + }), + ), + ), + (subscriber) => + Effect.sync(() => { + for (const type of types) { + const subscribers = pubsub.typed.get(type) + subscribers?.delete(subscriber) + if (subscribers?.size === 0) pubsub.typed.delete(type) + } + }).pipe(Effect.andThen(PubSub.shutdown(subscriber))), + ).pipe(Effect.flatMap(PubSub.subscribe), Effect.map(Stream.fromSubscription)), + ), + ) } const streamLive = (): Stream.Stream => local(Stream.fromPubSub(pubsub.live)) diff --git a/packages/core/src/config/plugin/agent.ts b/packages/core/src/config/plugin/agent.ts index 6da638171dd2..e78ad1243bac 100644 --- a/packages/core/src/config/plugin/agent.ts +++ b/packages/core/src/config/plugin/agent.ts @@ -69,11 +69,9 @@ export const Plugin = define({ Stream.runForEach(notify), Effect.forkScoped({ startImmediately: true }), ) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), - Stream.runForEach(notify), - Effect.forkScoped({ startImmediately: true }), - ) + yield* ctx.event + .subscribe({ types: ["config.updated"] }) + .pipe(Stream.runForEach(notify), Effect.forkScoped({ startImmediately: true })) const updates = yield* PubSub.subscribe(changes) yield* Stream.fromSubscription(updates).pipe( Stream.debounce("100 millis"), diff --git a/packages/core/src/config/plugin/command.ts b/packages/core/src/config/plugin/command.ts index 5dc6c140acfe..9d739ac8c512 100644 --- a/packages/core/src/config/plugin/command.ts +++ b/packages/core/src/config/plugin/command.ts @@ -55,11 +55,9 @@ export const Plugin = define({ Stream.runForEach(notify), Effect.forkScoped({ startImmediately: true }), ) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), - Stream.runForEach(notify), - Effect.forkScoped({ startImmediately: true }), - ) + yield* ctx.event + .subscribe({ types: ["config.updated"] }) + .pipe(Stream.runForEach(notify), Effect.forkScoped({ startImmediately: true })) const updates = yield* PubSub.subscribe(changes) yield* Stream.fromSubscription(updates).pipe( Stream.debounce("100 millis"), diff --git a/packages/core/src/config/plugin/entry-observer.ts b/packages/core/src/config/plugin/entry-observer.ts index a62344014b8b..65e77f6b6d69 100644 --- a/packages/core/src/config/plugin/entry-observer.ts +++ b/packages/core/src/config/plugin/entry-observer.ts @@ -14,8 +14,7 @@ export const observe = Effect.fnUntraced(function* ( Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))), Effect.andThen(reload), ) - yield* event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* event.subscribe({ types: ["config.updated"] }).pipe( Stream.runForEach(() => refresh), Effect.forkScoped({ startImmediately: true }), ) diff --git a/packages/core/src/config/plugin/instruction.ts b/packages/core/src/config/plugin/instruction.ts index 406eaed10e62..d67a2239f358 100644 --- a/packages/core/src/config/plugin/instruction.ts +++ b/packages/core/src/config/plugin/instruction.ts @@ -217,8 +217,7 @@ export const Plugin = define({ Effect.forkScoped({ startImmediately: true }), ) // Subscribe before the first config read so an update in between still triggers a reload. - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* ctx.event.subscribe({ types: ["config.updated"] }).pipe( Stream.runForEach(() => reconfigure().pipe(Effect.andThen(refresh()), Effect.andThen(discovery.reload()))), Effect.forkScoped({ startImmediately: true }), ) diff --git a/packages/core/src/config/plugin/mcp.ts b/packages/core/src/config/plugin/mcp.ts index ff544c2ab1fb..8806276f3527 100644 --- a/packages/core/src/config/plugin/mcp.ts +++ b/packages/core/src/config/plugin/mcp.ts @@ -10,7 +10,7 @@ import { Mcp } from "../../mcp/index.js" export const Plugin = define({ id: "opencode.config.mcp", effect: Effect.fn(function* (ctx) { - yield* register(ctx.event.subscribe()) + yield* register(ctx.event.subscribe({ types: ["config.updated"] })) }), }) diff --git a/packages/core/src/config/plugin/skill.ts b/packages/core/src/config/plugin/skill.ts index be62b9c60c1f..3858570ffc01 100644 --- a/packages/core/src/config/plugin/skill.ts +++ b/packages/core/src/config/plugin/skill.ts @@ -178,8 +178,7 @@ export const Plugin = define({ yield* ctx.skill.transform((editor) => { for (const skill of loaded.skills) editor.add(skill) }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), + yield* ctx.event.subscribe({ types: ["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 aaade0ba1d5f..6477af57ae3a 100644 --- a/packages/core/src/plugin/host.ts +++ b/packages/core/src/plugin/host.ts @@ -44,7 +44,10 @@ type RpcEvent = Event.Payload & { readonly location: Location.Ref readonly data: Readonly> } -const isRpcEvent = (event: Event.Payload): event is RpcEvent => event.type.startsWith("rpc.") +const isRpcType = (type: string): type is RpcEvent["type"] => type.startsWith("rpc.") +const isRpcEvent = (event: Event.Payload): event is RpcEvent => isRpcType(event.type) +const isPublicEvent = (event: Event.Payload): event is EventManifest.ServerEvent | RpcEvent => + EventManifest.isServer(event) || isRpcEvent(event) export const make = Effect.fn("PluginHost.make")(function* ( plugin: Pick, pluginID: string = "test", @@ -258,15 +261,22 @@ export const make = Effect.fn("PluginHost.make")(function* ( activate: (input) => integration.connection.activate(input.credentialID), }, event: { - subscribe: () => - bus - .subscribe() - .pipe( - Stream.filter( - (event): event is EventManifest.ServerEvent | RpcEvent => - EventManifest.isServer(event) || isRpcEvent(event), - ), - ), + // The bus only delivers the requested types, which TypeScript cannot follow through the generic signature. + subscribe: ((input?: { readonly types?: ReadonlyArray }) => { + if (!input?.types?.length) return bus.subscribe().pipe(Stream.filter(isPublicEvent)) + const unknown = new Array() + const [first, ...rest] = input.types.flatMap((type) => { + if (isRpcType(type)) return [Rpc.eventDefinition(type)] + const definition = EventManifest.Server.get(type) + if (!definition) unknown.push(type) + return definition ? [definition] : [] + }) + const events = first ? bus.subscribe([first, ...rest]) : Stream.never + if (unknown.length === 0) return events + return Stream.unwrap( + Effect.logWarning("ignoring unknown event types", { pluginID, types: unknown }).pipe(Effect.as(events)), + ) + }) as Plugin.Context["event"]["subscribe"], }, experimental: { terminal: { diff --git a/packages/core/src/plugin/plan.ts b/packages/core/src/plugin/plan.ts index bf8cd4a316e3..63939deffa86 100644 --- a/packages/core/src/plugin/plan.ts +++ b/packages/core/src/plugin/plan.ts @@ -72,11 +72,7 @@ export const Plugin = define({ ) }) - yield* ctx.event.subscribe().pipe( - Stream.filter( - (event): event is SessionEvent.Created | SessionEvent.AgentSelected => - event.type === "session.created" || event.type === "session.agent.selected", - ), + yield* ctx.event.subscribe({ types: ["session.created", "session.agent.selected"] }).pipe( Stream.runForEach((event) => { const text = switchReminder(event, enterReminder) if (!text) return Effect.void diff --git a/packages/core/src/plugin/provider/local.ts b/packages/core/src/plugin/provider/local.ts index a9f906a4f077..3935461be2da 100644 --- a/packages/core/src/plugin/provider/local.ts +++ b/packages/core/src/plugin/provider/local.ts @@ -142,11 +142,9 @@ export function createLocalProviderPlugin(input: { yield* ctx.provider.reload() yield* refresh().pipe(Effect.ignore) }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), - Stream.runForEach(reload), - Effect.forkScoped({ startImmediately: true }), - ) + yield* ctx.event + .subscribe({ types: ["config.updated"] }) + .pipe(Stream.runForEach(reload), Effect.forkScoped({ startImmediately: true })) }), } satisfies PluginInternal.InternalPlugin) } diff --git a/packages/core/src/rpc.ts b/packages/core/src/rpc.ts index 41abed377462..134fe06b37e1 100644 --- a/packages/core/src/rpc.ts +++ b/packages/core/src/rpc.ts @@ -59,7 +59,7 @@ const layer = Layer.effect( const events = new Map( Object.entries(definition.events).map(([name, event]) => [ name, - { event, definition: eventDefinition(definition, name) }, + { event, definition: eventDefinition(eventType(definition, name)) }, ]), ) definitions.set(definition, events) @@ -192,8 +192,7 @@ function eventType ({ type, durability: "ephemeral" as const, durable: undefined, data })), diff --git a/packages/core/test/config/entry-observer.test.ts b/packages/core/test/config/entry-observer.test.ts index 03d5c6a57f5c..ceed25c627fa 100644 --- a/packages/core/test/config/entry-observer.test.ts +++ b/packages/core/test/config/entry-observer.test.ts @@ -5,6 +5,7 @@ import { Event } from "@opencode/schema/event" import { Deferred, Effect, PubSub, Ref, Stream } from "effect" import { ConfigEntryObserver } from "../../src/config/plugin/entry-observer" import { it } from "../lib/effect" +import { host } from "../plugin/host" describe("ConfigEntryObserver", () => { it.effect("closes the startup race and reloads later updates", () => @@ -16,10 +17,12 @@ describe("ConfigEntryObserver", () => { entries: () => Ref.get(current), changes: () => Stream.empty, }) - const event = { - subscribe: () => - Stream.unwrap(Ref.set(current, [document("raced")]).pipe(Effect.as(Stream.fromPubSub(updates)))), - } + const event = host({ + event: { + subscribe: () => + Stream.unwrap(Ref.set(current, [document("raced")]).pipe(Effect.as(Stream.fromPubSub(updates)))), + }, + }).event const loaded = yield* ConfigEntryObserver.observe( config, diff --git a/packages/core/test/plugin.test.ts b/packages/core/test/plugin.test.ts index 96723ea86a5f..cf08f84c428d 100644 --- a/packages/core/test/plugin.test.ts +++ b/packages/core/test/plugin.test.ts @@ -1,12 +1,15 @@ import { expect } from "bun:test" import path from "path" -import { Clock, Deferred, Effect } from "effect" +import { Clock, Deferred, Effect, Fiber, Stream } from "effect" import { TestClock } from "effect/testing" import { Command } from "@opencode/core/command" import { Bus } from "@opencode/core/bus" import { Credential } from "@opencode/core/credential" import { Integration } from "@opencode/core/integration" import { Plugin } from "@opencode/core/plugin" +import { PluginHost } from "@opencode/core/plugin/host" +import { Rpc } from "@opencode/core/rpc" +import { Event } from "@opencode/schema/config" import { PluginModule } from "@opencode/core/plugin/module" import { Session } from "@opencode/core/session" import { SessionSchema } from "@opencode/core/session/schema" @@ -644,3 +647,21 @@ it.effect("activates a saved credential through the context", () => expect(yield* integrations.connection.active(integrationID)).toMatchObject({ id: first.id, label: "First" }) }), ) + +it.effect("host event subscriptions deliver requested rpc types and ignore unknown types", () => + Effect.gen(function* () { + const bus = yield* Bus.Service + const host = yield* PluginHost.make(yield* Plugin.Service) + const fiber = yield* host.event + // @ts-expect-error untyped JavaScript plugins can still pass unknown type names + .subscribe({ types: ["rpc.test.ping", "test.unknown"] }) + .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped({ startImmediately: true })) + + yield* bus.publish(Event.Updated, {}) + yield* bus.publish(Rpc.eventDefinition("rpc.test.ping"), { value: 1 }) + + expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.type, event.data])).toEqual([ + ["rpc.test.ping", { value: 1 }], + ]) + }), +) diff --git a/packages/core/test/plugin/host.ts b/packages/core/test/plugin/host.ts index f0b8c54650c9..8fdb96d3bafe 100644 --- a/packages/core/test/plugin/host.ts +++ b/packages/core/test/plugin/host.ts @@ -11,8 +11,10 @@ import { AbsolutePath } from "@opencode/core/schema" import { WebSearch } from "@opencode/core/websearch" import { Effect, Stream } from "effect" -type Overrides = Partial> & { +type EventSource = ReturnType +type Overrides = Partial> & { readonly session?: Partial + readonly event?: { readonly subscribe: () => EventSource } } export function host(overrides: Overrides = {}): Plugin.Context { return { @@ -65,9 +67,7 @@ export function host(overrides: Overrides = {}): Plugin.Context { credential: overrides.credential ?? { activate: () => Effect.die("unused credential.activate"), }, - event: overrides.event ?? { - subscribe: () => Stream.empty, - }, + event: { subscribe: events(overrides.event?.subscribe ?? (() => Stream.empty)) }, experimental: overrides.experimental ?? { terminal: { read: () => Effect.die("unused experimental.terminal.read"), @@ -243,13 +243,15 @@ export function providerHost(providers: Provider.Interface): Plugin.Context["pro return { list: () => providers.available().pipe(Effect.map(located)), get: (input) => - providers.get(Provider.ID.make(input.providerID)).pipe( - Effect.flatMap((provider) => - provider === undefined - ? Effect.fail(new Error(`Provider not found: ${input.providerID}`)) - : Effect.succeed(located(provider)), + providers + .get(Provider.ID.make(input.providerID)) + .pipe( + Effect.flatMap((provider) => + provider === undefined + ? Effect.fail(new Error(`Provider not found: ${input.providerID}`)) + : Effect.succeed(located(provider)), + ), ), - ), reload: providers.reload, transform: (callback) => providers.transform((editor) => @@ -515,3 +517,10 @@ function located(data: A) { data, } } + +// Filter like the real host, so a fake source can emit any event type. +function events(source: () => EventSource) { + const subscribe = (input?: { readonly types?: ReadonlyArray }): EventSource => + source().pipe(Stream.filter((event) => !input?.types?.length || input.types.includes(event.type))) + return subscribe as Plugin.Context["event"]["subscribe"] +} diff --git a/packages/plugin-browser/src/connection.ts b/packages/plugin-browser/src/connection.ts index 3de3949cdfa9..c9e9a9c856bd 100644 --- a/packages/plugin-browser/src/connection.ts +++ b/packages/plugin-browser/src/connection.ts @@ -170,8 +170,7 @@ export const make = Effect.fn("BrowserConnection.make")(function* ( ), }) .pipe(Effect.orDie) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "session.deleted" || event.type === "session.moved"), + yield* ctx.event.subscribe({ types: ["session.deleted", "session.moved"] }).pipe( Stream.runForEach((event) => close(event.data.sessionID)), Effect.forkScoped({ startImmediately: true }), ) diff --git a/packages/plugin/src/effect/event.ts b/packages/plugin/src/effect/event.ts index d4ab10c0f4bf..6b751196aeb6 100644 --- a/packages/plugin/src/effect/event.ts +++ b/packages/plugin/src/effect/event.ts @@ -1,3 +1,10 @@ -import type { EventApi } from "@opencode/client/effect/api" +import type { EventSubscribeInput, EventSubscribeOutput, SelectedEvent } from "@opencode/client/effect/api" +import type { Stream } from "effect" -export interface EventDomain extends Pick, "subscribe"> {} +type EventType = NonNullable[number] + +export interface EventDomain { + readonly subscribe: (input?: { + readonly types?: ReadonlyArray + }) => Stream.Stream, unknown> +} diff --git a/packages/plugin/src/promise/adapter.ts b/packages/plugin/src/promise/adapter.ts index eebfadaf8122..a9dc64640cf1 100644 --- a/packages/plugin/src/promise/adapter.ts +++ b/packages/plugin/src/promise/adapter.ts @@ -1,6 +1,7 @@ import { Tool } from "@opencode/schema/tool" import type { Rpc } from "@opencode/schema/rpc" -import type { RpcCallOptions, RpcEventPayload } from "@opencode/client/promise/api" +import type { OpenCodeEvent } from "@opencode/client" +import type { RpcCallOptions, RpcEventPayload, SelectedEvent, SubscribeOptions } from "@opencode/client/promise/api" import { Effect, Schema, SchemaAST, Stream } from "effect" import type { Scope } from "effect" import { HttpApiEndpoint, HttpApiSchema } from "effect/http-api" @@ -14,7 +15,6 @@ import type { RpcDomain, RpcHandlers } from "./rpc.js" type HostRegistration = { readonly dispose: Effect.Effect } type Registration = { readonly dispose: () => Promise } type PromiseContext = Parameters[0] -type PromiseEvent = ReturnType extends AsyncIterable ? Event : never type HostRpc = Parameters[0]["effect"]>[0]["rpc"] type StreamAdapter = ( stream: Stream.Stream, @@ -329,11 +329,13 @@ export function fromPromise(plugin: Plugin) { activate: adaptApiMethod(CredentialEndpoints["credential.activate"], host.credential.activate), }, event: { - subscribe: (options) => + subscribe: >( + options?: SubscribeOptions, + ) => streams( - host.event.subscribe().pipe( + host.event.subscribe({ types: options?.types }).pipe( Stream.mapEffect((event) => encodeOpenCodeEvent(event)), - Stream.map((event) => event as PromiseEvent), + Stream.map((event) => event as SelectedEvent), ), options, ), diff --git a/packages/plugin/src/promise/event.ts b/packages/plugin/src/promise/event.ts index 74ac6e5fde2e..1bf00a8b5c63 100644 --- a/packages/plugin/src/promise/event.ts +++ b/packages/plugin/src/promise/event.ts @@ -1,3 +1,10 @@ -import type { EventApi } from "@opencode/client/promise/api" +import type { OpenCodeEvent } from "@opencode/client" +import type { SelectedEvent, SubscribeOptions } from "@opencode/client/promise/api" -export interface EventDomain extends Pick {} +type EventType = Exclude + +export interface EventDomain { + readonly subscribe: ( + options?: SubscribeOptions, + ) => AsyncIterable> +} diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index 139662c8f45c..bb2492ed3a9b 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -10099,7 +10099,242 @@ "get": { "tags": ["event"], "operationId": "event.subscribe", - "parameters": [], + "parameters": [ + { + "name": "types", + "in": "query", + "schema": { + "anyOf": [ + { + "anyOf": [ + { + "anyOf": [ + { + "type": "string", + "enum": [ + "location.shutdown", + "models-dev.refreshed", + "credential.updated", + "credential.switched", + "integration.updated", + "provider.updated", + "model.updated", + "agent.updated", + "session.created", + "session.agent.selected", + "session.model.selected", + "session.moved", + "session.renamed", + "session.metadata.updated", + "session.permissions", + "session.viewed", + "session.archived", + "session.unarchived", + "session.usage.updated", + "session.deleted", + "session.forked", + "session.inbox.delivered", + "session.inbox.enqueued", + "session.inbox.cancelled", + "session.inbox.delivery.changed", + "session.execution.started", + "session.execution.succeeded", + "session.execution.failed", + "session.execution.interrupted", + "session.instructions.updated", + "session.synthetic", + "session.skill.activated", + "session.shell.started", + "session.shell.ended", + "session.step.started", + "session.step.streamed", + "session.step.ended", + "session.step.failed", + "session.text.started", + "session.text.delta", + "session.text.ended", + "session.reasoning.started", + "session.reasoning.delta", + "session.reasoning.ended", + "session.tool.input.started", + "session.tool.input.delta", + "session.tool.input.ended", + "session.tool.called", + "session.tool.progress", + "session.tool.success", + "session.tool.failed", + "session.retry.scheduled", + "session.compaction.started", + "session.compaction.delta", + "session.compaction.ended", + "session.compaction.failed", + "session.revert.staged", + "session.revert.cleared", + "session.revert.committed", + "filesystem.changed", + "reference.updated", + "permission.asked", + "permission.replied", + "plugin.updated", + "project.updated", + "worktree.updated", + "worktree.resolved", + "command.updated", + "config.updated", + "skill.updated", + "pty.created", + "pty.updated", + "pty.exited", + "pty.deleted", + "persistent-pty.added", + "persistent-pty.removed", + "shell.created", + "shell.exited", + "shell.deleted", + "form.created", + "form.replied", + "form.cancelled", + "websearch.updated", + "session.status", + "session.idle", + "tui.prompt.append", + "tui.command.execute", + "tui.toast.show", + "tui.session.select", + "installation.updated", + "installation.update-available", + "vcs.branch.updated", + "mcp.status.changed", + "mcp.resources.changed" + ] + }, + { + "type": "string", + "pattern": "^rpc\\.[\\s\\S]*?$" + } + ] + }, + { + "type": "array", + "items": { + "anyOf": [ + { + "type": "string", + "enum": [ + "location.shutdown", + "models-dev.refreshed", + "credential.updated", + "credential.switched", + "integration.updated", + "provider.updated", + "model.updated", + "agent.updated", + "session.created", + "session.agent.selected", + "session.model.selected", + "session.moved", + "session.renamed", + "session.metadata.updated", + "session.permissions", + "session.viewed", + "session.archived", + "session.unarchived", + "session.usage.updated", + "session.deleted", + "session.forked", + "session.inbox.delivered", + "session.inbox.enqueued", + "session.inbox.cancelled", + "session.inbox.delivery.changed", + "session.execution.started", + "session.execution.succeeded", + "session.execution.failed", + "session.execution.interrupted", + "session.instructions.updated", + "session.synthetic", + "session.skill.activated", + "session.shell.started", + "session.shell.ended", + "session.step.started", + "session.step.streamed", + "session.step.ended", + "session.step.failed", + "session.text.started", + "session.text.delta", + "session.text.ended", + "session.reasoning.started", + "session.reasoning.delta", + "session.reasoning.ended", + "session.tool.input.started", + "session.tool.input.delta", + "session.tool.input.ended", + "session.tool.called", + "session.tool.progress", + "session.tool.success", + "session.tool.failed", + "session.retry.scheduled", + "session.compaction.started", + "session.compaction.delta", + "session.compaction.ended", + "session.compaction.failed", + "session.revert.staged", + "session.revert.cleared", + "session.revert.committed", + "filesystem.changed", + "reference.updated", + "permission.asked", + "permission.replied", + "plugin.updated", + "project.updated", + "worktree.updated", + "worktree.resolved", + "command.updated", + "config.updated", + "skill.updated", + "pty.created", + "pty.updated", + "pty.exited", + "pty.deleted", + "persistent-pty.added", + "persistent-pty.removed", + "shell.created", + "shell.exited", + "shell.deleted", + "form.created", + "form.replied", + "form.cancelled", + "websearch.updated", + "session.status", + "session.idle", + "tui.prompt.append", + "tui.command.execute", + "tui.toast.show", + "tui.session.select", + "installation.updated", + "installation.update-available", + "vcs.branch.updated", + "mcp.status.changed", + "mcp.resources.changed" + ] + }, + { + "type": "string", + "pattern": "^rpc\\.[\\s\\S]*?$" + } + ] + } + } + ] + }, + { + "type": "null" + } + ], + "description": "Only send events of these types, plus the initial server.connected event. When omitted, all events are sent." + }, + "required": false + } + ], "security": [], "responses": { "200": { diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index 288709c44090..3d8423dd90f9 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -35,6 +35,23 @@ const schema = >(definitions ]), ]).annotate({ identifier: "V2Event" }) +const query = >(definitions: Definitions) => + Schema.Struct({ + types: Schema.optional( + Schema.ArrayEnsure( + Schema.Union([ + Schema.Literals( + definitions.map((definition) => definition.type) as ReadonlyArray, + ), + rpcEvent.fields.type, + ]), + ), + ).annotate({ + description: + "Only send events of these types, plus the initial server.connected event. When omitted, all events are sent.", + }), + }).annotate({ identifier: "EventSubscribeQuery" }) + const make = >(definitions: Definitions) => { const EventSchema = schema(definitions) return { @@ -42,6 +59,7 @@ const make = >(definitions: group: HttpApiGroup.make("server.event") .add( HttpApiEndpoint.get("event.subscribe", "/api/event", { + query: query(definitions), success: HttpApiSchema.StreamSse({ data: EventSchema }), }).annotateMerge( OpenApi.annotations({ diff --git a/packages/server/src/event-feed.ts b/packages/server/src/event-feed.ts index f22c393fe296..19cf5f014c1f 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -21,7 +21,7 @@ export class EncodingError extends Schema.TaggedError()("EventFee export type Error = SubscriberOverflowError | EncodingError export interface Interface { - readonly subscribe: Effect.Effect, never, Scope.Scope> + readonly subscribe: (types?: ReadonlyArray) => Effect.Effect, never, Scope.Scope> } export class Service extends Context.Service()("@opencode/server/EventFeed") {} @@ -36,32 +36,33 @@ export const make = Effect.fn("EventFeed.make")(function* ( ) { const capacity = options?.capacity ?? SubscriberCapacity const render = options?.encode ?? frame - const subscribers = new Set>() + const subscribers = new Map, ReadonlySet | undefined>() const fail = (error: Error) => Effect.sync(() => { - const current = Array.from(subscribers) + const current = Array.from(subscribers.keys()) subscribers.clear() for (const subscriber of current) Queue.failCauseUnsafe(subscriber, Cause.fail(error)) }) const publish = Effect.fnUntraced(function* (event: Event.Payload) { if (!isOpenCodeEvent(event)) return - if (subscribers.size === 0) return - const encoded = yield* Effect.try({ - try: () => render(event), - catch: (cause) => new EncodingError({ eventID: event.id, eventType: event.type, cause }), - }).pipe( - Effect.catch((error) => - Effect.logError("Failed to encode public event", { - eventID: error.eventID, - eventType: error.eventType, - cause: error.cause, - }).pipe(Effect.andThen(fail(error)), Effect.as(undefined)), - ), - ) - if (encoded === undefined) return - for (const subscriber of subscribers) { + let encoded: string | undefined + for (const [subscriber, types] of subscribers) { + if (types && !types.has(event.type)) continue + encoded ??= yield* Effect.try({ + try: () => render(event), + catch: (cause) => new EncodingError({ eventID: event.id, eventType: event.type, cause }), + }).pipe( + Effect.catch((error) => + Effect.logError("Failed to encode public event", { + eventID: error.eventID, + eventType: error.eventType, + cause: error.cause, + }).pipe(Effect.andThen(fail(error)), Effect.as(undefined)), + ), + ) + if (encoded === undefined) return if (Queue.offerUnsafe(subscriber, encoded)) continue subscribers.delete(subscriber) Queue.failCauseUnsafe(subscriber, Cause.fail(new SubscriberOverflowError({ capacity }))) @@ -72,11 +73,14 @@ export const make = Effect.fn("EventFeed.make")(function* ( yield* Effect.addFinalizer(() => unsubscribe) return Service.of({ - subscribe: Effect.acquireRelease( - Queue.dropping(capacity).pipe(Effect.tap((queue) => Effect.sync(() => subscribers.add(queue)))), - (queue) => - Effect.sync(() => subscribers.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid), - ).pipe(Effect.map(Stream.fromQueue)), + subscribe: (types) => + Effect.acquireRelease( + Queue.dropping(capacity).pipe( + Effect.tap((queue) => Effect.sync(() => subscribers.set(queue, types?.length ? new Set(types) : undefined))), + ), + (queue) => + Effect.sync(() => subscribers.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid), + ).pipe(Effect.map(Stream.fromQueue)), }) }) diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index 987f2879d3aa..491cb53cf58c 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -8,7 +8,7 @@ import { EventFeed } from "../event-feed" export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) => Effect.gen(function* () { const feed = yield* EventFeed.Service - return handlers.handleRaw("event.subscribe", () => + return handlers.handleRaw("event.subscribe", (ctx) => Effect.gen(function* () { const connected = { id: Event.ID.create(), @@ -16,7 +16,9 @@ export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) data: {}, } as const const output = Stream.unwrap( - feed.subscribe.pipe(Effect.map((live) => Stream.make(EventFeed.frame(connected)).pipe(Stream.concat(live)))), + feed + .subscribe(ctx.query.types) + .pipe(Effect.map((live) => Stream.make(EventFeed.frame(connected)).pipe(Stream.concat(live)))), ) const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n")) return HttpServerResponse.stream( diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index 861546f2d769..9a367256e2fa 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -54,8 +54,8 @@ describe("EventFeed", () => { return event.type }, }) - const first = yield* feed.subscribe - const second = yield* feed.subscribe + const first = yield* feed.subscribe() + const second = yield* feed.subscribe() const left = yield* first.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) const right = yield* second.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) @@ -69,6 +69,31 @@ describe("EventFeed", () => { }), ) + it.effect("delivers only the requested event types to filtered subscribers", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (event) => event.type }) + const filtered = yield* feed.subscribe([Credential.Event.Switched.type]) + const all = yield* feed.subscribe() + const left = yield* filtered.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const right = yield* all.pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) + + yield* source.publish(event("ignored")) + yield* source.publish({ + id: Event.ID.create(), + created: Date.now(), + type: Credential.Event.Switched.type, + data: { + credentialID: Credential.ID.make("cred_test"), + integrationID: IntegrationID.make("openai"), + }, + }) + + expect(Array.from(yield* Fiber.join(left))).toEqual([Credential.Event.Switched.type]) + expect(Array.from(yield* Fiber.join(right))).toEqual([Agent.Event.Updated.type, Credential.Event.Switched.type]) + }), + ) + it.effect("fails only the subscriber that exceeds its lag capacity", () => Effect.gen(function* () { const source = makeSource() @@ -76,8 +101,8 @@ describe("EventFeed", () => { capacity: 1, encode: (event) => event.id, }) - const slow = yield* feed.subscribe - const fast = yield* feed.subscribe + const slow = yield* feed.subscribe() + const fast = yield* feed.subscribe() const first = yield* Deferred.make() const second = yield* Deferred.make() const received = new Array() @@ -117,7 +142,7 @@ describe("EventFeed", () => { Effect.gen(function* () { const source = makeSource() const feed = yield* EventFeed.make(source.observe, { encode: (event) => event.type }) - const stream = yield* feed.subscribe + const stream = yield* feed.subscribe() const received = yield* stream.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) yield* source.publish({ @@ -138,7 +163,7 @@ describe("EventFeed", () => { Effect.gen(function* () { const source = makeSource() const feed = yield* EventFeed.make(source.observe, { capacity: 1, encode: (event) => event.type }) - const stream = yield* feed.subscribe + const stream = yield* feed.subscribe() yield* source.publish(internal("one")) yield* source.publish(internal("two")) @@ -157,13 +182,13 @@ describe("EventFeed", () => { return event.id }, }) - const current = yield* feed.subscribe + const current = yield* feed.subscribe() const failed = yield* current.pipe(Stream.runCollect, Effect.exit, Effect.forkScoped) yield* source.publish(event("bad")) const exit = yield* Fiber.join(failed) - const next = yield* feed.subscribe + const next = yield* feed.subscribe() const received = yield* next.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) yield* source.publish(event("good")) diff --git a/packages/tui/test/mini/stream-v2.transport.test.ts b/packages/tui/test/mini/stream-v2.transport.test.ts index fff67730e802..907b08f7dedb 100644 --- a/packages/tui/test/mini/stream-v2.transport.test.ts +++ b/packages/tui/test/mini/stream-v2.transport.test.ts @@ -157,7 +157,9 @@ function sdk(input: { }) { const client = OpenCode.make({ baseUrl: "https://opencode.test" }) let subscription = 0 - spyOn(client.event, "subscribe").mockImplementation(() => input.streams[subscription++]?.stream ?? feed().stream) + spyOn(client.event, "subscribe").mockImplementation( + (() => input.streams[subscription++]?.stream ?? feed().stream) as OpenCodeClient["event"]["subscribe"], + ) spyOn(client.message, "list").mockImplementation((request) => ok({ data: input.messages?.[request.sessionID] ?? [ diff --git a/services/www/src/docs/content/build/client/index.mdx b/services/www/src/docs/content/build/client/index.mdx index 5634a90868e8..f1b84315df60 100644 --- a/services/www/src/docs/content/build/client/index.mdx +++ b/services/www/src/docs/content/build/client/index.mdx @@ -75,6 +75,16 @@ Subscriptions are live-only, with no replay or automatic reconnection. A source failure ends current subscriptions; subscribe again after recovery. A late native subscriber receives the current `server.connected` marker, not past business events. +Pass `types` to receive only some event types. The events are typed to match, and +`server.connected` always comes through so a filtered subscriber still knows when to resync. +Filtering happens on the shared connection, inside the client. + +```ts +for await (const event of client.event.subscribe({ types: ["config.updated"] })) { + // event.type is "config.updated" | "server.connected" +} +``` + ## Plugin RPC Import a plugin's shared contract and pass it to `client.rpc`: diff --git a/services/www/src/docs/content/build/plugins/effect.mdx b/services/www/src/docs/content/build/plugins/effect.mdx index 98b4c48db207..8abceb6ff3b3 100644 --- a/services/www/src/docs/content/build/plugins/effect.mdx +++ b/services/www/src/docs/content/build/plugins/effect.mdx @@ -1222,14 +1222,13 @@ effect: (ctx) => }), ``` -Use Stream operators to select and process event types. +Pass `types` to receive only the event types you need. The stream is typed to match, so no type guard is needed. ```ts effect: (ctx) => Effect.gen(function* () { const event = ctx.event - yield* event.subscribe().pipe( - Stream.filter((item) => item.type === "config.updated"), + yield* event.subscribe({ types: ["config.updated"] }).pipe( Stream.runForEach(() => Effect.logInfo("configuration changed")), Effect.forkScoped, ) @@ -1239,14 +1238,13 @@ effect: (ctx) => Schema: [`V2EventEncoded`](/api#schema-V2EventEncoded). ```ts -type EventSubscribeOutput = OpenCodeEvent -type EventSubscribeOperation = () => Stream.Stream +type EventType = Exclude -interface EventApi { - readonly subscribe: EventSubscribeOperation +interface EventDomain { + readonly subscribe: (input?: { + readonly types?: ReadonlyArray + }) => Stream.Stream, unknown> } - -interface EventDomain extends Pick, "subscribe"> {} ``` ## Hooks diff --git a/services/www/src/docs/content/build/plugins/index.mdx b/services/www/src/docs/content/build/plugins/index.mdx index 4683395efb36..18a054890bac 100644 --- a/services/www/src/docs/content/build/plugins/index.mdx +++ b/services/www/src/docs/content/build/plugins/index.mdx @@ -1329,7 +1329,7 @@ interface WebSearchEditor { ### Events -Subscribe to the connected server's public event stream. Abort the stream during plugin cleanup. +Subscribe to the connected server's public event stream. Abort the stream during plugin cleanup. Pass `types`, for example `ctx.event.subscribe({ types: ["config.updated"] })`, to receive only those event types; the events are typed to match. Plugins run inside the server, so unlike SDK clients they never receive `server.connected`. Unknown type names are ignored with a warning in the server log. ```ts const controller = new AbortController() @@ -1346,8 +1346,13 @@ return () => controller.abort() Schema: [`V2EventEncoded`](/api#schema-V2EventEncoded) ```ts +type EventType = Exclude + interface EventContext { - subscribe(options?: { signal?: AbortSignal }): AsyncIterable + subscribe(options?: { + types?: Type[] + signal?: AbortSignal + }): AsyncIterable> } ```