diff --git a/packages/opencode/src/plugin/index.ts b/packages/opencode/src/plugin/index.ts index 3bdcf3e17e..b00277e97f 100644 --- a/packages/opencode/src/plugin/index.ts +++ b/packages/opencode/src/plugin/index.ts @@ -11,7 +11,7 @@ import { Session } from "../session" import { NamedError } from "@opencode-ai/util/error" import { CopilotAuthPlugin } from "./copilot" import { gitlabAuthPlugin as GitlabAuthPlugin } from "opencode-gitlab-auth" -import { Effect, Layer, ServiceMap } from "effect" +import { Effect, Layer, ServiceMap, Stream } from "effect" import { InstanceState } from "@/effect/instance-state" import { makeRuntime } from "@/effect/run-service" @@ -52,6 +52,8 @@ export namespace Plugin { export const layer = Layer.effect( Service, Effect.gen(function* () { + const bus = yield* Bus.Service + const cache = yield* InstanceState.make( Effect.fn("Plugin.state")(function* (ctx) { const hooks: Hooks[] = [] @@ -140,17 +142,19 @@ export namespace Plugin { } }) - // Subscribe to bus events, clean up when scope is closed - yield* Effect.acquireRelease( - Effect.sync(() => - Bus.subscribeAll(async (input) => { - for (const hook of hooks) { - hook["event"]?.({ event: input }) - } - }), - ), - (unsub) => Effect.sync(unsub), - ) + // Subscribe to bus events, fiber interrupted when scope closes + yield* bus + .subscribeAll() + .pipe( + Stream.runForEach((input) => + Effect.sync(() => { + for (const hook of hooks) { + hook["event"]?.({ event: input as any }) + } + }), + ), + Effect.forkScoped, + ) return { hooks } }), @@ -186,7 +190,8 @@ export namespace Plugin { }), ) - const { runPromise } = makeRuntime(Service, layer) + const defaultLayer = layer.pipe(Layer.provide(Bus.layer)) + const { runPromise } = makeRuntime(Service, defaultLayer) export async function trigger< Name extends TriggerName, diff --git a/packages/opencode/src/project/vcs.ts b/packages/opencode/src/project/vcs.ts index 8a3789db80..3986313443 100644 --- a/packages/opencode/src/project/vcs.ts +++ b/packages/opencode/src/project/vcs.ts @@ -1,4 +1,4 @@ -import { Effect, Layer, ServiceMap } from "effect" +import { Effect, Layer, ServiceMap, Stream } from "effect" import { Bus } from "@/bus" import { BusEvent } from "@/bus/bus-event" import { InstanceState } from "@/effect/instance-state" @@ -44,6 +44,8 @@ export namespace Vcs { export const layer = Layer.effect( Service, Effect.gen(function* () { + const bus = yield* Bus.Service + const state = yield* InstanceState.make( Effect.fn("Vcs.state")((ctx) => Effect.gen(function* () { @@ -65,23 +67,22 @@ export namespace Vcs { } log.info("initialized", { branch: value.current }) - yield* Effect.acquireRelease( - Effect.sync(() => - Bus.subscribe( - FileWatcher.Event.Updated, - Instance.bind(async (evt) => { - if (!evt.properties.file.endsWith("HEAD")) return - const next = await getCurrentBranch() + yield* bus + .subscribe(FileWatcher.Event.Updated) + .pipe( + Stream.filter((evt) => evt.properties.file.endsWith("HEAD")), + Stream.runForEach((evt) => + Effect.gen(function* () { + const next = yield* Effect.promise(() => getCurrentBranch()) if (next !== value.current) { log.info("branch changed", { from: value.current, to: next }) value.current = next - Bus.publish(Event.BranchUpdated, { branch: next }) + yield* bus.publish(Event.BranchUpdated, { branch: next }) } }), ), - ), - (unsubscribe) => Effect.sync(unsubscribe), - ) + Effect.forkScoped, + ) return value }), @@ -99,7 +100,8 @@ export namespace Vcs { }), ) - const { runPromise } = makeRuntime(Service, layer) + const defaultLayer = layer.pipe(Layer.provide(Bus.layer)) + const { runPromise } = makeRuntime(Service, defaultLayer) export function init() { return runPromise((svc) => svc.init()) diff --git a/packages/opencode/src/session/status.ts b/packages/opencode/src/session/status.ts index 2d79fa3850..d0ca78848c 100644 --- a/packages/opencode/src/session/status.ts +++ b/packages/opencode/src/session/status.ts @@ -6,6 +6,7 @@ import { SessionID } from "./schema" import { Effect, Layer, ServiceMap } from "effect" import z from "zod" + export namespace SessionStatus { export const Info = z .union([ @@ -55,6 +56,8 @@ export namespace SessionStatus { export const layer = Layer.effect( Service, Effect.gen(function* () { + const bus = yield* Bus.Service + const state = yield* InstanceState.make( Effect.fn("SessionStatus.state")(() => Effect.succeed(new Map())), ) @@ -70,9 +73,9 @@ export namespace SessionStatus { const set = Effect.fn("SessionStatus.set")(function* (sessionID: SessionID, status: Info) { const data = yield* InstanceState.get(state) - yield* Effect.promise(() => Bus.publish(Event.Status, { sessionID, status })) + yield* bus.publish(Event.Status, { sessionID, status }) if (status.type === "idle") { - yield* Effect.promise(() => Bus.publish(Event.Idle, { sessionID })) + yield* bus.publish(Event.Idle, { sessionID }) data.delete(sessionID) return } @@ -83,7 +86,8 @@ export namespace SessionStatus { }), ) - const { runPromise } = makeRuntime(Service, layer) + const defaultLayer = layer.pipe(Layer.provide(Bus.layer)) + const { runPromise } = makeRuntime(Service, defaultLayer) export async function get(sessionID: SessionID) { return runPromise((svc) => svc.get(sessionID)) diff --git a/packages/opencode/test/bus/bus-integration.test.ts b/packages/opencode/test/bus/bus-integration.test.ts new file mode 100644 index 0000000000..e42bd5299e --- /dev/null +++ b/packages/opencode/test/bus/bus-integration.test.ts @@ -0,0 +1,87 @@ +import { afterEach, describe, expect, test } from "bun:test" +import z from "zod" +import { Bus } from "../../src/bus" +import { BusEvent } from "../../src/bus/bus-event" +import { Instance } from "../../src/project/instance" +import { tmpdir } from "../fixture/fixture" + +const TestEvent = BusEvent.define("test.integration", z.object({ value: z.number() })) + +function withInstance(directory: string, fn: () => Promise) { + return Instance.provide({ directory, fn }) +} + +describe("Bus integration: acquireRelease subscriber pattern", () => { + afterEach(() => Instance.disposeAll()) + + test("subscriber via callback facade receives events and cleans up on unsub", async () => { + await using tmp = await tmpdir() + const received: number[] = [] + + await withInstance(tmp.path, async () => { + const unsub = Bus.subscribe(TestEvent, (evt) => { + received.push(evt.properties.value) + }) + await Bun.sleep(10) + await Bus.publish(TestEvent, { value: 1 }) + await Bus.publish(TestEvent, { value: 2 }) + await Bun.sleep(10) + + expect(received).toEqual([1, 2]) + + unsub() + await Bun.sleep(10) + await Bus.publish(TestEvent, { value: 3 }) + await Bun.sleep(10) + + expect(received).toEqual([1, 2]) + }) + }) + + test("subscribeAll receives events from multiple types", async () => { + await using tmp = await tmpdir() + const received: Array<{ type: string; value?: number }> = [] + + const OtherEvent = BusEvent.define("test.other", z.object({ value: z.number() })) + + await withInstance(tmp.path, async () => { + Bus.subscribeAll((evt) => { + received.push({ type: evt.type, value: evt.properties.value }) + }) + await Bun.sleep(10) + await Bus.publish(TestEvent, { value: 10 }) + await Bus.publish(OtherEvent, { value: 20 }) + await Bun.sleep(10) + }) + + expect(received).toEqual([ + { type: "test.integration", value: 10 }, + { type: "test.other", value: 20 }, + ]) + }) + + test("subscriber cleanup on instance disposal interrupts the stream", async () => { + await using tmp = await tmpdir() + const received: number[] = [] + let disposed = false + + await withInstance(tmp.path, async () => { + Bus.subscribeAll((evt) => { + if (evt.type === Bus.InstanceDisposed.type) { + disposed = true + return + } + received.push(evt.properties.value) + }) + await Bun.sleep(10) + await Bus.publish(TestEvent, { value: 1 }) + await Bun.sleep(10) + }) + + await Instance.disposeAll() + await Bun.sleep(50) + + expect(received).toEqual([1]) + expect(disposed).toBe(true) + }) +}) diff --git a/packages/opencode/test/fixture/instance.ts b/packages/opencode/test/fixture/instance.ts index 67af82fc8b..776ed98a1c 100644 --- a/packages/opencode/test/fixture/instance.ts +++ b/packages/opencode/test/fixture/instance.ts @@ -1,5 +1,6 @@ import { ConfigProvider, Layer, ManagedRuntime } from "effect" import { InstanceContext } from "../../src/effect/instance-context" +import { memoMap } from "../../src/effect/run-service" import { Instance } from "../../src/project/instance" /** ConfigProvider that enables the experimental file watcher. */ diff --git a/packages/opencode/test/project/vcs.test.ts b/packages/opencode/test/project/vcs.test.ts index 11463b7950..868d8cb6eb 100644 --- a/packages/opencode/test/project/vcs.test.ts +++ b/packages/opencode/test/project/vcs.test.ts @@ -2,9 +2,7 @@ import { $ } from "bun" import { afterEach, describe, expect, test } from "bun:test" import fs from "fs/promises" import path from "path" -import { Effect, Layer, ManagedRuntime } from "effect" import { tmpdir } from "../fixture/fixture" -import { watcherConfigLayer, withServices } from "../fixture/instance" import { FileWatcher } from "../../src/file/watcher" import { Instance } from "../../src/project/instance" import { GlobalBus } from "../../src/bus/global" @@ -17,21 +15,16 @@ const describeVcs = FileWatcher.hasNativeBinding() && !process.env.CI ? describe // Helpers // --------------------------------------------------------------------------- -function withVcs( - directory: string, - body: (rt: ManagedRuntime.ManagedRuntime) => Promise, -) { - return withServices( +async function withVcs(directory: string, body: () => Promise) { + return Instance.provide({ directory, - Layer.merge(FileWatcher.layer, Vcs.layer), - async (rt) => { - await rt.runPromise(FileWatcher.Service.use((s) => s.init())) - await rt.runPromise(Vcs.Service.use((s) => s.init())) + fn: async () => { + FileWatcher.init() + Vcs.init() await Bun.sleep(500) - await body(rt) + await body() }, - { provide: [watcherConfigLayer] }, - ) + }) } type BranchEvent = { directory?: string; payload: { type: string; properties: { branch?: string } } } @@ -74,8 +67,8 @@ describeVcs("Vcs", () => { test("branch() returns current branch name", async () => { await using tmp = await tmpdir({ git: true }) - await withVcs(tmp.path, async (rt) => { - const branch = await rt.runPromise(Vcs.Service.use((s) => s.branch())) + await withVcs(tmp.path, async () => { + const branch = await Vcs.branch() expect(branch).toBeDefined() expect(typeof branch).toBe("string") }) @@ -84,8 +77,8 @@ describeVcs("Vcs", () => { test("branch() returns undefined for non-git directories", async () => { await using tmp = await tmpdir() - await withVcs(tmp.path, async (rt) => { - const branch = await rt.runPromise(Vcs.Service.use((s) => s.branch())) + await withVcs(tmp.path, async () => { + const branch = await Vcs.branch() expect(branch).toBeUndefined() }) }) @@ -111,14 +104,14 @@ describeVcs("Vcs", () => { const branch = `test-${Math.random().toString(36).slice(2)}` await $`git branch ${branch}`.cwd(tmp.path).quiet() - await withVcs(tmp.path, async (rt) => { + await withVcs(tmp.path, async () => { const pending = nextBranchUpdate(tmp.path) const head = path.join(tmp.path, ".git", "HEAD") await fs.writeFile(head, `ref: refs/heads/${branch}\n`) await pending - const current = await rt.runPromise(Vcs.Service.use((s) => s.branch())) + const current = await Vcs.branch() expect(current).toBe(branch) }) })