effectify Bus service: migrate to Effect PubSub + InstanceState (#18579)

2026-03-25 20:19:24 -04:00
parent b90de755f9
commit 97c15a087d
42 changed files with 1113 additions and 522 deletions
--- a/packages/opencode/test/bus/bus-effect.test.ts
+++ b/packages/opencode/test/bus/bus-effect.test.ts
@@ -0,0 +1,164 @@
+import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"
+import { describe, expect } from "bun:test"
+import { Deferred, Effect, Layer, Stream } from "effect"
+import z from "zod"
+import { Bus } from "../../src/bus"
+import { BusEvent } from "../../src/bus/bus-event"
+import { Instance } from "../../src/project/instance"
+import { provideInstance, provideTmpdirInstance, tmpdirScoped } from "../fixture/fixture"
+import { testEffect } from "../lib/effect"
+
+const TestEvent = {
+  Ping: BusEvent.define("test.effect.ping", z.object({ value: z.number() })),
+  Pong: BusEvent.define("test.effect.pong", z.object({ message: z.string() })),
+}
+
+const node = NodeChildProcessSpawner.layer.pipe(
+  Layer.provideMerge(Layer.mergeAll(NodeFileSystem.layer, NodePath.layer)),
+)
+
+const live = Layer.mergeAll(Bus.layer, node)
+
+const it = testEffect(live)
+
+describe("Bus (Effect-native)", () => {
+  it.effect("publish + subscribe stream delivers events", () =>
+    provideTmpdirInstance(() =>
+      Effect.gen(function* () {
+        const bus = yield* Bus.Service
+        const received: number[] = []
+        const done = yield* Deferred.make<void>()
+
+        yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+          Effect.sync(() => {
+            received.push(evt.properties.value)
+            if (received.length === 2) Deferred.doneUnsafe(done, Effect.void)
+          }),
+        ).pipe(Effect.forkScoped)
+
+        yield* Effect.sleep("10 millis")
+        yield* bus.publish(TestEvent.Ping, { value: 1 })
+        yield* bus.publish(TestEvent.Ping, { value: 2 })
+        yield* Deferred.await(done)
+
+        expect(received).toEqual([1, 2])
+      }),
+    ),
+  )
+
+  it.effect("subscribe filters by event type", () =>
+    provideTmpdirInstance(() =>
+      Effect.gen(function* () {
+        const bus = yield* Bus.Service
+        const pings: number[] = []
+        const done = yield* Deferred.make<void>()
+
+        yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+          Effect.sync(() => {
+            pings.push(evt.properties.value)
+            Deferred.doneUnsafe(done, Effect.void)
+          }),
+        ).pipe(Effect.forkScoped)
+
+        yield* Effect.sleep("10 millis")
+        yield* bus.publish(TestEvent.Pong, { message: "ignored" })
+        yield* bus.publish(TestEvent.Ping, { value: 42 })
+        yield* Deferred.await(done)
+
+        expect(pings).toEqual([42])
+      }),
+    ),
+  )
+
+  it.effect("subscribeAll receives all types", () =>
+    provideTmpdirInstance(() =>
+      Effect.gen(function* () {
+        const bus = yield* Bus.Service
+        const types: string[] = []
+        const done = yield* Deferred.make<void>()
+
+        yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
+          Effect.sync(() => {
+            types.push(evt.type)
+            if (types.length === 2) Deferred.doneUnsafe(done, Effect.void)
+          }),
+        ).pipe(Effect.forkScoped)
+
+        yield* Effect.sleep("10 millis")
+        yield* bus.publish(TestEvent.Ping, { value: 1 })
+        yield* bus.publish(TestEvent.Pong, { message: "hi" })
+        yield* Deferred.await(done)
+
+        expect(types).toContain("test.effect.ping")
+        expect(types).toContain("test.effect.pong")
+      }),
+    ),
+  )
+
+  it.effect("multiple subscribers each receive the event", () =>
+    provideTmpdirInstance(() =>
+      Effect.gen(function* () {
+        const bus = yield* Bus.Service
+        const a: number[] = []
+        const b: number[] = []
+        const doneA = yield* Deferred.make<void>()
+        const doneB = yield* Deferred.make<void>()
+
+        yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+          Effect.sync(() => {
+            a.push(evt.properties.value)
+            Deferred.doneUnsafe(doneA, Effect.void)
+          }),
+        ).pipe(Effect.forkScoped)
+
+        yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+          Effect.sync(() => {
+            b.push(evt.properties.value)
+            Deferred.doneUnsafe(doneB, Effect.void)
+          }),
+        ).pipe(Effect.forkScoped)
+
+        yield* Effect.sleep("10 millis")
+        yield* bus.publish(TestEvent.Ping, { value: 99 })
+        yield* Deferred.await(doneA)
+        yield* Deferred.await(doneB)
+
+        expect(a).toEqual([99])
+        expect(b).toEqual([99])
+      }),
+    ),
+  )
+
+  it.effect("subscribeAll stream sees InstanceDisposed on disposal", () =>
+    Effect.gen(function* () {
+      const dir = yield* tmpdirScoped()
+      const types: string[] = []
+      const seen = yield* Deferred.make<void>()
+      const disposed = yield* Deferred.make<void>()
+
+      // Set up subscriber inside the instance
+      yield* Effect.gen(function* () {
+        const bus = yield* Bus.Service
+
+        yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
+          Effect.sync(() => {
+            types.push(evt.type)
+            if (evt.type === TestEvent.Ping.type) Deferred.doneUnsafe(seen, Effect.void)
+            if (evt.type === Bus.InstanceDisposed.type) Deferred.doneUnsafe(disposed, Effect.void)
+          }),
+        ).pipe(Effect.forkScoped)
+
+        yield* Effect.sleep("10 millis")
+        yield* bus.publish(TestEvent.Ping, { value: 1 })
+        yield* Deferred.await(seen)
+      }).pipe(provideInstance(dir))
+
+      // Dispose from OUTSIDE the instance scope
+      yield* Effect.promise(() => Instance.disposeAll())
+      yield* Deferred.await(disposed).pipe(Effect.timeout("2 seconds"))
+
+      expect(types).toContain("test.effect.ping")
+      expect(types).toContain(Bus.InstanceDisposed.type)
+    }),
+  )
+})
--- a/packages/opencode/test/bus/bus-integration.test.ts
+++ 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<void>) {
+  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)
+  })
+})
--- a/packages/opencode/test/bus/bus.test.ts
+++ b/packages/opencode/test/bus/bus.test.ts
@@ -0,0 +1,219 @@
+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 = {
+  Ping: BusEvent.define("test.ping", z.object({ value: z.number() })),
+  Pong: BusEvent.define("test.pong", z.object({ message: z.string() })),
+}
+
+function withInstance(directory: string, fn: () => Promise<void>) {
+  return Instance.provide({ directory, fn })
+}
+
+describe("Bus", () => {
+  afterEach(() => Instance.disposeAll())
+
+  describe("publish + subscribe", () => {
+    test("subscriber is live immediately after subscribe returns", async () => {
+      await using tmp = await tmpdir()
+      const received: number[] = []
+
+      await withInstance(tmp.path, async () => {
+        Bus.subscribe(TestEvent.Ping, (evt) => {
+          received.push(evt.properties.value)
+        })
+        await Bus.publish(TestEvent.Ping, { value: 42 })
+        await Bun.sleep(10)
+      })
+
+      expect(received).toEqual([42])
+    })
+
+    test("subscriber receives matching events", async () => {
+      await using tmp = await tmpdir()
+      const received: number[] = []
+
+      await withInstance(tmp.path, async () => {
+        Bus.subscribe(TestEvent.Ping, (evt) => {
+          received.push(evt.properties.value)
+        })
+        // Give the subscriber fiber time to start consuming
+        await Bun.sleep(10)
+        await Bus.publish(TestEvent.Ping, { value: 42 })
+        await Bus.publish(TestEvent.Ping, { value: 99 })
+        // Give subscriber time to process
+        await Bun.sleep(10)
+      })
+
+      expect(received).toEqual([42, 99])
+    })
+
+    test("subscriber does not receive events of other types", async () => {
+      await using tmp = await tmpdir()
+      const pings: number[] = []
+
+      await withInstance(tmp.path, async () => {
+        Bus.subscribe(TestEvent.Ping, (evt) => {
+          pings.push(evt.properties.value)
+        })
+        await Bun.sleep(10)
+        await Bus.publish(TestEvent.Pong, { message: "hello" })
+        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await Bun.sleep(10)
+      })
+
+      expect(pings).toEqual([1])
+    })
+
+    test("publish with no subscribers does not throw", async () => {
+      await using tmp = await tmpdir()
+
+      await withInstance(tmp.path, async () => {
+        await Bus.publish(TestEvent.Ping, { value: 1 })
+      })
+    })
+  })
+
+  describe("unsubscribe", () => {
+    test("unsubscribe stops delivery", async () => {
+      await using tmp = await tmpdir()
+      const received: number[] = []
+
+      await withInstance(tmp.path, async () => {
+        const unsub = Bus.subscribe(TestEvent.Ping, (evt) => {
+          received.push(evt.properties.value)
+        })
+        await Bun.sleep(10)
+        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await Bun.sleep(10)
+        unsub()
+        await Bun.sleep(10)
+        await Bus.publish(TestEvent.Ping, { value: 2 })
+        await Bun.sleep(10)
+      })
+
+      expect(received).toEqual([1])
+    })
+  })
+
+  describe("subscribeAll", () => {
+    test("subscribeAll is live immediately after subscribe returns", async () => {
+      await using tmp = await tmpdir()
+      const received: string[] = []
+
+      await withInstance(tmp.path, async () => {
+        Bus.subscribeAll((evt) => {
+          received.push(evt.type)
+        })
+        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await Bun.sleep(10)
+      })
+
+      expect(received).toEqual(["test.ping"])
+    })
+
+    test("receives all event types", async () => {
+      await using tmp = await tmpdir()
+      const received: string[] = []
+
+      await withInstance(tmp.path, async () => {
+        Bus.subscribeAll((evt) => {
+          received.push(evt.type)
+        })
+        await Bun.sleep(10)
+        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await Bus.publish(TestEvent.Pong, { message: "hi" })
+        await Bun.sleep(10)
+      })
+
+      expect(received).toContain("test.ping")
+      expect(received).toContain("test.pong")
+    })
+  })
+
+  describe("multiple subscribers", () => {
+    test("all subscribers for same event type are called", async () => {
+      await using tmp = await tmpdir()
+      const a: number[] = []
+      const b: number[] = []
+
+      await withInstance(tmp.path, async () => {
+        Bus.subscribe(TestEvent.Ping, (evt) => {
+          a.push(evt.properties.value)
+        })
+        Bus.subscribe(TestEvent.Ping, (evt) => {
+          b.push(evt.properties.value)
+        })
+        await Bun.sleep(10)
+        await Bus.publish(TestEvent.Ping, { value: 7 })
+        await Bun.sleep(10)
+      })
+
+      expect(a).toEqual([7])
+      expect(b).toEqual([7])
+    })
+  })
+
+  describe("instance isolation", () => {
+    test("events in one directory do not reach subscribers in another", async () => {
+      await using tmpA = await tmpdir()
+      await using tmpB = await tmpdir()
+      const receivedA: number[] = []
+      const receivedB: number[] = []
+
+      await withInstance(tmpA.path, async () => {
+        Bus.subscribe(TestEvent.Ping, (evt) => {
+          receivedA.push(evt.properties.value)
+        })
+        await Bun.sleep(10)
+      })
+
+      await withInstance(tmpB.path, async () => {
+        Bus.subscribe(TestEvent.Ping, (evt) => {
+          receivedB.push(evt.properties.value)
+        })
+        await Bun.sleep(10)
+      })
+
+      await withInstance(tmpA.path, async () => {
+        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await Bun.sleep(10)
+      })
+
+      await withInstance(tmpB.path, async () => {
+        await Bus.publish(TestEvent.Ping, { value: 2 })
+        await Bun.sleep(10)
+      })
+
+      expect(receivedA).toEqual([1])
+      expect(receivedB).toEqual([2])
+    })
+  })
+
+  describe("instance disposal", () => {
+    test("InstanceDisposed is delivered to wildcard subscribers before stream ends", async () => {
+      await using tmp = await tmpdir()
+      const received: string[] = []
+
+      await withInstance(tmp.path, async () => {
+        Bus.subscribeAll((evt) => {
+          received.push(evt.type)
+        })
+        await Bun.sleep(10)
+        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await Bun.sleep(10)
+      })
+
+      // Instance.disposeAll triggers the finalizer which publishes InstanceDisposed
+      await Instance.disposeAll()
+      await Bun.sleep(50)
+
+      expect(received).toContain("test.ping")
+      expect(received).toContain(Bus.InstanceDisposed.type)
+    })
+  })
+})
--- a/packages/opencode/test/effect/run-service.test.ts
+++ b/packages/opencode/test/effect/run-service.test.ts
@@ -1,10 +1,10 @@
 import { expect, test } from "bun:test"
 import { Effect, Layer, ServiceMap } from "effect"
