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: () => Stream.Stream<Event.Payload>
   /**
    * Durable, ordered per-aggregate log read. Forked aggregates may reserve an
    * inherited prefix before their first child-authored event. `follow: false`
@@ -761,6 +767,10 @@ 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)
+
+        const subscribeGlobal = (): Stream.Stream<Event.Payload> => streamLiveGlobal()
+
         const readAfter = (
           aggregateID: string,
           after: number,
@@ -893,6 +903,7 @@ export function configured(options?: Options) {
           publish,
           publishAll,
           subscribe,
+          subscribeGlobal,
           log,
           listen,
           project,

packages/core/src/plugin/host.ts

@@ -4,9 +4,11 @@ import { Plugin } from "@opencode/plugin/effect"
 import type { IntegrationMethodRegistration } from "@opencode/plugin/effect/integration"
 import { EventManifest } from "@opencode/schema/event-manifest"
 import type { Event } from "@opencode/schema/event"
+import type { Session as SchemaSession } from "@opencode/schema/session"
 import { ServerConfig } from "@opencode/schema/mcp"
 import { App } from "../app.js"
-import { Effect, Schema, Stream } from "effect"
+import path from "path"
+import { Effect, Queue, Ref, Schema, Stream } from "effect"
 import { Agent } from "../agent.js"
 import { AISDK } from "../aisdk.js"
 import { Command } from "../command.js"
@@ -16,6 +18,7 @@ import { Integration } from "../integration.js"
 import { KV } from "../kv.js"
 import { Location } from "../location.js"
 import { LocationServiceMap } from "../location-service-map.js"
+import { permissionForSession } from "./permission.js"
 import { Model } from "../model.js"
 import { Mcp } from "../mcp/index.js"
 import { Session } from "../session.js"
@@ -33,6 +36,7 @@ import { Worktree } from "../worktree.js"
 import { WorktreeStrategies } from "../worktree/strategies.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/util/effect/layer-node"
@@ -44,6 +48,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",
@@ -112,6 +124,22 @@ export const make = Effect.fn("PluginHost.make")(function* (
   )
 
   // Keep the instance graph's inferred types independent of Session handles.
+  // Permission state is location-scoped while sessions are global, so every
+  // session-keyed permission op resolves the session's own Permission
+  // instance first (see `./permission.ts`); without this, requests asked in
+  // another directory are invisible here and every reply fails "not found".
+  const permissionScoped = (ref: Location.Ref) =>
+    Effect.gen(function* () {
+      return yield* Permission.Service
+    }).pipe(
+      Effect.provide(locations.get(ref)),
+      Effect.orDie,
+    )
+  const permissionIn = <A, E>(
+    sessionID: SchemaSession.ID,
+    use: (permission: Permission.Interface) => Effect.Effect<A, E, never>,
+  ) => permissionForSession({ sessions, scoped: permissionScoped, ambient: permission, sessionID, use })
+
   const context: Plugin.Context = {
     app,
     location: locationInfo(),
@@ -262,6 +290,52 @@ 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 dropped with a
+       * cumulative count in the warning. Consumers that need gap-free delivery
+       * must use the durable `log`, not this loss-tolerant live bridge.
+       */
+      subscribeGlobal: () =>
+        Stream.scoped(
+          Stream.unwrap(
+            Effect.gen(function* () {
+              const queue = yield* Queue.dropping<Event.Payload>(GLOBAL_EVENT_QUEUE_CAPACITY)
+              const dropped = yield* Ref.make(0)
+              yield* bus.subscribeGlobal().pipe(
+                Stream.runForEach((event: Event.Payload) =>
+                  Queue.offer(queue, event).pipe(
+                    Effect.flatMap((accepted) =>
+                      accepted
+                        ? Effect.void
+                        : Ref.updateAndGet(dropped, (total) => total + 1).pipe(
+                            Effect.flatMap((total) =>
+                              Effect.logWarning("Plugin global event buffer full, dropping event", {
+                                type: event.type,
+                                droppedTotal: total,
+                              }),
+                            ),
+                          ),
+                    ),
+                  ),
+                ),
+                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: {
@@ -394,7 +468,7 @@ export const make = Effect.fn("PluginHost.make")(function* (
     },
     permission: {
       hook: (name, callback) => hooks.register("permission", name, callback),
-      list: (input) => permission.forSession(input.sessionID),
+      list: (input) => permissionIn(input.sessionID, (permission) => permission.forSession(input.sessionID)),
       get: (input) =>
         permission
           .get(input.requestID)
@@ -406,15 +480,17 @@ export const make = Effect.fn("PluginHost.make")(function* (
             ),
           ),
       reply: (input) =>
-        permission
-          .get(input.requestID)
-          .pipe(
-            Effect.flatMap((request) =>
-              request?.sessionID === input.sessionID
-                ? permission.reply({ requestID: input.requestID, reply: input.decision, message: input.message })
-                : Effect.fail(new Error(`Permission request not found: ${input.requestID}`)),
+        permissionIn(input.sessionID, (permission) =>
+          permission
+            .get(input.requestID)
+            .pipe(
+              Effect.flatMap((request) =>
+                request?.sessionID === input.sessionID
+                  ? permission.reply({ requestID: input.requestID, reply: input.decision, message: input.message })
+                  : Effect.fail(new Error(`Permission request not found: ${input.requestID}`)),
+              ),
             ),
-          ),
+        ),
     },
     plugin: {
       list: () => response(plugin.list()),
@@ -516,6 +592,14 @@ 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.fail(new Error(`session.list directory must be absolute: ${input.directory}`))
+        return sessions.list({
+          ...(input?.directory === undefined ? {} : { directory: AbsolutePath.make(input.directory) }),
+          limit: input?.limit,
+        })
+      },
       create: (input) =>
         sessions.create({
           id: input?.id,
@@ -547,6 +631,19 @@ 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),
+      compact: sessions.compact,
+      skill: (input) => sessions.skill({ ...input, skill: input.id }),
+      revert: {
+        stage: sessions.revert.stage,
+        clear: (input) => sessions.revert.clear(input.sessionID),
+        commit: (input) => sessions.revert.commit(input.sessionID),
+      },
+      // Form list surfaces pending forms only, matching Form.Service semantics.
+      form: {
+        list: (input: { sessionID: string }) => sessions.form.list({ sessionID: input.sessionID }),
+        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/plugin/permission.ts

@@ -0,0 +1,38 @@
+import { Effect } from "effect"
+import type { Session } from "@opencode/schema/session"
+import type { Location } from "@opencode/schema/location"
+import { Permission } from "../permission.js"
+
+/**
+ * Run a permission op against the Permission instance owning the session.
+ *
+ * Permission state is location-scoped while sessions (and their events) are
+ * global: a request asked in another directory pends in that location's
+ * instance, invisible to the caller's own `permission` service (whose lookup
+ * then fails "not found"). Resolve the session's location first and use that
+ * instance instead, like the HTTP session-location middleware does. Sessions
+ * that no longer resolve fall back to the ambient instance (legacy behavior).
+ *
+ * Known limit: a request that pended before `session.move` stays in the old
+ * location's instance while the session now resolves to the new one, so it
+ * still reports not found (same move race as session forms). Migrating
+ * in-flight permission/form state on move is future work; callers that need
+ * it should reply before moving the session.
+ */
+export function permissionForSession<A, E>(input: {
+  sessions: {
+    get: (sessionID: Session.ID) => Effect.Effect<{ readonly location: Location.Ref }, unknown>
+  }
+  scoped: (ref: Location.Ref) => Effect.Effect<Permission.Interface, never, never>
+  ambient: Permission.Interface
+  sessionID: Session.ID
+  use: (permission: Permission.Interface) => Effect.Effect<A, E, never>
+}): Effect.Effect<A, E, never> {
+  return Effect.flatMap(
+    input.sessions.get(input.sessionID).pipe(Effect.orElseSucceed(() => undefined)),
+    (session) =>
+      session === undefined
+        ? input.use(input.ambient)
+        : Effect.flatMap(input.scoped(session.location), input.use),
+  )
+}

packages/core/src/session.ts

@@ -11,6 +11,7 @@ import { Location } from "./location.js"
 import { SessionMessage } from "./session/message.js"
 import { PromptInput } from "@opencode/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"
@@ -218,6 +219,19 @@ 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: {
+    /** Pending forms only, matching Form.Service semantics. */
+    readonly list: (input: { readonly sessionID: string }) => Effect.Effect<readonly Form.Info[], 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") {}
@@ -242,6 +256,26 @@ const layer = Layer.effect(
     const sessions = yield* Session.make()
     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.
+    // Malformed session IDs fail with NotFoundError; the store lookup decides
+    // existence for well-formed IDs.
+    const formFor = <A, E>(
+      sessionID: string,
+      run: (form: Form.Interface) => Effect.Effect<A, E>,
+    ) => {
+      const id = sessionID as SessionSchema.ID
+      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()
@@ -448,6 +482,11 @@ 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 })),
+        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/core/bus"
+import { Database } from "@opencode/core/database/database"
+import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
+import { Location } from "@opencode/core/location"
+import { AbsolutePath } from "@opencode/schema/schema"
+import { LayerNode } from "@opencode/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

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