From 95080c42f69af6b4706065265bbdee53a0594e01 Mon Sep 17 00:00:00 2001 From: Michael Blakeley Date: Tue, 1 Sep 2026 15:52:18 -0700 Subject: [PATCH 1/2] feat(plugin): expose session forms, session list, and global event stream --- packages/core/src/plugin/host.ts | 43 +++++++++++++++++- packages/core/src/session.ts | 44 +++++++++++++++++++ packages/core/test/config/agent.test.ts | 2 +- packages/core/test/config/command.test.ts | 4 +- packages/core/test/config/compaction.test.ts | 4 +- .../core/test/config/entry-observer.test.ts | 1 + packages/core/test/filesystem/watcher.test.ts | 2 +- packages/core/test/plugin/host.ts | 9 ++++ packages/core/test/plugin/plan.test.ts | 1 + packages/plugin/src/effect/event.ts | 12 ++++- packages/plugin/src/effect/session.ts | 19 +++++++- packages/plugin/src/promise/adapter.ts | 17 +++++++ packages/plugin/src/promise/event.ts | 11 ++++- packages/plugin/src/promise/session.ts | 28 ++++++++++++ packages/server/src/process.ts | 41 +++++++++++++++++ 15 files changed, 228 insertions(+), 10 deletions(-) diff --git a/packages/core/src/plugin/host.ts b/packages/core/src/plugin/host.ts index 0d128bdcb66d..c7499e2abc92 100644 --- a/packages/core/src/plugin/host.ts +++ b/packages/core/src/plugin/host.ts @@ -6,7 +6,7 @@ import { EventManifest } from "@opencode-ai/schema/event-manifest" import type { Event } from "@opencode-ai/schema/event" import { ServerConfig } from "@opencode-ai/schema/mcp" import { App } from "../app.js" -import { Effect, Schema, Stream } from "effect" +import { Effect, Queue, Schema, Stream } from "effect" import { Agent } from "../agent.js" import { AISDK } from "../aisdk.js" import { Catalog } from "../catalog.js" @@ -32,6 +32,7 @@ import { Vcs } from "../vcs.js" import { WebSearch } from "../websearch.js" import { Generate } from "../generate.js" import { Permission } from "../permission.js" +import { Form } from "../form.js" import { PluginHooks } from "./hooks.js" import type { Interface } from "../plugin.js" import { LayerNode } from "@opencode-ai/util/effect/layer-node" @@ -43,6 +44,14 @@ type RpcEvent = Event.Payload & { readonly data: Readonly> } const isRpcEvent = (event: Event.Payload): event is RpcEvent => event.type.startsWith("rpc.") + +// Backpressure cap for the global event bridge. Dropping the tail (rather than +// blocking the bus) is intentional: the bridge is a loss-tolerant observer hub +// for always-on plugins (e.g. Telegram), and a subscriber that can't keep up +// must not stall event publication for every other consumer. 4096 events is +// comfortably above any realistic burst without buffering unboundedly. +const GLOBAL_EVENT_QUEUE_CAPACITY = 4096 + export const make = Effect.fn("PluginHost.make")(function* ( plugin: Pick, pluginID: string = "test", @@ -235,6 +244,24 @@ export const make = Effect.fn("PluginHost.make")(function* ( EventManifest.isServer(event) || isRpcEvent(event), ), ), + /** + * Global (location-unfiltered) event stream. `ctx.event.subscribe()` is + * scoped to the plugin's ambient Location, so a cross-location observer + * (e.g. a Telegram bot that /cd's between projects) would silently miss + * events for sessions in other directories. This variant feeds from the + * bus's unfiltered listener channel and delivers every server event. + */ + subscribeGlobal: () => + Stream.scoped( + Stream.unwrap( + Effect.gen(function* () { + const queue = yield* Queue.dropping(GLOBAL_EVENT_QUEUE_CAPACITY) + const unsubscribe = yield* bus.listen((event) => Queue.offer(queue, event).pipe(Effect.asVoid)) + yield* Effect.addFinalizer((_exit) => unsubscribe) + return Stream.fromQueue(queue).pipe(Stream.filter(EventManifest.isServer)) + }), + ), + ), }, experimental: { terminal: { @@ -462,6 +489,13 @@ export const make = Effect.fn("PluginHost.make")(function* ( }, session: { hook: (name, callback, options) => hooks.register("session", name, callback, options), + list: (input) => + sessions.list({ + ...(input?.directory === undefined ? {} : { directory: AbsolutePath.make(input.directory) }), + search: input?.search, + limit: input?.limit, + order: input?.order, + }), create: (input) => sessions.create({ id: input?.id, @@ -486,6 +520,13 @@ export const make = Effect.fn("PluginHost.make")(function* ( .pipe(Effect.map((interrupted) => ({ interrupted }))), wait: (input) => sessions.wait(input.sessionID), context: (input) => sessions.context(input.sessionID), + form: { + list: (input: { sessionID: string }) => sessions.form.list({ sessionID: input.sessionID }), + get: (input: { sessionID: string; formID: Form.ID }) => sessions.form.get(input), + state: (input: { sessionID: string; formID: Form.ID }) => sessions.form.state(input), + reply: (input: { sessionID: string; formID: Form.ID; answer: Form.Answer }) => sessions.form.reply(input), + cancel: (input: { sessionID: string; formID: Form.ID }) => sessions.form.cancel(input), + }, }, } return context diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index df1e4b223684..c340310e8e13 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -11,6 +11,7 @@ import { Location } from "./location.js" import { SessionMessage } from "./session/message.js" import { PromptInput } from "@opencode-ai/schema/prompt-input" import { Bus } from "./bus.js" +import { Form } from "./form.js" import { Instance } from "./instance/service.js" import { Database } from "./database/database.js" import { SessionProjector } from "./session/projector.js" @@ -220,6 +221,26 @@ export interface Interface { readonly clear: (sessionID: SessionSchema.ID) => Effect.Effect readonly commit: (sessionID: SessionSchema.ID) => Effect.Effect } + readonly form: { + readonly list: (input: { readonly sessionID: string }) => Effect.Effect + readonly get: (input: { + readonly sessionID: string + readonly formID: Form.ID + }) => Effect.Effect + readonly state: (input: { + readonly sessionID: string + readonly formID: Form.ID + }) => Effect.Effect + readonly reply: (input: { + readonly sessionID: string + readonly formID: Form.ID + readonly answer: Form.Answer + }) => Effect.Effect + readonly cancel: (input: { + readonly sessionID: string + readonly formID: Form.ID + }) => Effect.Effect + } } export class Service extends Context.Service()("@opencode/Session") {} @@ -246,6 +267,22 @@ const layer = Layer.effect( const admission = yield* SessionInbox.Service const isDurableSessionEvent = Schema.is(SessionEvent.Durable) + // Forms are created by tools running under the session's instance, so the + // ambient Form.Service here is NOT where they live. Resolve the session's + // location and route through that location's Form.Service instead. + const formFor = ( + sessionID: string, + run: (form: Form.Interface) => Effect.Effect, + ) => + store.get(SessionSchema.ID.descending(sessionID)).pipe( + Effect.flatMap( + (session): Effect.Effect => + session === undefined + ? Effect.fail(new NotFoundError({ sessionID: SessionSchema.ID.descending(sessionID) })) + : instances.provide(session)(Form.Service.use(run)), + ), + ) + const result = Service.of({ create: Effect.fn("Session.create")(function* (input) { const sessionID = input.id ?? SessionSchema.ID.create() @@ -478,6 +515,13 @@ const layer = Layer.effect( clear: (sessionID) => sessions.forSession(sessionID).revert.clear(), commit: (sessionID) => sessions.forSession(sessionID).revert.commit(), }, + form: { + list: (input) => formFor(input.sessionID, (f) => f.list({ sessionID: input.sessionID })), + get: (input) => formFor(input.sessionID, (f) => f.get(input.formID)), + state: (input) => formFor(input.sessionID, (f) => f.state(input.formID)), + reply: (input) => formFor(input.sessionID, (f) => f.reply({ id: input.formID, answer: input.answer })), + cancel: (input) => formFor(input.sessionID, (f) => f.cancel(input.formID)), + }, }) return result diff --git a/packages/core/test/config/agent.test.ts b/packages/core/test/config/agent.test.ts index 7dcd90773e6f..6b23f8635ae9 100644 --- a/packages/core/test/config/agent.test.ts +++ b/packages/core/test/config/agent.test.ts @@ -513,7 +513,7 @@ Use native v2 fields.`, ...agentHost(agents), reload: () => agents.reload().pipe(Effect.tap(() => Effect.sync(() => reloads++))), }, - event: { subscribe: () => bus.subscribe(Event.Updated) }, + event: { subscribe: () => bus.subscribe(Event.Updated), subscribeGlobal: () => Stream.empty }, }), ) diff --git a/packages/core/test/config/command.test.ts b/packages/core/test/config/command.test.ts index c8764ead26af..fbff73bb2546 100644 --- a/packages/core/test/config/command.test.ts +++ b/packages/core/test/config/command.test.ts @@ -85,7 +85,7 @@ Review files`, transform: command.transform, reload: command.reload, }, - event: { subscribe: () => Stream.fromPubSub(updates) }, + event: { subscribe: () => Stream.fromPubSub(updates), subscribeGlobal: () => Stream.empty }, session: { prompt: (input) => Effect.sync(() => { @@ -257,7 +257,7 @@ Review files`, transform: command.transform, reload: () => command.reload().pipe(Effect.tap(() => Effect.sync(() => reloads++))), }, - event: { subscribe: () => bus.subscribe(Event.Updated) }, + event: { subscribe: () => bus.subscribe(Event.Updated), subscribeGlobal: () => Stream.empty }, }), ) diff --git a/packages/core/test/config/compaction.test.ts b/packages/core/test/config/compaction.test.ts index 16950880c333..80bd62af5013 100644 --- a/packages/core/test/config/compaction.test.ts +++ b/packages/core/test/config/compaction.test.ts @@ -71,7 +71,9 @@ describe("ConfigCompactionPlugin.Plugin", () => { }), }), ]) - yield* ConfigCompactionPlugin.Plugin.effect(host({ event: { subscribe: () => bus.subscribe(Event.Updated) } })) + yield* ConfigCompactionPlugin.Plugin.effect( + host({ event: { subscribe: () => bus.subscribe(Event.Updated), subscribeGlobal: () => Stream.empty } }), + ) expect(compaction.required(nearInput)).toBe(false) const started = yield* bus diff --git a/packages/core/test/config/entry-observer.test.ts b/packages/core/test/config/entry-observer.test.ts index 42ff6354c4f4..87a3eeeee7eb 100644 --- a/packages/core/test/config/entry-observer.test.ts +++ b/packages/core/test/config/entry-observer.test.ts @@ -19,6 +19,7 @@ describe("ConfigEntryObserver", () => { const event = { subscribe: () => Stream.unwrap(Ref.set(current, [document("raced")]).pipe(Effect.as(Stream.fromPubSub(updates)))), + subscribeGlobal: () => Stream.empty, } const loaded = yield* ConfigEntryObserver.observe( diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index 7b73020eddaa..35786bf92cd2 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -267,7 +267,7 @@ describe("LocationWatcher subscriptions", () => { entries.current = [new Document({ type: "document", info: new Info({ watcher: { ignore: [".git"] } }) })] yield* ConfigLocationWatcherPlugin.Plugin.effect( - host({ event: { subscribe: () => bus.subscribe(Event.Updated) } }), + host({ event: { subscribe: () => bus.subscribe(Event.Updated), subscribeGlobal: () => Stream.empty } }), ) yield* Effect.sync(() => counts.active).pipe( Effect.filterOrFail((count) => count === 0), diff --git a/packages/core/test/plugin/host.ts b/packages/core/test/plugin/host.ts index 586b23c06c9e..d201cc32f1fb 100644 --- a/packages/core/test/plugin/host.ts +++ b/packages/core/test/plugin/host.ts @@ -65,6 +65,7 @@ export function host(overrides: Overrides = {}): Plugin.Context { }, event: overrides.event ?? { subscribe: () => Stream.empty, + subscribeGlobal: () => Stream.empty, }, experimental: overrides.experimental ?? { terminal: { @@ -153,6 +154,7 @@ export function host(overrides: Overrides = {}): Plugin.Context { }, session: { hook: overrides.session?.hook ?? (() => Effect.die("unused session.hook")), + list: overrides.session?.list ?? (() => Effect.die("unused session.list")), create: overrides.session?.create ?? (() => Effect.die("unused session.create")), get: overrides.session?.get ?? (() => Effect.die("unused session.get")), switchAgent: overrides.session?.switchAgent ?? (() => Effect.die("unused session.switchAgent")), @@ -166,6 +168,13 @@ export function host(overrides: Overrides = {}): Plugin.Context { interrupt: overrides.session?.interrupt ?? (() => Effect.die("unused session.interrupt")), wait: overrides.session?.wait ?? (() => Effect.die("unused session.wait")), context: overrides.session?.context ?? (() => Effect.die("unused session.context")), + form: overrides.session?.form ?? { + list: () => Effect.die("unused session.form.list"), + get: () => Effect.die("unused session.form.get"), + state: () => Effect.die("unused session.form.state"), + reply: () => Effect.die("unused session.form.reply"), + cancel: () => Effect.die("unused session.form.cancel"), + }, }, } } diff --git a/packages/core/test/plugin/plan.test.ts b/packages/core/test/plugin/plan.test.ts index 25b14394b2b0..3e851df8e8a1 100644 --- a/packages/core/test/plugin/plan.test.ts +++ b/packages/core/test/plugin/plan.test.ts @@ -84,6 +84,7 @@ const run = Effect.fnUntraced(function* (events: ReadonlyArray Stream.fromIterable(events), + subscribeGlobal: () => Stream.empty, }, session: { hook: (name, callback) => { diff --git a/packages/plugin/src/effect/event.ts b/packages/plugin/src/effect/event.ts index 283d4109f05e..d1875adc12fb 100644 --- a/packages/plugin/src/effect/event.ts +++ b/packages/plugin/src/effect/event.ts @@ -1,3 +1,13 @@ import type { EventApi } from "@opencode-ai/client/effect/api" +import type { Event } from "@opencode-ai/schema/event" +import type { Stream } from "effect" -export interface EventDomain extends Pick, "subscribe"> {} +export interface EventDomain extends Pick, "subscribe"> { + /** + * Subscribe to the global (location-unfiltered) event stream. Unlike + * `subscribe`, this delivers server events for every location, which is + * needed by cross-location observers (e.g. a bot that streams sessions from + * multiple directories). + */ + readonly subscribeGlobal: () => Stream.Stream +} diff --git a/packages/plugin/src/effect/session.ts b/packages/plugin/src/effect/session.ts index a56c890b6954..6ed1185cdbbb 100644 --- a/packages/plugin/src/effect/session.ts +++ b/packages/plugin/src/effect/session.ts @@ -1,4 +1,4 @@ -import type { SessionApi } from "@opencode-ai/client/effect/api" +import type { SessionApi, FormApi } from "@opencode-ai/client/effect/api" import type { GenerationOptionsFields, Message, SystemPart } from "@opencode-ai/ai" import type { Agent } from "@opencode-ai/schema/agent" import type { Model } from "@opencode-ai/schema/model" @@ -7,7 +7,7 @@ import type { Session } from "@opencode-ai/schema/session" import type { SessionInbox } from "@opencode-ai/schema/session-inbox" import type { SessionError } from "@opencode-ai/schema/session-error" import type { SessionMessage } from "@opencode-ai/schema/session-message" -import type { JsonSchema, Types } from "effect" +import { Effect, type JsonSchema, type Types } from "effect" import type { ModelHooks } from "./registration.js" export interface SessionPrompt { @@ -73,6 +73,19 @@ export interface SessionHooks { readonly retry: SessionRetry } +export interface SessionList { + /** Filter to sessions created in this directory. */ + readonly directory?: string + readonly search?: string + readonly order?: "asc" | "desc" + readonly limit?: number +} + +/** In-process session listing — data layer shape, without HTTP cursor encoding. */ +export type SessionListResult = { + readonly data: Session.Info[] +} + export type SessionDomain = Pick< SessionApi, | "create" @@ -89,5 +102,7 @@ export type SessionDomain = Pick< | "wait" | "context" > & { + readonly list: (input?: SessionList) => Effect.Effect readonly hook: ModelHooks + readonly form: Pick, "list" | "get" | "state" | "reply" | "cancel"> } diff --git a/packages/plugin/src/promise/adapter.ts b/packages/plugin/src/promise/adapter.ts index ee0f69c01441..708ad09e2248 100644 --- a/packages/plugin/src/promise/adapter.ts +++ b/packages/plugin/src/promise/adapter.ts @@ -222,6 +222,7 @@ export function fromPromise(plugin: Plugin) { const AgentEndpoints = ClientApi.groups["server.agent"].endpoints const CommandEndpoints = ClientApi.groups["server.command"].endpoints const ExperimentalEndpoints = ClientApi.groups["server.experimental"].endpoints + const FormEndpoints = ClientApi.groups["server.form"].endpoints const GenerateEndpoints = ClientApi.groups["server.generate"].endpoints const IntegrationEndpoints = ClientApi.groups["server.integration"].endpoints const McpEndpoints = ClientApi.groups["server.mcp"].endpoints @@ -334,6 +335,13 @@ export function fromPromise(plugin: Plugin) { ), options, ), + subscribeGlobal: () => + Stream.toAsyncIterable( + host.event.subscribeGlobal().pipe( + Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)), + Stream.map((event) => event as unknown as PromiseEvent), + ), + ), }, experimental: { terminal: { @@ -543,6 +551,8 @@ export function fromPromise(plugin: Plugin) { register( host.session.hook(name, (event) => Effect.promise(() => Promise.resolve(callback(event))), options), ), + list: (input) => + run(host.session.list(input ?? {})).then((result) => ({ data: result.data })), create: adaptApiMethod(SessionEndpoints["session.create"], host.session.create), get: adaptApiMethod(SessionEndpoints["session.get"], host.session.get), switchAgent: adaptApiMethod(SessionEndpoints["session.switchAgent"], host.session.switchAgent), @@ -556,6 +566,13 @@ export function fromPromise(plugin: Plugin) { move: adaptApiMethod(SessionEndpoints["session.move"], host.session.move), wait: adaptApiMethod(SessionEndpoints["session.wait"], host.session.wait), context: adaptApiMethod(SessionEndpoints["session.context"], host.session.context), + form: { + list: adaptApiMethod(FormEndpoints["session.form.list"], host.session.form.list), + get: adaptApiMethod(FormEndpoints["session.form.get"], host.session.form.get), + state: adaptApiMethod(FormEndpoints["session.form.state"], host.session.form.state), + reply: adaptApiMethod(FormEndpoints["session.form.reply"], host.session.form.reply), + cancel: adaptApiMethod(FormEndpoints["session.form.cancel"], host.session.form.cancel), + }, }, shell: { hook: (name, callback) => diff --git a/packages/plugin/src/promise/event.ts b/packages/plugin/src/promise/event.ts index 344f5d6f30a6..a312275f9fa6 100644 --- a/packages/plugin/src/promise/event.ts +++ b/packages/plugin/src/promise/event.ts @@ -1,3 +1,12 @@ import type { EventApi } from "@opencode-ai/client/promise/api" +import type { OpenCodeEvent } from "@opencode-ai/client/promise" -export interface EventDomain extends Pick {} +export interface EventDomain extends Pick { + /** + * Subscribe to the global (location-unfiltered) event stream. Unlike + * `subscribe`, this delivers server events for every location, which is + * needed by cross-location observers (e.g. a bot that streams sessions from + * multiple directories). + */ + readonly subscribeGlobal: () => AsyncIterable +} diff --git a/packages/plugin/src/promise/session.ts b/packages/plugin/src/promise/session.ts index 11b2ee3703d6..fcc6c4b4ce88 100644 --- a/packages/plugin/src/promise/session.ts +++ b/packages/plugin/src/promise/session.ts @@ -1,6 +1,7 @@ import type { SessionApi } from "@opencode-ai/client/promise/api" import type { GenerationOptionsFields, Message, SystemPart } from "@opencode-ai/ai" import type { Agent } from "@opencode-ai/schema/agent" +import type { Form } from "@opencode-ai/schema/form" import type { Model } from "@opencode-ai/schema/model" import type { PromptInput } from "@opencode-ai/schema/prompt-input" import type { Session } from "@opencode-ai/schema/session" @@ -73,6 +74,19 @@ export interface SessionHooks { readonly retry: SessionRetry } +export interface SessionList { + /** Filter to sessions created in this directory. */ + readonly directory?: string + readonly search?: string + readonly order?: "asc" | "desc" + readonly limit?: number +} + +/** In-process session listing — data layer shape, without HTTP cursor encoding. */ +export type SessionListResult = { + readonly data: Session.Info[] +} + export type SessionDomain = Pick< SessionApi, | "create" @@ -89,5 +103,19 @@ export type SessionDomain = Pick< | "wait" | "context" > & { + readonly list: (input?: SessionList) => Promise readonly hook: ModelHooks + readonly form: FormDomain +} + +export interface FormDomain { + readonly list: (input: { readonly sessionID: string }) => Promise> + readonly get: (input: { readonly sessionID: string; readonly formID: string }) => Promise + readonly state: (input: { readonly sessionID: string; readonly formID: string }) => Promise + readonly reply: (input: { + readonly sessionID: string + readonly formID: string + readonly answer: Form.Answer + }) => Promise + readonly cancel: (input: { readonly sessionID: string; readonly formID: string }) => Promise } diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index f515b1a20c2f..2ab38a0c180c 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -2,6 +2,10 @@ export * as ServerProcess from "./process" import { NodeHttpServer } from "@effect/platform-node" import { Bus } from "@opencode-ai/core/bus" +import { Location } from "@opencode-ai/core/location" +import { LocationServiceMap } from "@opencode-ai/core/location-services" +import { Plugin } from "@opencode-ai/core/plugin" +import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionRestart } from "@opencode-ai/core/session/execution/restart" import { hasPtyConnectTicketURL } from "@opencode-ai/protocol/groups/pty" import { hasPersistentPtyConnectTicketURL } from "@opencode-ai/protocol/groups/persistent-pty" @@ -103,6 +107,43 @@ export const start = Effect.fn("ServerProcess.start")(function* ( ).pipe(Layer.provideMerge(NodeHttpServer.layerHttpServices)), applicationScope, ) + + // Eagerly build the server's default location so its plugins activate at + // boot rather than on the first location-scoped request. This is what lets + // always-on in-process plugins (e.g. the Telegram bot) come up immediately + // and keeps the instance alive for the server's whole lifetime. + // + // We deliberately boot against `process.cwd()` (the directory the serve + // command was launched from) rather than `options.config.directory`: that + // field is the *global* config store (OPENCODE_CONFIG_DIR, ~/.config/opencode), + // not a project, so binding the always-on plugins there would be wrong. The + // eager boot is just a default that comes up at startup; any project can + // still be targeted per-request via the `x-opencode-directory` header, which + // bootstraps that location's instance on demand. cwd is the only directory + // we know at boot before any request identifies a project. + yield* Effect.gen(function* () { + const locations = yield* LocationServiceMap.Service + const services = locations.get( + Location.Ref.make({ directory: AbsolutePath.make(process.cwd()) }), + ) + const instance = yield* Layer.build(services).pipe( + Effect.provideService(Scope.Scope, applicationScope), + ) + // Wait (tolerantly) for the initial plugin generation to settle so the + // bot's long-poll loop is running before the server reports ready. + yield* Context.get(instance, Plugin.Service).awaitActivation.pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => Effect.void, + }), + ) + }).pipe( + Effect.provideService( + LocationServiceMap.Service, + Context.get(context, LocationServiceMap.Service), + ), + ) + if (lifecycle) { yield* installRestartContinuity(Context.get(context, SessionRestart.Service)).pipe( Effect.provideService(Scope.Scope, applicationScope), From 4e25d0c51ae884f02719b65632d7ea5880306552 Mon Sep 17 00:00:00 2001 From: Michael Blakeley Date: Wed, 2 Sep 2026 21:49:33 -0700 Subject: [PATCH 2/2] feat(plugin): expose session forms, session list, and global event stream-- changes per internal review --- packages/core/src/bus.ts | 25 +++++++++ packages/core/src/plugin/host.ts | 39 +++++++++++--- packages/core/src/session.ts | 12 +++-- .../core/test/bus-subscribe-global.test.ts | 51 +++++++++++++++++++ packages/plugin/src/effect/event.ts | 7 +-- packages/plugin/src/effect/session.ts | 6 ++- packages/plugin/src/promise/adapter.ts | 5 +- packages/plugin/src/promise/event.ts | 9 ++-- packages/plugin/src/promise/session.ts | 6 ++- packages/server/src/process.ts | 6 +++ 10 files changed, 142 insertions(+), 24 deletions(-) create mode 100644 packages/core/test/bus-subscribe-global.test.ts diff --git a/packages/core/src/bus.ts b/packages/core/src/bus.ts index 4d3f79653265..3cdfa94f821c 100644 --- a/packages/core/src/bus.ts +++ b/packages/core/src/bus.ts @@ -146,6 +146,12 @@ export interface Interface { events: I, ) => Effect.Effect> readonly subscribe: Subscribe + /** + * Unfiltered live channel: every event published from now on, across all + * locations. Prefer `subscribe` for location-scoped consumers; use this for + * cross-location observers that must not miss other directories. + */ + readonly subscribeGlobal: Subscribe /** * Durable, ordered per-aggregate log read. Forked aggregates may reserve an * inherited prefix before their first child-authored event. `follow: false` @@ -761,6 +767,24 @@ export function configured(options?: Options) { const streamLive = (): Stream.Stream => local(Stream.fromPubSub(pubsub.live)) + const streamLiveGlobal = (): Stream.Stream => Stream.fromPubSub(pubsub.live) + + function subscribeGlobal(): Stream.Stream + function subscribeGlobal(definition: D): Stream.Stream> + function subscribeGlobal( + definitions: D, + ): Stream.Stream> + function subscribeGlobal( + input?: Event.Definition | readonly Event.Definition[], + ): Stream.Stream { + if (input === undefined) return streamLiveGlobal() + if (isDefinition(input)) { + return Stream.unwrap(getOrCreate(input).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))) + } + const types = new Set(input.map((definition) => definition.type)) + return streamLiveGlobal().pipe(Stream.filter((event) => types.has(event.type))) + } + const readAfter = ( aggregateID: string, after: number, @@ -893,6 +917,7 @@ export function configured(options?: Options) { publish, publishAll, subscribe, + subscribeGlobal, log, listen, project, diff --git a/packages/core/src/plugin/host.ts b/packages/core/src/plugin/host.ts index 228b31066f2e..22530ecd2da9 100644 --- a/packages/core/src/plugin/host.ts +++ b/packages/core/src/plugin/host.ts @@ -6,6 +6,7 @@ import { EventManifest } from "@opencode-ai/schema/event-manifest" import type { Event } from "@opencode-ai/schema/event" import { ServerConfig } from "@opencode-ai/schema/mcp" import { App } from "../app.js" +import path from "path" import { Effect, Queue, Schema, Stream } from "effect" import { Agent } from "../agent.js" import { AISDK } from "../aisdk.js" @@ -249,16 +250,36 @@ export const make = Effect.fn("PluginHost.make")(function* ( * scoped to the plugin's ambient Location, so a cross-location observer * (e.g. a Telegram bot that /cd's between projects) would silently miss * events for sessions in other directories. This variant feeds from the - * bus's unfiltered listener channel and delivers every server event. + * bus's unfiltered channel and delivers every server event (plus rpc + * events, matching `subscribe`). The bounded dropping queue keeps a slow + * consumer from stalling publication; overflows are logged and dropped. */ subscribeGlobal: () => Stream.scoped( Stream.unwrap( Effect.gen(function* () { const queue = yield* Queue.dropping(GLOBAL_EVENT_QUEUE_CAPACITY) - const unsubscribe = yield* bus.listen((event) => Queue.offer(queue, event).pipe(Effect.asVoid)) - yield* Effect.addFinalizer((_exit) => unsubscribe) - return Stream.fromQueue(queue).pipe(Stream.filter(EventManifest.isServer)) + yield* bus.subscribeGlobal().pipe( + Stream.runForEach((event: Event.Payload) => + Queue.offer(queue, event).pipe( + Effect.flatMap((accepted) => + accepted + ? Effect.void + : Effect.logWarning("Plugin global event buffer full, dropping event", { + type: event.type, + }), + ), + ), + ), + Effect.forkScoped, + ) + yield* Effect.addFinalizer(() => Queue.shutdown(queue)) + return Stream.fromQueue(queue).pipe( + Stream.filter( + (event): event is EventManifest.ServerEvent | RpcEvent => + EventManifest.isServer(event) || isRpcEvent(event), + ), + ) }), ), ), @@ -491,13 +512,16 @@ export const make = Effect.fn("PluginHost.make")(function* ( }, session: { hook: (name, callback, options) => hooks.register("session", name, callback, options), - list: (input) => - sessions.list({ + list: (input) => { + if (input?.directory !== undefined && !path.isAbsolute(input.directory)) + return Effect.die(new Error(`session.list directory must be absolute: ${input.directory}`)) + return sessions.list({ ...(input?.directory === undefined ? {} : { directory: AbsolutePath.make(input.directory) }), search: input?.search, limit: input?.limit, order: input?.order, - }), + }) + }, create: (input) => sessions.create({ id: input?.id, @@ -522,6 +546,7 @@ export const make = Effect.fn("PluginHost.make")(function* ( .pipe(Effect.map((interrupted) => ({ interrupted }))), wait: (input) => sessions.wait(input.sessionID), context: (input) => sessions.context(input.sessionID), + // Form list surfaces pending forms only, matching Form.Service semantics. form: { list: (input: { sessionID: string }) => sessions.form.list({ sessionID: input.sessionID }), get: (input: { sessionID: string; formID: Form.ID }) => sessions.form.get(input), diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index c340310e8e13..a331ad81d71c 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -269,19 +269,23 @@ const layer = Layer.effect( // Forms are created by tools running under the session's instance, so the // ambient Form.Service here is NOT where they live. Resolve the session's - // location and route through that location's Form.Service instead. + // location and route through that location's Form.Service instead. The + // lookup-then-route is best-effort: a session that moves between the two + // steps resolves against the pre-move location and may report NotFound. const formFor = ( sessionID: string, run: (form: Form.Interface) => Effect.Effect, - ) => - store.get(SessionSchema.ID.descending(sessionID)).pipe( + ) => { + const id = SessionSchema.ID.descending(sessionID) + return store.get(id).pipe( Effect.flatMap( (session): Effect.Effect => session === undefined - ? Effect.fail(new NotFoundError({ sessionID: SessionSchema.ID.descending(sessionID) })) + ? Effect.fail(new NotFoundError({ sessionID: id })) : instances.provide(session)(Form.Service.use(run)), ), ) + } const result = Service.of({ create: Effect.fn("Session.create")(function* (input) { diff --git a/packages/core/test/bus-subscribe-global.test.ts b/packages/core/test/bus-subscribe-global.test.ts new file mode 100644 index 000000000000..00f22688aac8 --- /dev/null +++ b/packages/core/test/bus-subscribe-global.test.ts @@ -0,0 +1,51 @@ +import { describe, expect } from "bun:test" +import { Effect, Fiber, Stream } from "effect" +import { Bus } from "@opencode-ai/core/bus" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { Location } from "@opencode-ai/core/location" +import { AbsolutePath } from "@opencode-ai/schema/schema" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { location } from "./fixture/location" +import { testEffect } from "./lib/effect" + +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node]))) + +const a = Location.Ref.make({ directory: AbsolutePath.make("/a") }) +const b = Location.Ref.make({ directory: AbsolutePath.make("/b") }) +const Done = Bus.ephemeral({ type: "test.global.done", schema: {} }) +const Ping = Bus.ephemeral({ type: "test.global.ping", schema: {} }) + +describe("Bus subscribeGlobal", () => { + it.effect("sees other-location events that scoped subscribe filters out", () => + Effect.gen(function* () { + const bus = yield* Bus.Service + const scoped = yield* bus + .subscribe() + .pipe( + Stream.takeUntil((event) => event.type === Done.type), + Stream.runCollect, + Effect.provideService(Location.Service, location(b)), + Effect.forkScoped({ startImmediately: true }), + ) + const global = yield* bus + .subscribeGlobal() + .pipe( + Stream.takeUntil((event) => event.type === Done.type), + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }), + ) + // Both subscriptions fork with startImmediately, matching the + // bus-session-routing pattern, before publishing. + const ping = yield* bus.publish(Ping, {}, { location: a }) + const done = yield* bus.publish(Done, {}, { global: true }) + const scopedEvents = Array.from(yield* Fiber.join(scoped)) + const globalEvents = Array.from(yield* Fiber.join(global)) + // Scoped to /b: location-A ping is filtered, global done passes through. + expect(scopedEvents.map((event) => event.type)).toEqual([Done.type]) + expect(globalEvents.map((event) => event.type)).toEqual([Ping.type, Done.type]) + expect(globalEvents[0]).toEqual(ping) + expect(globalEvents[1]).toEqual(done) + }), + ) +}) diff --git a/packages/plugin/src/effect/event.ts b/packages/plugin/src/effect/event.ts index d1875adc12fb..f601d044e1a8 100644 --- a/packages/plugin/src/effect/event.ts +++ b/packages/plugin/src/effect/event.ts @@ -5,9 +5,10 @@ import type { Stream } from "effect" export interface EventDomain extends Pick, "subscribe"> { /** * Subscribe to the global (location-unfiltered) event stream. Unlike - * `subscribe`, this delivers server events for every location, which is - * needed by cross-location observers (e.g. a bot that streams sessions from - * multiple directories). + * `subscribe`, this delivers server and rpc events for every location, which + * is needed by cross-location observers (e.g. a bot that streams sessions + * from multiple directories). The stream is loss-tolerant: a slow consumer + * drops buffered events rather than stalling publication. */ readonly subscribeGlobal: () => Stream.Stream } diff --git a/packages/plugin/src/effect/session.ts b/packages/plugin/src/effect/session.ts index 6ed1185cdbbb..39c02fb0ac7f 100644 --- a/packages/plugin/src/effect/session.ts +++ b/packages/plugin/src/effect/session.ts @@ -73,15 +73,17 @@ export interface SessionHooks { readonly retry: SessionRetry } +/** Intentional subset of SessionApi["list"]: in-process only, no cursor/pagination. */ export interface SessionList { - /** Filter to sessions created in this directory. */ + /** Filter to sessions created in this directory. Must be absolute. */ readonly directory?: string readonly search?: string readonly order?: "asc" | "desc" + /** Maximum sessions to return. Truncates without a cursor; there is no pagination. */ readonly limit?: number } -/** In-process session listing — data layer shape, without HTTP cursor encoding. */ +/** In-process session listing — data layer shape, without HTTP cursor encoding. Truncated at `limit`. */ export type SessionListResult = { readonly data: Session.Info[] } diff --git a/packages/plugin/src/promise/adapter.ts b/packages/plugin/src/promise/adapter.ts index 708ad09e2248..77822f1dac0b 100644 --- a/packages/plugin/src/promise/adapter.ts +++ b/packages/plugin/src/promise/adapter.ts @@ -335,12 +335,13 @@ export function fromPromise(plugin: Plugin) { ), options, ), - subscribeGlobal: () => - Stream.toAsyncIterable( + subscribeGlobal: (options) => + streams( host.event.subscribeGlobal().pipe( Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)), Stream.map((event) => event as unknown as PromiseEvent), ), + options, ), }, experimental: { diff --git a/packages/plugin/src/promise/event.ts b/packages/plugin/src/promise/event.ts index a312275f9fa6..4d45c5e7f79a 100644 --- a/packages/plugin/src/promise/event.ts +++ b/packages/plugin/src/promise/event.ts @@ -4,9 +4,10 @@ import type { OpenCodeEvent } from "@opencode-ai/client/promise" export interface EventDomain extends Pick { /** * Subscribe to the global (location-unfiltered) event stream. Unlike - * `subscribe`, this delivers server events for every location, which is - * needed by cross-location observers (e.g. a bot that streams sessions from - * multiple directories). + * `subscribe`, this delivers server and rpc events for every location, which + * is needed by cross-location observers (e.g. a bot that streams sessions + * from multiple directories). The stream is loss-tolerant: a slow consumer + * drops buffered events rather than stalling publication. */ - readonly subscribeGlobal: () => AsyncIterable + readonly subscribeGlobal: (options?: { readonly signal?: AbortSignal }) => AsyncIterable } diff --git a/packages/plugin/src/promise/session.ts b/packages/plugin/src/promise/session.ts index fcc6c4b4ce88..bb0e6e3af190 100644 --- a/packages/plugin/src/promise/session.ts +++ b/packages/plugin/src/promise/session.ts @@ -74,15 +74,17 @@ export interface SessionHooks { readonly retry: SessionRetry } +/** Intentional subset of SessionApi["list"]: in-process only, no cursor/pagination. */ export interface SessionList { - /** Filter to sessions created in this directory. */ + /** Filter to sessions created in this directory. Must be absolute. */ readonly directory?: string readonly search?: string readonly order?: "asc" | "desc" + /** Maximum sessions to return. Truncates without a cursor; there is no pagination. */ readonly limit?: number } -/** In-process session listing — data layer shape, without HTTP cursor encoding. */ +/** In-process session listing — data layer shape, without HTTP cursor encoding. Truncated at `limit`. */ export type SessionListResult = { readonly data: Session.Info[] } diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index 2ab38a0c180c..1748d965b30a 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -142,6 +142,12 @@ export const start = Effect.fn("ServerProcess.start")(function* ( LocationServiceMap.Service, Context.get(context, LocationServiceMap.Service), ), + // The default location is best-effort: a plugin init defect here must not + // take down the whole server when lazy per-request boot would only fail + // one request. + Effect.catchCause((cause) => + Effect.logWarning("Failed to eagerly boot default location, continuing without it", { cause }), + ), ) if (lifecycle) {