-import { makeRunPromise } from "../../src/effect/run-service"
+import { makeRuntime } from "../../src/effect/run-service"

 class Shared extends ServiceMap.Service<Shared, { readonly id: number }>()("@test/Shared") {}

-test("makeRunPromise shares dependent layers through the shared memo map", async () => {
+test("makeRuntime shares dependent layers through the shared memo map", async () => {
  let n = 0

  const shared = Layer.effect(
@@ -37,8 +37,8 @@ test("makeRunPromise shares dependent layers through the shared memo map", async
    }),
  ).pipe(Layer.provide(shared))

-  const runOne = makeRunPromise(One, one)
-  const runTwo = makeRunPromise(Two, two)
+  const { runPromise: runOne } = makeRuntime(One, one)
+  const { runPromise: runTwo } = makeRuntime(Two, two)

  expect(await runOne((svc) => svc.get())).toBe(1)
  expect(await runTwo((svc) => svc.get())).toBe(1)
--- a/packages/opencode/test/file/watcher.test.ts
+++ b/packages/opencode/test/file/watcher.test.ts
@@ -2,9 +2,8 @@ import { $ } from "bun"
 import { afterEach, describe, expect, test } from "bun:test"
 import fs from "fs/promises"
 import path from "path"
-import { Deferred, Effect, Option } from "effect"
+import { ConfigProvider, Deferred, Effect, Layer, ManagedRuntime, Option } from "effect"
 import { tmpdir } from "../fixture/fixture"
-import { watcherConfigLayer, withServices } from "../fixture/instance"
 import { Bus } from "../../src/bus"
 import { FileWatcher } from "../../src/file/watcher"
 import { Instance } from "../../src/project/instance"
@@ -16,20 +15,33 @@ const describeWatcher = FileWatcher.hasNativeBinding() && !process.env.CI ? desc
 // Helpers
 // ---------------------------------------------------------------------------

+const watcherConfigLayer = ConfigProvider.layer(
+  ConfigProvider.fromUnknown({
+    OPENCODE_EXPERIMENTAL_FILEWATCHER: "true",
+    OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "false",
+  }),
+)
+
 type WatcherEvent = { file: string; event: "add" | "change" | "unlink" }

 /** Run `body` with a live FileWatcher service. */
 function withWatcher<E>(directory: string, body: Effect.Effect<void, E>) {
-  return withServices(
+  return Instance.provide({
    directory,
-    FileWatcher.layer,
-    async (rt) => {
-      await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
-      await Effect.runPromise(ready(directory))
-      await Effect.runPromise(body)
+    fn: async () => {
+      const layer: Layer.Layer<FileWatcher.Service, never, never> = FileWatcher.layer.pipe(
+        Layer.provide(watcherConfigLayer),
+      )
+      const rt = ManagedRuntime.make(layer)
+      try {
+        await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
+        await Effect.runPromise(ready(directory))
+        await Effect.runPromise(body)
+      } finally {
+        await rt.dispose()
+      }
    },
-    { provide: [watcherConfigLayer] },
-  )
+  })
 }

 function listen(directory: string, check: (evt: WatcherEvent) => boolean, hit: (evt: WatcherEvent) => void) {
--- a/packages/opencode/test/fixture/fixture.ts
+++ b/packages/opencode/test/fixture/fixture.ts
@@ -2,7 +2,10 @@ import { $ } from "bun"
 import * as fs from "fs/promises"
 import os from "os"
 import path from "path"
+import { Effect, FileSystem, ServiceMap } from "effect"
+import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
 import type { Config } from "../../src/config/config"
+import { Instance } from "../../src/project/instance"

 // Strip null bytes from paths (defensive fix for CI environment issues)
 function sanitizePath(p: string): string {
@@ -71,3 +74,68 @@ export async function tmpdir<T>(options?: TmpDirOptions<T>) {
  }
  return result
 }
+
+/** Effectful scoped tmpdir. Cleaned up when the scope closes. Make sure these stay in sync */
+export function tmpdirScoped(options?: { git?: boolean; config?: Partial<Config.Info> }) {
+  return Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem
+    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
+    const dir = yield* fs.makeTempDirectoryScoped({ prefix: "opencode-test-" })
+
+    const git = (...args: string[]) =>
+      spawner.spawn(ChildProcess.make("git", args, { cwd: dir })).pipe(Effect.flatMap((handle) => handle.exitCode))
+
+    if (options?.git) {
+      yield* git("init")
+      yield* git("config", "core.fsmonitor", "false")
+      yield* git("config", "user.email", "test@opencode.test")
+      yield* git("config", "user.name", "Test")
+      yield* git("commit", "--allow-empty", "-m", "root commit")
+    }
+
+    if (options?.config) {
+      yield* fs.writeFileString(
+        path.join(dir, "opencode.json"),
+        JSON.stringify({ $schema: "https://opencode.ai/config.json", ...options.config }),
+      )
+    }
+
+    return dir
+  })
+}
+
+export const provideInstance =
+  (directory: string) =>
+  <A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
+    Effect.servicesWith((services: ServiceMap.ServiceMap<R>) =>
+      Effect.promise<A>(async () =>
+        Instance.provide({
+          directory,
+          fn: () => Effect.runPromiseWith(services)(self),
+        }),
+      ),
+    )
+
+export function provideTmpdirInstance<A, E, R>(
+  self: (path: string) => Effect.Effect<A, E, R>,
+  options?: { git?: boolean; config?: Partial<Config.Info> },
+) {
+  return Effect.gen(function* () {
+    const path = yield* tmpdirScoped(options)
+    let provided = false
+
+    yield* Effect.addFinalizer(() =>
+      provided
+        ? Effect.promise(() =>
+            Instance.provide({
+              directory: path,
+              fn: () => Instance.dispose(),
+            }),
+          ).pipe(Effect.ignore)
+        : Effect.void,
+    )
+
+    provided = true
+    return yield* self(path).pipe(provideInstance(path))
+  })
+}
--- a/packages/opencode/test/fixture/instance.ts
+++ b/packages/opencode/test/fixture/instance.ts
@@ -1,51 +0,0 @@
-import { ConfigProvider, Layer, ManagedRuntime } from "effect"
-import { InstanceContext } from "../../src/effect/instance-context"
-import { Instance } from "../../src/project/instance"
-
-/** ConfigProvider that enables the experimental file watcher. */
-export const watcherConfigLayer = ConfigProvider.layer(
-  ConfigProvider.fromUnknown({
-    OPENCODE_EXPERIMENTAL_FILEWATCHER: "true",
-    OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "false",
-  }),
-)
-
-/**
- * Boot an Instance with the given service layers and run `body` with
- * the ManagedRuntime. Cleanup is automatic — the runtime is disposed
- * and Instance context is torn down when `body` completes.
- *
- * Layers may depend on InstanceContext (provided automatically).
- * Pass extra layers via `options.provide` (e.g. ConfigProvider.layer).
- */
-export function withServices<S>(
-  directory: string,
-  layer: Layer.Layer<S, any, InstanceContext>,
-  body: (rt: ManagedRuntime.ManagedRuntime<S, never>) => Promise<void>,
-  options?: { provide?: Layer.Layer<never>[] },
-) {
-  return Instance.provide({
-    directory,
-    fn: async () => {
-      const ctx = Layer.sync(InstanceContext, () =>
-        InstanceContext.of({
-          directory: Instance.directory,
-          worktree: Instance.worktree,
-          project: Instance.project,
-        }),
-      )
-      let resolved: Layer.Layer<S> = layer.pipe(Layer.provide(ctx)) as any
-      if (options?.provide) {
-        for (const l of options.provide) {
-          resolved = resolved.pipe(Layer.provide(l)) as any
-        }
-      }
-      const rt = ManagedRuntime.make(resolved)
-      try {
-        await body(rt)
-      } finally {
-        await rt.dispose()
-      }
-    },
-  })
-}
--- a/packages/opencode/test/format/format.test.ts
+++ b/packages/opencode/test/format/format.test.ts
@@ -1,172 +1,182 @@
-import { Effect } from "effect"
-import { afterEach, describe, expect, test } from "bun:test"
-import { tmpdir } from "../fixture/fixture"
-import { withServices } from "../fixture/instance"
-import { Bus } from "../../src/bus"
-import { File } from "../../src/file"
+import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"
+import { describe, expect } from "bun:test"
+import { Effect, Layer } from "effect"
+import { provideTmpdirInstance } from "../fixture/fixture"
+import { testEffect } from "../lib/effect"
 import { Format } from "../../src/format"
 import * as Formatter from "../../src/format/formatter"
-import { Instance } from "../../src/project/instance"
+
+const node = NodeChildProcessSpawner.layer.pipe(
+  Layer.provideMerge(Layer.mergeAll(NodeFileSystem.layer, NodePath.layer)),
+)
+
+const it = testEffect(Layer.mergeAll(Format.layer, node))

 describe("Format", () => {
-  afterEach(async () => {
-    await Instance.disposeAll()
-  })
+  it.effect("status() returns built-in formatters when no config overrides", () =>
+    provideTmpdirInstance(() =>
+      Format.Service.use((fmt) =>
+        Effect.gen(function* () {
+          const statuses = yield* fmt.status()
+          expect(Array.isArray(statuses)).toBe(true)
+          expect(statuses.length).toBeGreaterThan(0)

-  test("status() returns built-in formatters when no config overrides", async () => {
-    await using tmp = await tmpdir()
+          for (const item of statuses) {
+            expect(typeof item.name).toBe("string")
+            expect(Array.isArray(item.extensions)).toBe(true)
+            expect(typeof item.enabled).toBe("boolean")
+          }

-    await withServices(tmp.path, Format.layer, async (rt) => {
-      const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
-      expect(Array.isArray(statuses)).toBe(true)
-      expect(statuses.length).toBeGreaterThan(0)
+          const gofmt = statuses.find((item) => item.name === "gofmt")
+          expect(gofmt).toBeDefined()
+          expect(gofmt!.extensions).toContain(".go")
+        }),
+      ),
+    ),
+  )

-      for (const s of statuses) {
-        expect(typeof s.name).toBe("string")
-        expect(Array.isArray(s.extensions)).toBe(true)
-        expect(typeof s.enabled).toBe("boolean")
-      }
+  it.effect("status() returns empty list when formatter is disabled", () =>
+    provideTmpdirInstance(
+      () =>
+        Format.Service.use((fmt) =>
+          Effect.gen(function* () {
+            expect(yield* fmt.status()).toEqual([])
+          }),
+        ),
+      { config: { formatter: false } },
+    ),
+  )

-      const gofmt = statuses.find((s) => s.name === "gofmt")
-      expect(gofmt).toBeDefined()
-      expect(gofmt!.extensions).toContain(".go")
-    })
-  })
-
-  test("status() returns empty list when formatter is disabled", async () => {
-    await using tmp = await tmpdir({
-      config: { formatter: false },
-    })
-
-    await withServices(tmp.path, Format.layer, async (rt) => {
-      const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
-      expect(statuses).toEqual([])
-    })
-  })
-
-  test("status() excludes formatters marked as disabled in config", async () => {
-    await using tmp = await tmpdir({
-      config: {
-        formatter: {
-          gofmt: { disabled: true },
+  it.effect("status() excludes formatters marked as disabled in config", () =>
+    provideTmpdirInstance(
+      () =>
+        Format.Service.use((fmt) =>
+          Effect.gen(function* () {
+            const statuses = yield* fmt.status()
+            const gofmt = statuses.find((item) => item.name === "gofmt")
+            expect(gofmt).toBeUndefined()
+          }),
+        ),
+      {
+        config: {
+          formatter: {
+            gofmt: { disabled: true },
+          },
        },
      },
-    })
+    ),
+  )

-    await withServices(tmp.path, Format.layer, async (rt) => {
-      const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
-      const gofmt = statuses.find((s) => s.name === "gofmt")
-      expect(gofmt).toBeUndefined()
-    })
-  })
+  it.effect("service initializes without error", () =>
+    provideTmpdirInstance(() => Format.Service.use(() => Effect.void)),
+  )

-  test("service initializes without error", async () => {
-    await using tmp = await tmpdir()
-
-    await withServices(tmp.path, Format.layer, async (rt) => {
-      await rt.runPromise(Format.Service.use(() => Effect.void))
-    })
-  })
-
-  test("status() initializes formatter state per directory", async () => {
-    await using off = await tmpdir({
-      config: { formatter: false },
-    })
-    await using on = await tmpdir()
-
-    const a = await Instance.provide({
-      directory: off.path,
-      fn: () => Format.status(),
-    })
-    const b = await Instance.provide({
-      directory: on.path,
-      fn: () => Format.status(),
-    })
-
-    expect(a).toEqual([])
-    expect(b.length).toBeGreaterThan(0)
-  })
-
-  test("runs enabled checks for matching formatters in parallel", async () => {
-    await using tmp = await tmpdir()
-
-    const file = `${tmp.path}/test.parallel`
-    await Bun.write(file, "x")
-
-    const one = {
-      extensions: Formatter.gofmt.extensions,
-      enabled: Formatter.gofmt.enabled,
-      command: Formatter.gofmt.command,
-    }
-    const two = {
-      extensions: Formatter.mix.extensions,
-      enabled: Formatter.mix.enabled,
-      command: Formatter.mix.command,
-    }
-
-    let active = 0
-    let max = 0
-
-    Formatter.gofmt.extensions = [".parallel"]
-    Formatter.mix.extensions = [".parallel"]
-    Formatter.gofmt.command = ["sh", "-c", "true"]
-    Formatter.mix.command = ["sh", "-c", "true"]
-    Formatter.gofmt.enabled = async () => {
-      active++
-      max = Math.max(max, active)
-      await Bun.sleep(20)
-      active--
-      return true
-    }
-    Formatter.mix.enabled = async () => {
-      active++
-      max = Math.max(max, active)
-      await Bun.sleep(20)
-      active--
-      return true
-    }
-
-    try {
-      await withServices(tmp.path, Format.layer, async (rt) => {
-        await rt.runPromise(Format.Service.use((s) => s.init()))
-        await Bus.publish(File.Event.Edited, { file })
+  it.effect("status() initializes formatter state per directory", () =>
+    Effect.gen(function* () {
+      const a = yield* provideTmpdirInstance(() => Format.Service.use((fmt) => fmt.status()), {
+        config: { formatter: false },
      })
-    } finally {
-      Formatter.gofmt.extensions = one.extensions
-      Formatter.gofmt.enabled = one.enabled
-      Formatter.gofmt.command = one.command
-      Formatter.mix.extensions = two.extensions
-      Formatter.mix.enabled = two.enabled
-      Formatter.mix.command = two.command
-    }
+      const b = yield* provideTmpdirInstance(() => Format.Service.use((fmt) => fmt.status()))

-    expect(max).toBe(2)
-  })
+      expect(a).toEqual([])
+      expect(b.length).toBeGreaterThan(0)
+    }),
+  )

-  test("runs matching formatters sequentially for the same file", async () => {
-    await using tmp = await tmpdir({
-      config: {
-        formatter: {
-          first: {
-            command: ["sh", "-c", 'sleep 0.05; v=$(cat "$1"); printf \'%sA\' "$v" > "$1"', "sh", "$FILE"],
-            extensions: [".seq"],
-          },
-          second: {
-            command: ["sh", "-c", 'v=$(cat "$1"); printf \'%sB\' "$v" > "$1"', "sh", "$FILE"],
-            extensions: [".seq"],
+  it.effect("runs enabled checks for matching formatters in parallel", () =>
+    provideTmpdirInstance((path) =>
+      Effect.gen(function* () {
+        const file = `${path}/test.parallel`
+        yield* Effect.promise(() => Bun.write(file, "x"))
+
+        const one = {
+          extensions: Formatter.gofmt.extensions,
+          enabled: Formatter.gofmt.enabled,
+          command: Formatter.gofmt.command,
+        }
+        const two = {
+          extensions: Formatter.mix.extensions,
+          enabled: Formatter.mix.enabled,
+          command: Formatter.mix.command,
+        }
+
+        let active = 0
+        let max = 0
+
+        yield* Effect.acquireUseRelease(
+          Effect.sync(() => {
+            Formatter.gofmt.extensions = [".parallel"]
+            Formatter.mix.extensions = [".parallel"]
+            Formatter.gofmt.command = ["sh", "-c", "true"]
+            Formatter.mix.command = ["sh", "-c", "true"]
+            Formatter.gofmt.enabled = async () => {
+              active++
+              max = Math.max(max, active)
+              await Bun.sleep(20)
+              active--
+              return true
+            }
+            Formatter.mix.enabled = async () => {
+              active++
+              max = Math.max(max, active)
+              await Bun.sleep(20)
+              active--
+              return true
+            }
+          }),
+          () =>
+            Format.Service.use((fmt) =>
+              Effect.gen(function* () {
+                yield* fmt.init()
+                yield* fmt.file(file)
+              }),
+            ),
+          () =>
+            Effect.sync(() => {
+              Formatter.gofmt.extensions = one.extensions
+              Formatter.gofmt.enabled = one.enabled
+              Formatter.gofmt.command = one.command
+              Formatter.mix.extensions = two.extensions
+              Formatter.mix.enabled = two.enabled
+              Formatter.mix.command = two.command
+            }),
+        )
+
+        expect(max).toBe(2)
+      }),
+    ),
+  )
+
+  it.effect("runs matching formatters sequentially for the same file", () =>
+    provideTmpdirInstance(
+      (path) =>
+        Effect.gen(function* () {
+          const file = `${path}/test.seq`
+          yield* Effect.promise(() => Bun.write(file, "x"))
+
+          yield* Format.Service.use((fmt) =>
+            Effect.gen(function* () {
+              yield* fmt.init()
+              yield* fmt.file(file)
+            }),
+          )
+
+          expect(yield* Effect.promise(() => Bun.file(file).text())).toBe("xAB")
+        }),
+      {
+        config: {
+          formatter: {
+            first: {
+              command: ["sh", "-c", 'sleep 0.05; v=$(cat "$1"); printf \'%sA\' "$v" > "$1"', "sh", "$FILE"],
+              extensions: [".seq"],
+            },
+            second: {
+              command: ["sh", "-c", 'v=$(cat "$1"); printf \'%sB\' "$v" > "$1"', "sh", "$FILE"],
+              extensions: [".seq"],
+            },
          },
        },
      },
-    })
-
-    const file = `${tmp.path}/test.seq`
-    await Bun.write(file, "x")
-
-    await withServices(tmp.path, Format.layer, async (rt) => {
-      await rt.runPromise(Format.Service.use((s) => s.init()))
-      await Bus.publish(File.Event.Edited, { file })
-    })
-
-    expect(await Bun.file(file).text()).toBe("xAB")
-  })
+    ),
+  )
 })
--- 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,28 +15,26 @@ const describeVcs = FileWatcher.hasNativeBinding() && !process.env.CI ? describe
 // Helpers
 // ---------------------------------------------------------------------------

-function withVcs(
-  directory: string,
-  body: (rt: ManagedRuntime.ManagedRuntime<FileWatcher.Service | Vcs.Service, never>) => Promise<void>,
-) {
-  return withServices(
+async function withVcs(directory: string, body: () => Promise<void>) {
+  return Instance.provide({
    directory,
-    Layer.merge(FileWatcher.layer, Vcs.defaultLayer),
-    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] },
-  )
+  })
 }

-function withVcsOnly(
-  directory: string,
-  body: (rt: ManagedRuntime.ManagedRuntime<Vcs.Service, never>) => Promise<void>,
-) {
-  return withServices(directory, Vcs.defaultLayer, body)
+function withVcsOnly(directory: string, body: () => Promise<void>) {
+  return Instance.provide({
+    directory,
+    fn: async () => {
+      Vcs.init()
+      await body()
+    },
+  })
 }

 type BranchEvent = { directory?: string; payload: { type: string; properties: { branch?: string } } }
@@ -82,8 +78,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")
    })
@@ -92,8 +88,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()
    })
  })
@@ -119,14 +115,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)
    })
  })
