feat(plugin): expose session forms, session list, and global event stream

packages/core/src/bus.ts

@@ -146,6 +146,12 @@ export interface Interface {
     events: I,
   ) => Effect.Effect<PublishResult<I>>
   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<Event.Payload> => local(Stream.fromPubSub(pubsub.live))
 
+        const streamLiveGlobal = (): Stream.Stream<Event.Payload> => Stream.fromPubSub(pubsub.live)
+
+        function subscribeGlobal(): Stream.Stream<Event.Payload>
+        function subscribeGlobal<D extends Event.Definition>(definition: D): Stream.Stream<Event.Payload<D>>
+        function subscribeGlobal<const D extends readonly [Event.Definition, ...Event.Definition[]]>(
+          definitions: D,
+        ): Stream.Stream<SubscribePayload<D>>
+        function subscribeGlobal(
+          input?: Event.Definition | readonly Event.Definition[],
+        ): Stream.Stream<Event.Payload> {
+          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,

packages/core/src/plugin/host.ts

@@ -6,7 +6,8 @@ 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 path from "path"
+import { Effect, Queue, Schema, Stream } from "effect"
 import { Agent } from "../agent.js"
 import { AISDK } from "../aisdk.js"
 import { Catalog } from "../catalog.js"
@@ -32,6 +33,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 +45,14 @@ type RpcEvent = Event.Payload & {
   readonly data: Readonly<Record<string, unknown>>
 }
 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<Interface, "list">,
   pluginID: string = "test",
@@ -235,6 +245,44 @@ 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 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<Event.Payload>(GLOBAL_EVENT_QUEUE_CAPACITY)
+              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),
+                ),
+              )
+            }),
+          ),
+        ),
     },
     experimental: {
       terminal: {
@@ -464,6 +512,16 @@ export const make = Effect.fn("PluginHost.make")(function* (
     },
     session: {
       hook: (name, callback, options) => hooks.register("session", name, callback, options),
+      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,
@@ -488,6 +546,14 @@ 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),
+        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

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<void, NotFoundError | BusyError | Snapshot.Error>
     readonly commit: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | BusyError>
   }
+  readonly form: {
+    readonly list: (input: { readonly sessionID: string }) => Effect.Effect<readonly Form.Info[], NotFoundError>
+    readonly get: (input: {
+      readonly sessionID: string
+      readonly formID: Form.ID
+    }) => Effect.Effect<Form.Info, NotFoundError | Form.NotFoundError>
+    readonly state: (input: {
+      readonly sessionID: string
+      readonly formID: Form.ID
+    }) => Effect.Effect<Form.State, NotFoundError | Form.NotFoundError>
+    readonly reply: (input: {
+      readonly sessionID: string
+      readonly formID: Form.ID
+      readonly answer: Form.Answer
+    }) => Effect.Effect<void, NotFoundError | Form.NotFoundError | Form.AlreadySettledError | Form.InvalidAnswerError>
+    readonly cancel: (input: {
+      readonly sessionID: string
+      readonly formID: Form.ID
+    }) => Effect.Effect<void, NotFoundError | Form.NotFoundError | Form.AlreadySettledError>
+  }
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Session") {}
@@ -246,6 +267,26 @@ 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. 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 = <A, E>(
+      sessionID: string,
+      run: (form: Form.Interface) => Effect.Effect<A, E>,
+    ) => {
+      const id = SessionSchema.ID.descending(sessionID)
+      return store.get(id).pipe(
+        Effect.flatMap(
+          (session): Effect.Effect<A, NotFoundError | E> =>
+            session === undefined
+              ? Effect.fail(new NotFoundError({ sessionID: id }))
+              : 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 +519,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

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)
+    }),
+  )
+})

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 },
         }),
       )
 

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 },
         }),
       )
 

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

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(

packages/core/test/filesystem/watcher.test.ts

@@ -280,7 +280,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),

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"),
+      },
     },
   }
 }

packages/core/test/plugin/plan.test.ts

@@ -84,6 +84,7 @@ const run = Effect.fnUntraced(function* (events: ReadonlyArray<SessionEvent.Agen
       },
       event: {
         subscribe: () => Stream.fromIterable(events),
+        subscribeGlobal: () => Stream.empty,
       },
       session: {
         hook: (name, callback) => {

packages/plugin/src/effect/event.ts

@@ -1,3 +1,14 @@
 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<EventApi<unknown>, "subscribe"> {}
+export interface EventDomain extends Pick<EventApi<unknown>, "subscribe"> {
+  /**
+   * Subscribe to the global (location-unfiltered) event stream. Unlike
+   * `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<Event.Payload>
+}