From 922e283291f9e77710c60fb0a404cc11af85497d Mon Sep 17 00:00:00 2001 From: Michael Blakeley Date: Tue, 1 Sep 2026 15:52:18 -0700 Subject: [PATCH] feat(plugin): expose session forms, session list, permissions, skills, compact, revert and global event stream --- packages/core/src/bus.ts | 11 ++ packages/core/src/plugin/host.ts | 117 +++++++++++++-- packages/core/src/plugin/permission.ts | 38 +++++ packages/core/src/session.ts | 39 +++++ .../core/test/bus-subscribe-global.test.ts | 51 +++++++ 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 +- .../test/permission-location-routing.test.ts | 135 ++++++++++++++++++ packages/core/test/plugin/host.ts | 14 ++ packages/core/test/plugin/plan.test.ts | 1 + .../core/test/session-form-not-found.test.ts | 60 ++++++++ packages/core/test/worktree.test.ts | 2 +- packages/plugin/src/effect/event.ts | 15 +- packages/plugin/src/effect/session.ts | 22 +++ packages/plugin/src/promise/adapter.ts | 26 ++++ packages/plugin/src/promise/event.ts | 14 +- packages/plugin/src/promise/session.ts | 29 ++++ packages/server/src/location.ts | 14 +- packages/server/src/process.ts | 52 +++++++ .../src/docs/content/build/plugins/index.mdx | 6 +- 23 files changed, 637 insertions(+), 22 deletions(-) create mode 100644 packages/core/src/plugin/permission.ts create mode 100644 packages/core/test/bus-subscribe-global.test.ts create mode 100644 packages/core/test/permission-location-routing.test.ts create mode 100644 packages/core/test/session-form-not-found.test.ts diff --git a/packages/core/src/bus.ts b/packages/core/src/bus.ts index 30bf10a31363..8e17b6342a5a 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: () => Stream.Stream /** * 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 => local(Stream.fromPubSub(pubsub.live)) + const streamLiveGlobal = (): Stream.Stream => Stream.fromPubSub(pubsub.live) + + const subscribeGlobal = (): Stream.Stream => streamLiveGlobal() + const readAfter = ( aggregateID: string, after: number, @@ -893,6 +903,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 129dbe775137..26868efb7828 100644 --- a/packages/core/src/plugin/host.ts +++ b/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> } 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", @@ -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 = ( + sessionID: SchemaSession.ID, + use: (permission: Permission.Interface) => Effect.Effect, + ) => 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(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 diff --git a/packages/core/src/plugin/permission.ts b/packages/core/src/plugin/permission.ts new file mode 100644 index 000000000000..7f6faf86c019 --- /dev/null +++ b/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(input: { + sessions: { + get: (sessionID: Session.ID) => Effect.Effect<{ readonly location: Location.Ref }, unknown> + } + scoped: (ref: Location.Ref) => Effect.Effect + ambient: Permission.Interface + sessionID: Session.ID + use: (permission: Permission.Interface) => Effect.Effect +}): Effect.Effect { + 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), + ) +} diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index f2d4286230b1..d62102d6f838 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/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 readonly commit: (sessionID: SessionSchema.ID) => Effect.Effect } + readonly form: { + /** Pending forms only, matching Form.Service semantics. */ + readonly list: (input: { readonly sessionID: string }) => 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") {} @@ -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 = ( + sessionID: string, + run: (form: Form.Interface) => Effect.Effect, + ) => { + const id = sessionID as SessionSchema.ID + return store.get(id).pipe( + Effect.flatMap( + (session): Effect.Effect => + 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 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..d875913268d4 --- /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/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) + }), + ) +}) diff --git a/packages/core/test/config/agent.test.ts b/packages/core/test/config/agent.test.ts index 6456e6a20be1..dc16cde8a48f 100644 --- a/packages/core/test/config/agent.test.ts +++ b/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 }, }), ) diff --git a/packages/core/test/config/command.test.ts b/packages/core/test/config/command.test.ts index e3e7ea265aef..11ec5f3d0ce1 100644 --- a/packages/core/test/config/command.test.ts +++ b/packages/core/test/config/command.test.ts @@ -170,7 +170,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(() => { @@ -342,7 +342,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 e292f942ae9c..71caa88200b3 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 03d5c6a57f5c..0d713a0c5b38 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 feaaeecb8fec..1edcf818ae79 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -383,7 +383,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/permission-location-routing.test.ts b/packages/core/test/permission-location-routing.test.ts new file mode 100644 index 000000000000..3d4cd0f5e450 --- /dev/null +++ b/packages/core/test/permission-location-routing.test.ts @@ -0,0 +1,135 @@ +import fs from "fs/promises" +import path from "path" +import { describe, expect } from "bun:test" +import { Effect, Layer } from "effect" +import { Agent } from "@opencode/core/agent" +import { Database } from "@opencode/core/database/database" +import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" +import { LayerNode } from "@opencode/util/effect/layer-node" +import { Global } from "@opencode/util/global" +import { Bus } from "@opencode/core/bus" +import { Location } from "@opencode/core/location" +import { LocationServiceMap } from "@opencode/core/location-services" +import { Permission } from "@opencode/core/permission" +import { Project } from "@opencode/core/project" +import { ProjectTable } from "@opencode/core/project/sql" +import { AbsolutePath } from "@opencode/core/schema" +import type { Session } from "@opencode/schema/session" +import { SessionStore } from "@opencode/core/session/store" +import { SessionTable } from "@opencode/core/session/sql" +import { permissionForSession } from "@opencode/core/plugin/permission" +import { tmpdir } from "./fixture/tmpdir" +import { tempGlobalLayer } from "./fixture/global" +import { offlineModels } from "./fixture/models" +import { testEffect } from "./lib/effect" + +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionStore.node, LocationServiceMap.node]), [ + Global.node.replace(tempGlobalLayer), + offlineModels, + ]), +) + +const sesB = "ses_perm_loc_b" as Session.ID +const reqID = Permission.ID.create("per_perm_loc_routed") + +/** A real scratch directory with a planted (empty) project config. */ +const scratch = Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), +).pipe( + Effect.flatMap((dir) => + Effect.promise(async () => { + await fs.writeFile(path.join(dir.path, "opencode.json"), "{}") + return Location.Ref.make({ directory: AbsolutePath.make(dir.path) }) + }), + ), +) + +describe("permissionForSession", () => { + it.live("replies through the session's own location instance", () => + Effect.gen(function* () { + const refA = yield* scratch + const refB = yield* scratch + const locations = yield* LocationServiceMap.Service + const inScope = (ref: Location.Ref, effect: Effect.Effect) => + Effect.provide(effect, locations.get(ref)) + + // The session lives in location B: seed its row plus an empty ruleset + // (no matching rule evaluates to "ask") and leave one request pending. + const setup = Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: refB.directory, sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sesB, + project_id: Project.ID.global, + slug: "permission-routing", + directory: refB.directory, + title: "permission routing", + version: "test", + agent: "test", + }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + const agents = yield* Agent.Service + yield* agents.transform((editor) => + editor.update(Agent.ID.make("test"), (agent) => { + agent.permissions = [] + }), + ) + const permission = yield* Permission.Service + const asked = yield* permission.ask({ + id: reqID, + sessionID: sesB, + action: "read", + resources: ["src/index.ts"], + }) + expect(asked.effect).toBe("ask") + return permission + }) + const permB = yield* inScope(refB, setup) + const permA = yield* inScope(refA, Effect.gen(function* () { + return yield* Permission.Service + })) + + // The ambient (plugin-location) instance cannot see the request: this is + // the shape of the Telegram "Permission request not found" failure. + expect(yield* permA.get(reqID)).toBeUndefined() + + const store = yield* SessionStore.Service + const sessions = { + get: (id: Session.ID) => + Effect.flatMap(store.get(id), (session) => + session === undefined + ? Effect.fail(new Error(`Session not found: ${id}`)) + : Effect.succeed({ location: session.location }), + ), + } + const scoped = (ref: Location.Ref) => + Effect.gen(function* () { + return yield* Permission.Service + }).pipe( + Effect.provide(locations.get(ref)), + Effect.orDie, + ) + + yield* permissionForSession({ + sessions, + scoped, + ambient: permA, + sessionID: sesB, + use: (permission) => permission.reply({ requestID: reqID, reply: "reject" }), + }) + expect(yield* permB.get(reqID)).toBeUndefined() + expect(yield* permB.list()).toEqual([]) + }), + ) +}) diff --git a/packages/core/test/plugin/host.ts b/packages/core/test/plugin/host.ts index 6cbb5230a73c..5d5ef0cd43d8 100644 --- a/packages/core/test/plugin/host.ts +++ b/packages/core/test/plugin/host.ts @@ -64,6 +64,7 @@ export function host(overrides: Overrides = {}): Plugin.Context { }, event: overrides.event ?? { subscribe: () => Stream.empty, + subscribeGlobal: () => Stream.empty, }, experimental: overrides.experimental ?? { terminal: { @@ -162,6 +163,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")), @@ -174,7 +176,19 @@ export function host(overrides: Overrides = {}): Plugin.Context { synthetic: overrides.session?.synthetic ?? (() => Effect.die("unused session.synthetic")), interrupt: overrides.session?.interrupt ?? (() => Effect.die("unused session.interrupt")), wait: overrides.session?.wait ?? (() => Effect.die("unused session.wait")), + skill: overrides.session?.skill ?? (() => Effect.die("unused session.skill")), + compact: overrides.session?.compact ?? (() => Effect.die("unused session.compact")), + revert: overrides.session?.revert ?? { + stage: () => Effect.die("unused session.revert.stage"), + clear: () => Effect.die("unused session.revert.clear"), + commit: () => Effect.die("unused session.revert.commit"), + }, context: overrides.session?.context ?? (() => Effect.die("unused session.context")), + form: overrides.session?.form ?? { + list: () => Effect.die("unused session.form.list"), + 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 42780803f9d2..9f7df761e835 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/core/test/session-form-not-found.test.ts b/packages/core/test/session-form-not-found.test.ts new file mode 100644 index 000000000000..e4f287f155d0 --- /dev/null +++ b/packages/core/test/session-form-not-found.test.ts @@ -0,0 +1,60 @@ +import { describe, expect } from "bun:test" +import { Effect, Layer } from "effect" +import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" +import { LayerNode } from "@opencode/util/effect/layer-node" +import { Database } from "@opencode/core/database/database" +import { Bus } from "@opencode/core/bus" +import { Instance } from "@opencode/core/instance/service" +import { Project } from "@opencode/core/project" +import { Session } from "@opencode/core/session" +import { SessionExecution } from "@opencode/core/session/execution" +import { SessionModelTransport } from "@opencode/core/session/model-transport" +import { SessionProjector } from "@opencode/core/session/projector" +import { SessionStore } from "@opencode/core/session/store" +import { SessionEnvironment } from "@opencode/core/session/environment" +import { LocationServiceMap } from "@opencode/core/location-services" +import { testEffect } from "./lib/effect" +import { globalProjectNode } from "./lib/project" +import { offlineModels } from "./fixture/models" + +const transport = Layer.effect( + SessionModelTransport.Service, + Effect.gen(function* () { + return SessionModelTransport.Service.of({ + bind: () => ({ execute: () => Effect.die("Unexpected WebSocket execution") }), + close: () => Effect.void, + closeAll: Effect.void, + }) + }), +) +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([ + Database.node, + Bus.node, + SessionProjector.node, + SessionStore.node, + SessionEnvironment.node, + Session.node, + Instance.node, + LocationServiceMap.node, + ]), + [ + Project.node.replace(globalProjectNode), + SessionExecution.node.replace(SessionExecution.noopLayer), + SessionModelTransport.node.replace(transport), + offlineModels, + ], + ), +) + +describe("Session.form unknown sessions", () => { + it.effect("fails unknown IDs with NotFoundError", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const missing = Session.ID.create() + const error = yield* sessions.form.list({ sessionID: missing }).pipe(Effect.flip) + expect(error).toEqual(new Session.NotFoundError({ sessionID: missing })) + }), + ) +}) diff --git a/packages/core/test/worktree.test.ts b/packages/core/test/worktree.test.ts index 6d2869d442a2..fd298d27585c 100644 --- a/packages/core/test/worktree.test.ts +++ b/packages/core/test/worktree.test.ts @@ -911,7 +911,7 @@ describe("Worktree", () => { const git = yield* WorktreeGit.make yield* worktrees.transform((editor) => editor.add({ ...git, id: Worktree.StrategyID.make("custom") })) yield* ConfigWorktreePlugin.Plugin.effect( - host({ event: { subscribe: () => bus.subscribe().pipe(Stream.filter(EventManifest.isServer)) } }), + host({ event: { subscribe: () => bus.subscribe().pipe(Stream.filter(EventManifest.isServer)), subscribeGlobal: () => Stream.empty } }), ).pipe( Effect.provideService(WorktreeStrategies.Service, { ...strategies, diff --git a/packages/plugin/src/effect/event.ts b/packages/plugin/src/effect/event.ts index d4ab10c0f4bf..3dc39801b724 100644 --- a/packages/plugin/src/effect/event.ts +++ b/packages/plugin/src/effect/event.ts @@ -1,3 +1,16 @@ import type { EventApi } from "@opencode/client/effect/api" +import type { Event } from "@opencode/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 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. Drops are logged + * with a cumulative count; consumers that need gap-free delivery must use + * the durable `log`, not this live bridge. + */ + readonly subscribeGlobal: () => Stream.Stream +} diff --git a/packages/plugin/src/effect/session.ts b/packages/plugin/src/effect/session.ts index bea12a931283..244ffd7c507a 100644 --- a/packages/plugin/src/effect/session.ts +++ b/packages/plugin/src/effect/session.ts @@ -9,6 +9,7 @@ import type { SessionError } from "@opencode/schema/session-error" import type { SessionMessage } from "@opencode/schema/session-message" import type { TokenUsage } from "@opencode/schema/token-usage" import type { JsonSchema, Types } from "effect" +import type { Effect } from "effect" import type { ModelHooks } from "./registration.js" export interface SessionPrompt { @@ -150,6 +151,19 @@ export interface SessionHooks { readonly retry: SessionRetry } +/** Intentional subset of the HTTP session list: in-process only, no cursor/pagination. */ +export interface SessionList { + /** Filter to sessions created in this directory. Must be absolute; relative paths fail. */ + readonly directory?: string + /** 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. Truncated at `limit`. */ +export type SessionListResult = { + readonly data: Session.Info[] +} + export type SessionDomain = Pick< SessionApi, | "create" @@ -164,7 +178,15 @@ export type SessionDomain = Pick< | "update" | "move" | "wait" + | "compact" + | "skill" + | "revert" | "context" > & { + // `list` is intentionally not `SessionApi["list"]`: the HTTP operation + // paginates via an opaque cursor encoding, which has no meaning in-process. + // The subset keeps the data shape and documents the truncation explicitly. + readonly list: (input?: SessionList) => Effect.Effect readonly hook: ModelHooks + readonly form: Pick["form"], "list" | "reply" | "cancel"> } diff --git a/packages/plugin/src/promise/adapter.ts b/packages/plugin/src/promise/adapter.ts index ac0c845f1214..ae2234462785 100644 --- a/packages/plugin/src/promise/adapter.ts +++ b/packages/plugin/src/promise/adapter.ts @@ -329,6 +329,14 @@ export function fromPromise(plugin: Plugin) { ), options, ), + subscribeGlobal: (options) => + streams( + host.event.subscribeGlobal().pipe( + Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)), + Stream.map((event) => event as unknown as PromiseEvent), + ), + options, + ), }, experimental: { terminal: { @@ -573,6 +581,12 @@ export function fromPromise(plugin: Plugin) { register( host.session.hook(name, (event) => Effect.promise(() => Promise.resolve(callback(event))), options), ), + // list bypasses adaptApiMethod intentionally: the in-process + // SessionList is a cursor-free subset of the HTTP SessionsQuery + // (no anchor/cursor encoding), so there is no single endpoint + // schema to decode against. + 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), @@ -585,7 +599,19 @@ export function fromPromise(plugin: Plugin) { update: adaptApiMethod(SessionEndpoints["session.update"], host.session.update), move: adaptApiMethod(SessionEndpoints["session.move"], host.session.move), wait: adaptApiMethod(SessionEndpoints["session.wait"], host.session.wait), + compact: adaptApiMethod(SessionEndpoints["session.compact"], host.session.compact), + skill: adaptApiMethod(SessionEndpoints["session.skill"], host.session.skill), + revert: { + stage: adaptApiMethod(SessionEndpoints["session.revert.stage"], host.session.revert.stage), + clear: adaptApiMethod(SessionEndpoints["session.revert.clear"], host.session.revert.clear), + commit: adaptApiMethod(SessionEndpoints["session.revert.commit"], host.session.revert.commit), + }, context: adaptApiMethod(SessionEndpoints["session.context"], host.session.context), + form: { + list: adaptApiMethod(SessionEndpoints["session.form.list"], host.session.form.list), + reply: adaptApiMethod(SessionEndpoints["session.form.reply"], host.session.form.reply), + cancel: adaptApiMethod(SessionEndpoints["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 74ac6e5fde2e..cc443231c7b3 100644 --- a/packages/plugin/src/promise/event.ts +++ b/packages/plugin/src/promise/event.ts @@ -1,3 +1,15 @@ import type { EventApi } from "@opencode/client/promise/api" +import type { OpenCodeEvent } from "@opencode/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 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. Drops are logged + * with a cumulative count; consumers that need gap-free delivery must use + * the durable `log`, not this live bridge. + */ + readonly subscribeGlobal: (options?: { readonly signal?: AbortSignal }) => AsyncIterable +} diff --git a/packages/plugin/src/promise/session.ts b/packages/plugin/src/promise/session.ts index ac3aa3888a5b..807fd033c6e0 100644 --- a/packages/plugin/src/promise/session.ts +++ b/packages/plugin/src/promise/session.ts @@ -1,6 +1,7 @@ import type { SessionApi } from "@opencode/client/promise/api" import type { GenerationOptionsFields, Message, SystemPart } from "@opencode/ai" import type { Agent } from "@opencode/schema/agent" +import type { Form } from "@opencode/schema/form" import type { Model } from "@opencode/schema/model" import type { PromptInput } from "@opencode/schema/prompt-input" import type { Session } from "@opencode/schema/session" @@ -150,6 +151,19 @@ 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. Must be absolute; relative paths reject. */ + readonly directory?: string + /** 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. Truncated at `limit`. */ +export type SessionListResult = { + readonly data: Session.Info[] +} + export type SessionDomain = Pick< SessionApi, | "create" @@ -164,7 +178,22 @@ export type SessionDomain = Pick< | "update" | "move" | "wait" + | "compact" + | "skill" + | "revert" | "context" > & { + readonly list: (input?: SessionList) => Promise readonly hook: ModelHooks + readonly form: FormDomain +} + +export interface FormDomain { + readonly list: (input: { readonly sessionID: 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/location.ts b/packages/server/src/location.ts index 83e66df301dd..b9b10bb0c28d 100644 --- a/packages/server/src/location.ts +++ b/packages/server/src/location.ts @@ -36,11 +36,23 @@ export const sessionInfo = Effect.fnUntraced(function* (sessions: Session.Interf return yield* sessions.get(id).pipe(Effect.catchTag("Session.NotFoundError", missingSession)) }) +/** + * The location used when a request names none: the directory the serve + * command was launched from. This is the only directory known before any + * request identifies a project. + */ +export function defaultRef(): Location.Ref { + return Location.Ref.make({ + directory: AbsolutePath.make(process.cwd()), + }) +} + export function requestRef(request: HttpServerRequest.HttpServerRequest): Location.Ref { const query = new URL(request.url, "http://localhost").searchParams const directory = query.get("location[directory]") || - (request.headers["x-opencode-directory"] ? decode(request.headers["x-opencode-directory"]) : process.cwd()) + (request.headers["x-opencode-directory"] ? decode(request.headers["x-opencode-directory"]) : undefined) + if (directory === undefined) return defaultRef() return Location.Ref.make({ directory: AbsolutePath.make(directory), }) diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index 3a70dbd39bf3..1ebdab4eb084 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -2,6 +2,8 @@ export * as ServerProcess from "./process" import { NodeHttpServer } from "@effect/platform-node" import { Bus } from "@opencode/core/bus" +import { LocationServiceMap } from "@opencode/core/location-services" +import { Plugin } from "@opencode/core/plugin" import { SessionRestart } from "@opencode/core/session/execution/restart" import { InstallationEvent } from "@opencode/schema/installation-event" import { hasPtyConnectTicketURL } from "@opencode/protocol/groups/pty" @@ -22,6 +24,7 @@ import { isAllowedCorsOrigin } from "./cors" import { authorizedRequest } from "./middleware/authorization" import { withoutParentSpan } from "./request-tracing" import { createRoutes } from "./routes" +import { defaultRef } from "./location" import { ServerInfo } from "./server-info" import { Status } from "./service-status" import type { ServerOptions } from "./options" @@ -105,6 +108,55 @@ export const start = Effect.fn("ServerProcess.start")(function* ( ).pipe(Layer.provideMerge(NodeHttpServer.layerHttpServices)), applicationScope, ) + + // Eagerly build the default location (see `defaultRef` in `./location.ts`) + // 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. Any project can still be targeted + // per-request via the `x-opencode-directory` header, which bootstraps that + // location's instance on demand. + // + // The boot runs in the background and never blocks server readiness: a + // valid-but-slow project must not delay `ready`, so the 10s activation cap + // below only bounds how long the background task waits for plugins, not + // startup. A `/` cwd (systemd/cron accident) is skipped outright — the + // filesystem root is never a meaningful project location, and lazy + // per-request boot covers real projects. + yield* Effect.gen(function* () { + if (process.cwd() === "/") { + yield* Effect.logWarning( + "Skipping eager default-location boot: process cwd is /, waiting for the first location-scoped request instead", + ) + return yield* Effect.void + } + const locations = yield* LocationServiceMap.Service + const services = locations.get(defaultRef()) + 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 comes up promptly in the background. + 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), + ), + // 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 }), + ), + Effect.forkIn(applicationScope), + ) + if (lifecycle) { yield* installRestartContinuity(Context.get(context, SessionRestart.Service)).pipe( Effect.provideService(Scope.Scope, applicationScope), diff --git a/services/www/src/docs/content/build/plugins/index.mdx b/services/www/src/docs/content/build/plugins/index.mdx index 279cb1130109..76b8a8200323 100644 --- a/services/www/src/docs/content/build/plugins/index.mdx +++ b/services/www/src/docs/content/build/plugins/index.mdx @@ -727,14 +727,14 @@ Inspect or resolve pending permission requests. ```ts const pending = await ctx.permission.list({ sessionID }) const request = await ctx.permission.get({ sessionID, requestID }) -await ctx.permission.reply({ sessionID, requestID, reply: "once" }) +await ctx.permission.reply({ sessionID, requestID, decision: "once" }) ``` -Replace the session-scoped permission rules. They are evaluated after the agent's rules, and the +Replace the session-scoped permission rules with `ctx.session.update`. They are evaluated after the agent's rules, and the last matching rule wins. Child sessions inherit the rules in effect when they are created. ```ts -await ctx.permission.rules({ +await ctx.session.update({ sessionID, permissions: [{ action: "edit", resource: "/path/to/original/checkout/**", effect: "deny" }], })