@@ -141,8 +137,8 @@ describe("Vcs diff", () => {
    await using tmp = await tmpdir({ git: true })
    await $`git branch -M main`.cwd(tmp.path).quiet()

-    await withVcsOnly(tmp.path, async (rt) => {
-      const branch = await rt.runPromise(Vcs.Service.use((s) => s.defaultBranch()))
+    await withVcsOnly(tmp.path, async () => {
+      const branch = await Vcs.defaultBranch()
      expect(branch).toBe("main")
    })
  })
@@ -152,8 +148,8 @@ describe("Vcs diff", () => {
    await $`git branch -M trunk`.cwd(tmp.path).quiet()
    await $`git config init.defaultBranch trunk`.cwd(tmp.path).quiet()

-    await withVcsOnly(tmp.path, async (rt) => {
-      const branch = await rt.runPromise(Vcs.Service.use((s) => s.defaultBranch()))
+    await withVcsOnly(tmp.path, async () => {
+      const branch = await Vcs.defaultBranch()
      expect(branch).toBe("trunk")
    })
  })
@@ -165,10 +161,10 @@ describe("Vcs diff", () => {
    const dir = path.join(wt.path, "feature")
    await $`git worktree add -b feature/test ${dir} HEAD`.cwd(tmp.path).quiet()

-    await withVcsOnly(dir, async (rt) => {
+    await withVcsOnly(dir, async () => {
      const [branch, base] = await Promise.all([
-        rt.runPromise(Vcs.Service.use((s) => s.branch())),
-        rt.runPromise(Vcs.Service.use((s) => s.defaultBranch())),
+        Vcs.branch(),
+        Vcs.defaultBranch(),
      ])
      expect(branch).toBe("feature/test")
      expect(base).toBe("main")
@@ -182,8 +178,8 @@ describe("Vcs diff", () => {
    await $`git commit --no-gpg-sign -m "add file"`.cwd(tmp.path).quiet()
    await fs.writeFile(path.join(tmp.path, "file.txt"), "changed\n", "utf-8")

-    await withVcsOnly(tmp.path, async (rt) => {
-      const diff = await rt.runPromise(Vcs.Service.use((s) => s.diff("git")))
+    await withVcsOnly(tmp.path, async () => {
+      const diff = await Vcs.diff("git")
      expect(diff).toEqual(
        expect.arrayContaining([
          expect.objectContaining({
@@ -199,8 +195,8 @@ describe("Vcs diff", () => {
    await using tmp = await tmpdir({ git: true })
    await fs.writeFile(path.join(tmp.path, weird), "hello\n", "utf-8")

-    await withVcsOnly(tmp.path, async (rt) => {
-      const diff = await rt.runPromise(Vcs.Service.use((s) => s.diff("git")))
+    await withVcsOnly(tmp.path, async () => {
+      const diff = await Vcs.diff("git")
      expect(diff).toEqual(
        expect.arrayContaining([
          expect.objectContaining({
@@ -220,8 +216,8 @@ describe("Vcs diff", () => {
    await $`git add .`.cwd(tmp.path).quiet()
    await $`git commit --no-gpg-sign -m "branch file"`.cwd(tmp.path).quiet()

-    await withVcsOnly(tmp.path, async (rt) => {
-      const diff = await rt.runPromise(Vcs.Service.use((s) => s.diff("branch")))
+    await withVcsOnly(tmp.path, async () => {
+      const diff = await Vcs.diff("branch")
      expect(diff).toEqual(
        expect.arrayContaining([
          expect.objectContaining({
--- a/packages/opencode/test/sync/index.test.ts
+++ b/packages/opencode/test/sync/index.test.ts
@@ -110,10 +110,16 @@ describe("SyncEvent", () => {
          type: string
          properties: { id: string; name: string }
        }> = []
-        const unsub = Bus.subscribeAll((event) => events.push(event))
+        const received = new Promise<void>((resolve) => {
+          Bus.subscribeAll((event) => {
+            events.push(event)
+            resolve()
+          })
+        })

        SyncEvent.run(Created, { id: "evt_1", name: "test" })

+        await received
        expect(events).toHaveLength(1)
        expect(events[0]).toEqual({
          type: "item.created",
@@ -122,8 +128,6 @@ describe("SyncEvent", () => {
            name: "test",
          },
        })
-
-        unsub()
      }),
    )
  })
--- a/packages/opencode/test/tool/edit.test.ts
+++ b/packages/opencode/test/tool/edit.test.ts
@@ -89,7 +89,6 @@ describe("tool.edit", () => {
          const { FileWatcher } = await import("../../src/file/watcher")

          const events: string[] = []
-          const unsubEdited = Bus.subscribe(File.Event.Edited, () => events.push("edited"))
          const unsubUpdated = Bus.subscribe(FileWatcher.Event.Updated, () => events.push("updated"))

          const edit = await EditTool.init()
@@ -102,9 +101,7 @@ describe("tool.edit", () => {
            ctx,
          )

-          expect(events).toContain("edited")
          expect(events).toContain("updated")
-          unsubEdited()
          unsubUpdated()
        },
      })
@@ -305,11 +302,9 @@ describe("tool.edit", () => {
          await FileTime.read(ctx.sessionID, filepath)

          const { Bus } = await import("../../src/bus")
-          const { File } = await import("../../src/file")
          const { FileWatcher } = await import("../../src/file/watcher")

          const events: string[] = []
-          const unsubEdited = Bus.subscribe(File.Event.Edited, () => events.push("edited"))
          const unsubUpdated = Bus.subscribe(FileWatcher.Event.Updated, () => events.push("updated"))

          const edit = await EditTool.init()
@@ -322,9 +317,7 @@ describe("tool.edit", () => {
            ctx,
          )

-          expect(events).toContain("edited")
          expect(events).toContain("updated")
-          unsubEdited()
          unsubUpdated()
        },
      })