diff --git a/packages/opencode/src/automation/index.ts b/packages/opencode/src/automation/index.ts
index 23370ed30..8440bb773 100644
--- a/packages/opencode/src/automation/index.ts
+++ b/packages/opencode/src/automation/index.ts
@@ -1,7 +1,6 @@
 import z from "zod"
 import { Context as EffectContext, Effect, Layer } from "effect"
-import { InstanceState } from "@/effect/instance-state"
-import { makeRuntime } from "@/effect/run-service"
+import { registerDisposer } from "@/effect/instance-registry"
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
 import { GlobalBus } from "@/bus/global"
@@ -274,10 +273,22 @@ export namespace Automation {
     activeWriters: Set<string>
     activeRuns: Map<string, { writerKey: string; controller: AbortController; runID: string }>
   }
-  // Per-directory execution state. The container lives in InstanceState (owned by the
-  // Service layer below); the sync facade reads it through a runtime bridge (see `state`).
-  function state(): State {
-    return automationRuntime.runSync((svc) => svc.activeState())
+  const activeStates = new Map<string, State>()
+  registerDisposer(async (directory) => {
+    activeStates.delete(directory)
+  })
+
+  function createState(): State {
+    return { activeWriters: new Set<string>(), activeRuns: new Map() }
+  }
+
+  function state(scope: Scope = currentScope()): State {
+    let activeState = activeStates.get(scope.ownerDirectory)
+    if (!activeState) {
+      activeState = createState()
+      activeStates.set(scope.ownerDirectory, activeState)
+    }
+    return activeState
   }
 
   export type RunExecutor = (input: {
@@ -823,7 +834,11 @@ export namespace Automation {
       try {
         await stopLiveRunForSourceDelete(definition.id)
         const removed = await remove(definition.id)
-        await Bus.publish(Event.DefinitionDeleted, removed.tombstone)
+        GlobalBus.emit("event", {
+          directory: Instance.directory,
+          project: Instance.project.id,
+          payload: { type: Event.DefinitionDeleted.type, properties: removed.tombstone },
+        })
       } catch (error) {
         if (NotFoundError.isInstance(error)) continue
         throw error
@@ -1278,7 +1293,9 @@ export namespace Automation {
     return { items, nextCursor: page.length > limit ? items.at(-1)?.id ?? null : null }
   }
 
-  export const publishDefinitionUpdated = (definition: Definition) => Bus.publish(Event.DefinitionUpdated, definition)
+  export async function publishDefinitionUpdated(definition: Definition) {
+    publishDefinitionUpdatedForScope(definition, currentScope())
+  }
   export const publishDefinitionUpdatedForScope = (definition: Definition, scope: Scope) => {
     GlobalBus.emit("event", {
       directory: scope.ownerDirectory,
@@ -1286,7 +1303,9 @@ export namespace Automation {
       payload: { type: Event.DefinitionUpdated.type, properties: definition },
     })
   }
-  export const publishRunUpdated = (run: Run) => Bus.publish(Event.RunUpdated, run)
+  export async function publishRunUpdated(run: Run) {
+    publishRunUpdatedForScope(run, currentScope())
+  }
   export const publishRunUpdatedForScope = (run: Run, scope: Scope) => {
     GlobalBus.emit("event", {
       directory: scope.ownerDirectory,
@@ -1327,12 +1346,10 @@ export namespace Automation {
 
   export class Service extends EffectContext.Service<Service, Interface>()("@opencode/Automation") {}
 
-  export const layer: Layer.Layer<Service> = Layer.effect(
+  export const layer: Layer.Layer<Service, never, Bus.Service> = Layer.effect(
     Service,
     Effect.gen(function* () {
-      const activeStateHandle = yield* InstanceState.make<State>(() =>
-        Effect.sync(() => ({ activeWriters: new Set<string>(), activeRuns: new Map() })),
-      )
+      const bus = yield* Bus.Service
       return Service.of({
         list: () => Effect.sync(() => list()),
         get: (id) => Effect.sync(() => get(id)),
@@ -1358,18 +1375,16 @@ export namespace Automation {
           ),
         runNowExecuting: (id, options) => Effect.promise(() => runNowExecuting(id, options)),
         runs: (input) => Effect.sync(() => runs(input)),
-        publishDefinitionUpdated: (definition) => Effect.promise(() => publishDefinitionUpdated(definition)),
+        publishDefinitionUpdated: (definition) => bus.publish(Event.DefinitionUpdated, definition),
         publishDefinitionUpdatedForScope: (definition, scope) => Effect.sync(() => publishDefinitionUpdatedForScope(definition, scope)),
-        publishDefinitionDeleted: (tombstone) => Effect.promise(() => Bus.publish(Event.DefinitionDeleted, tombstone)),
-        publishRunUpdated: (run) => Effect.promise(() => publishRunUpdated(run)),
-        activeState: () => InstanceState.get(activeStateHandle),
+        publishDefinitionDeleted: (tombstone) => bus.publish(Event.DefinitionDeleted, tombstone),
+        publishRunUpdated: (run) => bus.publish(Event.RunUpdated, run),
+        activeState: () => Effect.sync(() => state()),
       })
     }),
   )
 
-  export const defaultLayer = layer
-
-  const automationRuntime = makeRuntime(Service, layer)
+  export const defaultLayer = layer.pipe(Layer.provide(Bus.defaultLayer))
 }
 
 export class ValidationError extends Error {
diff --git a/packages/opencode/src/bus/index.ts b/packages/opencode/src/bus/index.ts
index 5969b15fc..f68ac4cf1 100644
--- a/packages/opencode/src/bus/index.ts
+++ b/packages/opencode/src/bus/index.ts
@@ -6,7 +6,6 @@ import { BusEvent } from "./bus-event"
 import { GlobalBus } from "./global"
 import { WorkspaceContext } from "@/control-plane/workspace-context"
 import { InstanceState } from "@/effect/instance-state"
-import { makeRuntime } from "@/effect/run-service"
 import { LocalContext } from "@/util/local-context"
 
 export namespace Bus {
@@ -23,6 +22,12 @@ export namespace Bus {
     type: D["type"]
     properties: z.infer<D["properties"]>
   }
+  type GlobalEvent = {
+    directory?: string
+    project?: string
+    workspace?: string
+    payload: Payload
+  }
 
   type State = {
     wildcard: PubSub.PubSub<Payload>
@@ -83,25 +88,35 @@ export namespace Bus {
       }
 
       function publish<D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) {
+        const payload: Payload = { type: def.type, properties }
         return Effect.gen(function* () {
-          const s = yield* InstanceState.get(state)
-          const payload: Payload = { type: def.type, properties }
-          log.info("publishing", { type: def.type })
-
-          const ps = s.typed.get(def.type)
-          if (ps) yield* PubSub.publish(ps, payload)
-          yield* PubSub.publish(s.wildcard, payload)
-
-          const dir = yield* InstanceState.directory
-          const context = yield* InstanceState.context
-          const workspace = yield* InstanceState.workspaceID
-
-          GlobalBus.emit("event", {
-            directory: dir,
-            project: context.project.id,
-            workspace,
-            payload,
-          })
+          let event: GlobalEvent
+          try {
+            const s = yield* InstanceState.get(state)
+            log.info("publishing", { type: def.type })
+
+            const ps = s.typed.get(def.type)
+            if (ps) yield* PubSub.publish(ps, payload)
+            yield* PubSub.publish(s.wildcard, payload)
+
+            const dir = yield* InstanceState.directory
+            const context = yield* InstanceState.context
+            const workspace = yield* InstanceState.workspaceID
+
+            event = {
+              directory: dir,
+              project: context.project.id,
+              workspace,
+              payload,
+            }
+          } catch (error) {
+            if (!(error instanceof LocalContext.NotFound) || error.name !== "instance") throw error
+            event = {
+              directory: "global",
+              payload,
+            }
+          }
+          GlobalBus.emit("event", event)
         })
       }
 
@@ -172,31 +187,4 @@ export namespace Bus {
   )
 
   export const defaultLayer = layer
-
-  const { runPromise, runSync } = makeRuntime(Service, layer)
-
-  // runSync is safe here because the subscribe chain (InstanceState.get, PubSub.subscribe,
-  // Scope.make, Effect.forkScoped) is entirely synchronous. If any step becomes async, this will throw.
-  export async function publish<D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) {
-    try {
-      return await runPromise((svc) => svc.publish(def, properties))
-    } catch (error) {
-      if (!(error instanceof LocalContext.NotFound) || error.name !== "instance") throw error
-      GlobalBus.emit("event", {
-        directory: "global",
-        payload: { type: def.type, properties },
-      })
-    }
-  }
-
-  export function subscribe<D extends BusEvent.Definition>(
-    def: D,
-    callback: (event: { type: D["type"]; properties: z.infer<D["properties"]> }) => unknown,
-  ) {
-    return runSync((svc) => svc.subscribeCallback(def, callback))
-  }
-
-  export function subscribeAll(callback: (event: any) => unknown) {
-    return runSync((svc) => svc.subscribeAllCallback(callback))
-  }
 }
diff --git a/packages/opencode/src/cli/cmd/github.ts b/packages/opencode/src/cli/cmd/github.ts
index 111d31665..f0866cd27 100644
--- a/packages/opencode/src/cli/cmd/github.ts
+++ b/packages/opencode/src/cli/cmd/github.ts
@@ -917,33 +917,37 @@ export const GithubRunCommand = cmd({
         }
 
         let text = ""
-        Bus.subscribe(MessageV2.Event.PartUpdated, (evt) => {
-          if (evt.properties.part.sessionID !== session.id) return
-          //if (evt.properties.part.messageID === messageID) return
-          const part = evt.properties.part
-
-          if (part.type === "tool" && part.state.status === "completed") {
-            const [tool, color] = TOOL[part.tool] ?? [part.tool, UI.Style.TEXT_INFO_BOLD]
-            const title =
-              part.state.title || Object.keys(part.state.input).length > 0
-                ? JSON.stringify(part.state.input)
-                : "Unknown"
-            console.log()
-            printEvent(color, tool, title)
-          }
+        AppRuntime.runSync(
+          Bus.Service.use((bus) =>
+            bus.subscribeCallback(MessageV2.Event.PartUpdated, (evt) => {
+              if (evt.properties.part.sessionID !== session.id) return
+              //if (evt.properties.part.messageID === messageID) return
+              const part = evt.properties.part
+
+              if (part.type === "tool" && part.state.status === "completed") {
+                const [tool, color] = TOOL[part.tool] ?? [part.tool, UI.Style.TEXT_INFO_BOLD]
+                const title =
+                  part.state.title || Object.keys(part.state.input).length > 0
+                    ? JSON.stringify(part.state.input)
+                    : "Unknown"
+                console.log()
+                printEvent(color, tool, title)
+              }
 
-          if (part.type === "text") {
-            text = part.text
+              if (part.type === "text") {
+                text = part.text
 
-            if (part.time?.end) {
-              UI.empty()
-              UI.println(UI.markdown(text))
-              UI.empty()
-              text = ""
-              return
-            }
-          }
-        })
+                if (part.time?.end) {
+                  UI.empty()
+                  UI.println(UI.markdown(text))
+                  UI.empty()
+                  text = ""
+                  return
+                }
+              }
+            }),
+          ),
+        )
       }
 
       async function summarize(response: string) {
diff --git a/packages/opencode/src/cli/cmd/mcp.ts b/packages/opencode/src/cli/cmd/mcp.ts
index 7b2fb7e4b..a61ec1ec8 100644
--- a/packages/opencode/src/cli/cmd/mcp.ts
+++ b/packages/opencode/src/cli/cmd/mcp.ts
@@ -264,14 +264,18 @@ export const McpAuthCommand = cmd({
         spinner.start("Starting OAuth flow...")
 
         // Subscribe to browser open failure events to show URL for manual opening
-        const unsubscribe = Bus.subscribe(MCP.BrowserOpenFailed, (evt) => {
-          if (evt.properties.mcpName === serverName) {
-            spinner.stop("Could not open browser automatically")
-            prompts.log.warn("Please open this URL in your browser to authenticate:")
-            prompts.log.info(evt.properties.url)
-            spinner.start("Waiting for authorization...")
-          }
-        })
+        const unsubscribe = AppRuntime.runSync(
+          Bus.Service.use((bus) =>
+            bus.subscribeCallback(MCP.BrowserOpenFailed, (evt) => {
+              if (evt.properties.mcpName === serverName) {
+                spinner.stop("Could not open browser automatically")
+                prompts.log.warn("Please open this URL in your browser to authenticate:")
+                prompts.log.info(evt.properties.url)
+                spinner.start("Waiting for authorization...")
+              }
+            }),
+          ),
+        )
 
         try {
           const status = await AppRuntime.runPromise(MCP.Service.use((mcp) => mcp.authenticate(serverName)))
diff --git a/packages/opencode/src/cli/upgrade.ts b/packages/opencode/src/cli/upgrade.ts
index 0c7d8b2da..1f5af2909 100644
--- a/packages/opencode/src/cli/upgrade.ts
+++ b/packages/opencode/src/cli/upgrade.ts
@@ -18,8 +18,13 @@ export async function upgrade() {
   )
   if (!latest) return
 
+  const publishInstallationEvent = (
+    def: typeof Installation.Event.UpdateAvailable | typeof Installation.Event.Updated,
+    properties: { version: string },
+  ) => AppRuntime.runPromise(Bus.Service.use((bus) => bus.publish(def, properties)))
+
   if (Flag.OPENCODE_ALWAYS_NOTIFY_UPDATE) {
-    await Bus.publish(Installation.Event.UpdateAvailable, { version: latest })
+    await publishInstallationEvent(Installation.Event.UpdateAvailable, { version: latest })
     return
   }
 
@@ -29,12 +34,12 @@ export async function upgrade() {
   const kind = Installation.getReleaseType(Installation.VERSION, latest)
 
   if (config.autoupdate === "notify" || kind !== "patch") {
-    await Bus.publish(Installation.Event.UpdateAvailable, { version: latest })
+    await publishInstallationEvent(Installation.Event.UpdateAvailable, { version: latest })
     return
   }
 
   if (method === "unknown") return
   await AppRuntime.runPromise(Installation.Service.use((svc) => svc.upgrade(method, latest)))
-    .then(() => Bus.publish(Installation.Event.Updated, { version: latest }))
+    .then(() => publishInstallationEvent(Installation.Event.Updated, { version: latest }))
     .catch(() => {})
 }
diff --git a/packages/opencode/src/config/agent.ts b/packages/opencode/src/config/agent.ts
index 3d7f9305c..15f84d6a9 100644
--- a/packages/opencode/src/config/agent.ts
+++ b/packages/opencode/src/config/agent.ts
@@ -3,7 +3,7 @@ export * as ConfigAgent from "./agent"
 import path from "path"
 import { Schema } from "effect"
 import z from "zod"
-import { Bus } from "@/bus"
+import { GlobalBus } from "@/bus/global"
 import { zod } from "@/util/effect-zod"
 import { Log } from "../util"
 import { NamedError } from "@opencode-ai/util/error"
@@ -120,7 +120,10 @@ export type Info = z.infer<typeof Info>
 
 async function reportLoadError(error: { toObject(): any }, context: Record<string, unknown>) {
   const { Session } = await import("@/session")
-  void Bus.publish(Session.Event.Error, { error: error.toObject() })
+  GlobalBus.emit("event", {
+    directory: "global",
+    payload: { type: Session.Event.Error.type, properties: { error: error.toObject() } },
+  })
   log.error("failed to load agent", context)
 }
 
@@ -137,7 +140,10 @@ export async function load(dir: string) {
         ? err.data.message
         : `Failed to parse agent ${item}`
       const { Session } = await import("@/session")
-      void Bus.publish(Session.Event.Error, { error: new NamedError.Unknown({ message }).toObject() })
+      GlobalBus.emit("event", {
+        directory: "global",
+        payload: { type: Session.Event.Error.type, properties: { error: new NamedError.Unknown({ message }).toObject() } },
+      })
       log.error("failed to load agent", { agent: item, err })
       return undefined
     })
@@ -174,7 +180,10 @@ export async function loadMode(dir: string) {
         ? err.data.message
         : `Failed to parse mode ${item}`
       const { Session } = await import("@/session")
-      void Bus.publish(Session.Event.Error, { error: new NamedError.Unknown({ message }).toObject() })
+      GlobalBus.emit("event", {
+        directory: "global",
+        payload: { type: Session.Event.Error.type, properties: { error: new NamedError.Unknown({ message }).toObject() } },
+      })
       log.error("failed to load mode", { mode: item, err })
       return undefined
     })
diff --git a/packages/opencode/src/config/command.ts b/packages/opencode/src/config/command.ts
index aef5306dd..4039a32cb 100644
--- a/packages/opencode/src/config/command.ts
+++ b/packages/opencode/src/config/command.ts
@@ -5,7 +5,7 @@ import { Log } from "../util"
 import { Schema } from "effect"
 import { NamedError } from "@opencode-ai/util/error"
 import { Glob } from "@opencode-ai/core/util/glob"
-import { Bus } from "@/bus"
+import { GlobalBus } from "@/bus/global"
 import { zod } from "@/util/effect-zod"
 import { withStatics } from "@/util/schema"
 import { configEntryNameFromPath } from "./entry-name"
@@ -15,6 +15,14 @@ import { ConfigModelID } from "./model-id"
 
 const log = Log.create({ service: "config" })
 
+async function publishSessionError(error: { toObject(): any }) {
+  const { Session } = await import("@/session")
+  GlobalBus.emit("event", {
+    directory: "global",
+    payload: { type: Session.Event.Error.type, properties: { error: error.toObject() } },
+  })
+}
+
 function commandSourceRank(filePath: string) {
   const normalized = filePath.replaceAll("\\", "/")
   if (normalized.includes("/.opencode/command/") || normalized.includes("/command/")) return 0
@@ -22,8 +30,9 @@ function commandSourceRank(filePath: string) {
 }
 
 async function reportLoadError(error: { toObject(): any }, item: string, cause: unknown) {
-  const { Session } = await import("@/session")
-  void Bus.publish(Session.Event.Error, { error: error.toObject() })
+  void publishSessionError(error).catch((publishError) => {
+    log.error("failed to publish session error event", { command: item, err: publishError })
+  })
   log.error("failed to load command", { command: item, err: cause })
 }
 
diff --git a/packages/opencode/src/file/watcher.ts b/packages/opencode/src/file/watcher.ts
index 2c97f4656..614fe42b0 100644
--- a/packages/opencode/src/file/watcher.ts
+++ b/packages/opencode/src/file/watcher.ts
@@ -547,8 +547,11 @@ export namespace FileWatcher {
   export const layer = Layer.effect(
     Service,
     Effect.gen(function* () {
+      const bus = yield* Bus.Service
       const config = yield* Config.Service
       const git = yield* Git.Service
+      const publish = <D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) =>
+        Effect.runPromise(bus.publish(def, properties))
 
       const state = yield* InstanceState.make(
         Effect.fn("FileWatcher.state")(
@@ -581,7 +584,7 @@ export namespace FileWatcher {
                   subscription_dir: request.subscriptionDirectory ?? request.directory,
                   watch_scope: request.watchScope,
                 })
-                Bus.publish(Event.Rescan, { directory: request.directory }).catch((error) =>
+                publish(Event.Rescan, { directory: request.directory }).catch((error) =>
                   log.warn("failed to publish watcher rescan", { dir: request.directory, error }),
                 )
               },
@@ -618,11 +621,15 @@ export namespace FileWatcher {
                   log.error("watcher callback error", { err })
                   return
                 }
+                const publishUpdate = (event: z.output<typeof Event.Updated.properties>) =>
+                  publish(Event.Updated, event).catch((error) =>
+                    log.warn("failed to publish watcher update", { event, error }),
+                  )
                 for (const evt of evts) {
                   if (!shouldPublish(evt.path)) continue
-                  if (evt.type === "create") Bus.publish(Event.Updated, { file: evt.path, event: "add" })
-                  if (evt.type === "update") Bus.publish(Event.Updated, { file: evt.path, event: "change" })
-                  if (evt.type === "delete") Bus.publish(Event.Updated, { file: evt.path, event: "unlink" })
+                  if (evt.type === "create") void publishUpdate({ file: evt.path, event: "add" })
+                  if (evt.type === "update") void publishUpdate({ file: evt.path, event: "change" })
+                  if (evt.type === "delete") void publishUpdate({ file: evt.path, event: "unlink" })
                 }
               })
 
@@ -673,7 +680,7 @@ export namespace FileWatcher {
                 let watchPlanDisposed = false
                 const fallbackRescan = createFallbackRescanThrottle()
                 const publishWorkspaceRescan = () =>
-                  Bus.publish(Event.Rescan, { directory: ctx.directory }).catch((error) =>
+                  publish(Event.Rescan, { directory: ctx.directory }).catch((error) =>
                     log.warn("failed to publish watcher rescan", { dir: ctx.directory, error }),
                   )
                 const applyPlan = Effect.fn("FileWatcher.applyWorkspaceWatchPlan")(function* (
@@ -772,10 +779,12 @@ export namespace FileWatcher {
                           isDisposed: () => watchPlanDisposed,
                           applyPlan: (planSnapshot) => Effect.runPromise(applyPlan(planSnapshot)),
                           publishUpdate: (event) => {
-                            Bus.publish(Event.Updated, event)
+                            void publish(Event.Updated, event).catch((error) =>
+                              log.warn("failed to publish watcher update", { event, error }),
+                            )
                           },
                           publishRescan: (directory) =>
-                            Bus.publish(Event.Rescan, { directory }).catch((error) =>
+                            publish(Event.Rescan, { directory }).catch((error) =>
                               log.warn("failed to publish watcher rescan", { dir: directory, error }),
                             ),
                         }),
@@ -856,5 +865,9 @@ export namespace FileWatcher {
     }),
   )
 
-  export const defaultLayer = layer.pipe(Layer.provide(Config.defaultLayer), Layer.provide(Git.defaultLayer))
+  export const defaultLayer = layer.pipe(
+    Layer.provide(Bus.defaultLayer),
+    Layer.provide(Config.defaultLayer),
+    Layer.provide(Git.defaultLayer),
+  )
 }
diff --git a/packages/opencode/src/lsp/client.ts b/packages/opencode/src/lsp/client.ts
index f66cfb77f..c59d29858 100644
--- a/packages/opencode/src/lsp/client.ts
+++ b/packages/opencode/src/lsp/client.ts
@@ -1,5 +1,6 @@
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
+import { Effect } from "effect"
 import path from "path"
 import { pathToFileURL, fileURLToPath } from "url"
 import { createMessageConnection, StreamMessageReader, StreamMessageWriter } from "vscode-jsonrpc/node"
@@ -40,13 +41,18 @@ export namespace LSPClient {
     ),
   }
 
-  export async function create(input: { serverID: string; server: LSPServer.Handle; root: string }) {
-    const l = log.clone().tag("serverID", input.serverID)
+  export async function create(options: {
+    bus: Pick<Bus.Interface, "publish" | "subscribeCallback">
+    serverID: string
+    server: LSPServer.Handle
+    root: string
+  }) {
+    const l = log.clone().tag("serverID", options.serverID)
     l.info("starting client")
 
     const connection = createMessageConnection(
-      new StreamMessageReader(input.server.process.stdout as any),
-      new StreamMessageWriter(input.server.process.stdin as any),
+      new StreamMessageReader(options.server.process.stdout as any),
+      new StreamMessageWriter(options.server.process.stdin as any),
     )
 
     const diagnostics = new Map<string, Diagnostic[]>()
@@ -58,8 +64,8 @@ export namespace LSPClient {
       })
       const exists = diagnostics.has(filePath)
       diagnostics.set(filePath, params.diagnostics)
-      if (!exists && input.serverID === "typescript") return
-      Bus.publish(Event.Diagnostics, { path: filePath, serverID: input.serverID })
+      if (!exists && options.serverID === "typescript") return
+      void Effect.runPromise(options.bus.publish(Event.Diagnostics, { path: filePath, serverID: options.serverID }))
     })
     connection.onRequest("window/workDoneProgress/create", (params) => {
       l.info("window/workDoneProgress/create", params)
@@ -67,14 +73,14 @@ export namespace LSPClient {
     })
     connection.onRequest("workspace/configuration", async () => {
       // Return server initialization options
-      return [input.server.initialization ?? {}]
+      return [options.server.initialization ?? {}]
     })
     connection.onRequest("client/registerCapability", async () => {})
     connection.onRequest("client/unregisterCapability", async () => {})
     connection.onRequest("workspace/workspaceFolders", async () => [
       {
         name: "workspace",
-        uri: pathToFileURL(input.root).href,
+        uri: pathToFileURL(options.root).href,
       },
     ])
     connection.listen()
@@ -82,16 +88,16 @@ export namespace LSPClient {
     l.info("sending initialize")
     await withTimeout(
       connection.sendRequest("initialize", {
-        rootUri: pathToFileURL(input.root).href,
-        processId: input.server.process.pid,
+        rootUri: pathToFileURL(options.root).href,
+        processId: options.server.process.pid,
         workspaceFolders: [
           {
             name: "workspace",
-            uri: pathToFileURL(input.root).href,
+            uri: pathToFileURL(options.root).href,
           },
         ],
         initializationOptions: {
-          ...input.server.initialization,
+          ...options.server.initialization,
         },
         capabilities: {
           window: {
@@ -118,7 +124,7 @@ export namespace LSPClient {
     ).catch((err) => {
       l.error("initialize error", { error: err })
       throw new InitializeError(
-        { serverID: input.serverID },
+        { serverID: options.serverID },
         {
           cause: err,
         },
@@ -127,9 +133,9 @@ export namespace LSPClient {
 
     await connection.sendNotification("initialized", {})
 
-    if (input.server.initialization) {
+    if (options.server.initialization) {
       await connection.sendNotification("workspace/didChangeConfiguration", {
-        settings: input.server.initialization,
+        settings: options.server.initialization,
       })
     }
 
@@ -138,9 +144,9 @@ export namespace LSPClient {
     } = {}
 
     const result = {
-      root: input.root,
+      root: options.root,
       get serverID() {
-        return input.serverID
+        return options.serverID
       },
       get connection() {
         return connection
@@ -216,17 +222,23 @@ export namespace LSPClient {
         let debounceTimer: ReturnType<typeof setTimeout> | undefined
         return await withTimeout(
           new Promise<void>((resolve) => {
-            unsub = Bus.subscribe(Event.Diagnostics, (event) => {
-              if (event.properties.path === normalizedPath && event.properties.serverID === result.serverID) {
-                // Debounce to allow LSP to send follow-up diagnostics (e.g., semantic after syntax)
-                if (debounceTimer) clearTimeout(debounceTimer)
-                debounceTimer = setTimeout(() => {
-                  log.info("got diagnostics", { path: normalizedPath })
-                  unsub?.()
-                  resolve()
-                }, DIAGNOSTICS_DEBOUNCE_MS)
-              }
-            })
+            try {
+              unsub = Effect.runSync(
+                options.bus.subscribeCallback(Event.Diagnostics, (event) => {
+                  if (event.properties.path === normalizedPath && event.properties.serverID === result.serverID) {
+                    // Debounce to allow LSP to send follow-up diagnostics (e.g., semantic after syntax)
+                    if (debounceTimer) clearTimeout(debounceTimer)
+                    debounceTimer = setTimeout(() => {
+                      log.info("got diagnostics", { path: normalizedPath })
+                      unsub?.()
+                      resolve()
+                    }, DIAGNOSTICS_DEBOUNCE_MS)
+                  }
+                }),
+              )
+            } catch {
+              resolve()
+            }
           }),
           3000,
         )
@@ -240,7 +252,7 @@ export namespace LSPClient {
         l.info("shutting down")
         connection.end()
         connection.dispose()
-        await Process.stop(input.server.process)
+        await Process.stop(options.server.process)
         l.info("shutdown")
       },
     }
diff --git a/packages/opencode/src/lsp/index.ts b/packages/opencode/src/lsp/index.ts
index 634e5e819..ca61419d2 100644
--- a/packages/opencode/src/lsp/index.ts
+++ b/packages/opencode/src/lsp/index.ts
@@ -174,6 +174,7 @@ export namespace LSP {
   export const layer = Layer.effect(
     Service,
     Effect.gen(function* () {
+      const bus = yield* Bus.Service
       const settings = yield* Settings.Service
 
       const state = yield* InstanceState.make<State>(
@@ -251,6 +252,7 @@ export namespace LSP {
             log.info("spawned lsp server", { serverID: server.id, root })
 
             const client = await LSPClient.create({
+              bus,
               serverID: server.id,
               server: handle,
               root,
@@ -313,7 +315,7 @@ export namespace LSP {
             if (!client) continue
 
             result.push(client)
-            Bus.publish(Event.Updated, {})
+            await Effect.runPromise(bus.publish(Event.Updated, {}))
           }
 
           return result
@@ -512,7 +514,7 @@ export namespace LSP {
         // killed clients; without this the status popover keeps showing
         // connected servers until another lsp event fires.
         if (hadClients) {
-          yield* Effect.promise(() => Bus.publish(Event.Updated, {}).catch(() => {}))
+          yield* bus.publish(Event.Updated, {}).pipe(Effect.ignore)
         }
       })
 
@@ -541,7 +543,7 @@ export namespace LSP {
     }),
   )
 
-  export const defaultLayer = layer.pipe(Layer.provide(Settings.defaultLayer))
+  export const defaultLayer = layer.pipe(Layer.provide(Bus.defaultLayer), Layer.provide(Settings.defaultLayer))
 
   export namespace Diagnostic {
     const MAX_PER_FILE = 20
diff --git a/packages/opencode/src/lsp/server.ts b/packages/opencode/src/lsp/server.ts
index 34185dffc..39108a326 100644
--- a/packages/opencode/src/lsp/server.ts
+++ b/packages/opencode/src/lsp/server.ts
@@ -2,6 +2,7 @@ import type { ChildProcessWithoutNullStreams } from "child_process"
 import path from "path"
 import os from "os"
 import { Global } from "../global"
+import { GlobalBus } from "@/bus/global"
 import { Log } from "@opencode-ai/core/util/log"
 import { text } from "node:stream/consumers"
 import { Instance } from "../project/instance"
@@ -235,11 +236,16 @@ export namespace LSPServer {
       if (err instanceof Npm.InstallFailedError) {
         try {
           const { LSP } = await import("./index")
-          const { Bus } = await import("../bus")
-          await Bus.publish(LSP.Event.InstallFailed, {
-            add: Array.from(err.add ?? [pkg]),
-            dir: err.dir,
-            error: err.cause instanceof Error ? err.cause.message : String(err.cause),
+          GlobalBus.emit("event", {
+            directory: "global",
+            payload: {
+              type: LSP.Event.InstallFailed.type,
+              properties: {
+                add: Array.from(err.add ?? [pkg]),
+                dir: err.dir,
+                error: err.cause instanceof Error ? err.cause.message : String(err.cause),
+              },
+            },
           })
         } catch (publishErr) {
           log.warn("failed to emit lsp install failed event", { error: publishErr })
diff --git a/packages/opencode/src/server/instance/event.ts b/packages/opencode/src/server/instance/event.ts
index a7a8d7e31..d7aed32cb 100644
--- a/packages/opencode/src/server/instance/event.ts
+++ b/packages/opencode/src/server/instance/event.ts
@@ -1,5 +1,6 @@
 import { Log } from "@opencode-ai/core/util/log"
 import { Bus } from "@/bus"
+import { AppRuntime } from "@/effect/app-runtime"
 import { AsyncQueue } from "../../util/queue"
 import { createSseResponse } from "../sse"
 
@@ -45,12 +46,16 @@ export function handleInstanceEventStream(request: Request, options: { heartbeat
         log.info("event disconnected")
       }
 
-      const unsub = Bus.subscribeAll((event) => {
-        q.push(JSON.stringify(event))
-        if (event.type === Bus.InstanceDisposed.type) {
-          stop()
-        }
-      })
+      const unsub = AppRuntime.runSync(
+        Bus.Service.use((bus) =>
+          bus.subscribeAllCallback((event) => {
+            q.push(JSON.stringify(event))
+            if (event.type === Bus.InstanceDisposed.type) {
+              stop()
+            }
+          }),
+        ),
+      )
 
       void (async () => {
         try {
diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts
index 907ef4d48..73261d90e 100644
--- a/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts
+++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts
@@ -146,10 +146,13 @@ function noContentAfter(effect: Effect.Effect<unknown, unknown, unknown>) {
 
 function publishPromptAsyncError(sessionID: SessionID, error: unknown) {
   const message = error instanceof Error ? error.message : String(error)
-  log.error("prompt_async failed", { sessionID, error })
-  void Bus.publish(SessionNs.Event.Error, {
-    sessionID,
-    error: new NamedError.Unknown({ message }).toObject(),
+  return Effect.gen(function* () {
+    const bus = yield* Bus.Service
+    log.error("prompt_async failed", { sessionID, error })
+    yield* bus.publish(SessionNs.Event.Error, {
+      sessionID,
+      error: new NamedError.Unknown({ message }).toObject(),
+    })
   })
 }
 
@@ -350,8 +353,8 @@ export const sessionHandlers = HttpApiBuilder.group(SessionApi, "session", (hand
         if (HttpServerResponse.isHttpServerResponse(body)) return body
         const sessionID = ctx.params.sessionID
         yield* SessionRouteEffects.promptSession({ ...body, sessionID }).pipe(
-          Effect.catch((error) => Effect.sync(() => publishPromptAsyncError(sessionID, error))),
-          Effect.catchDefect((error) => Effect.sync(() => publishPromptAsyncError(sessionID, error))),
+          Effect.catch((error) => publishPromptAsyncError(sessionID, error)),
+          Effect.catchDefect((error) => publishPromptAsyncError(sessionID, error)),
           Effect.forkDetach({ startImmediately: true }),
         )
         return HttpServerResponse.empty()
diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts
index f319cc822..5cea4f00f 100644
--- a/packages/opencode/src/session/llm.ts
+++ b/packages/opencode/src/session/llm.ts
@@ -16,7 +16,6 @@ import { Flag } from "@opencode-ai/core/flag/flag"
 import { Permission } from "@/permission"
 import { PermissionID } from "@/permission/schema"
 import { TOOL_INFO_ID, buildDeferredHint } from "../tool/tool-info"
-import { Bus } from "@/bus"
 import { Wildcard } from "@/util/wildcard"
 import { SessionID } from "@/session/schema"
 import { Auth } from "@/auth"
@@ -74,11 +73,7 @@ export interface Interface {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/LLM") {}
 
-const live: Layer.Layer<
-  Service,
-  never,
-  Auth.Service | Config.Service | Provider.Service | Plugin.Service | Permission.Service
-> = Layer.effect(
+const live: Layer.Layer<Service, never, Auth.Service | Config.Service | Provider.Service | Plugin.Service | Permission.Service> = Layer.effect(
   Service,
   Effect.gen(function* () {
     const auth = yield* Auth.Service
@@ -297,11 +292,7 @@ const live: Layer.Layer<
           }
 
           const id = PermissionID.ascending()
-          let unsub: (() => void) | undefined
           try {
-            unsub = Bus.subscribe(Permission.Event.Replied, (evt) => {
-              if (evt.properties.requestID === id) void evt.properties.reply
-            })
             const toolPatterns = approvalTools.map((t: { name: string; args: string }) => {
               try {
                 const parsed = JSON.parse(t.args) as Record<string, unknown>
@@ -328,8 +319,6 @@ const live: Layer.Layer<
             return { approved: true }
           } catch {
             return { approved: false }
-          } finally {
-            unsub?.()
           }
         })
       }
diff --git a/packages/opencode/src/sync/index.ts b/packages/opencode/src/sync/index.ts
index 96b678104..980957bdb 100644
--- a/packages/opencode/src/sync/index.ts
+++ b/packages/opencode/src/sync/index.ts
@@ -1,10 +1,12 @@
 import z from "zod"
 import type { ZodObject } from "zod"
 import { EventEmitter } from "events"
-import { Context, Effect, Layer, Schema } from "effect"
+import { Context, Effect, Fiber, Layer, Schema } from "effect"
 import { Database, eq } from "@/storage/db"
 import { Bus as ProjectBus } from "@/bus"
+import { GlobalBus } from "@/bus/global"
 import { BusEvent } from "@/bus/bus-event"
+import { Instance } from "@/project/instance"
 import { EventSequenceTable, EventTable } from "./event.sql"
 import { EventID } from "./schema"
 import { Flag } from "@opencode-ai/core/flag/flag"
@@ -84,6 +86,30 @@ export namespace SyncEvent {
       }
     })
 
+  function currentProjectBus(): ProjectBus.Interface | undefined {
+    const fiber = Fiber.getCurrent()
+    if (!fiber) return undefined
+    try {
+      return Context.getUnsafe(fiber.context, ProjectBus.Service)
+    } catch {
+      return undefined
+    }
+  }
+
+  function publishProjectEvent(def: Definition, data: Record<string, unknown>) {
+    const bus = currentProjectBus()
+    if (bus) {
+      Effect.runSync(bus.publish({ type: def.type, properties: def.schema }, data))
+      return
+    }
+
+    GlobalBus.emit("event", {
+      directory: Instance.directory,
+      project: Instance.project.id,
+      payload: { type: def.type, properties: data },
+    })
+  }
+
   export const layer = Layer.succeed(
     Service,
     Service.of({
@@ -205,10 +231,10 @@ export namespace SyncEvent {
           const result = convertEvent(def.type, event.data)
           if (result instanceof Promise) {
             result.then((data) => {
-              ProjectBus.publish({ type: def.type, properties: def.schema }, data)
+              publishProjectEvent(def, data)
             })
           } else {
-            ProjectBus.publish({ type: def.type, properties: def.schema }, result)
+            publishProjectEvent(def, result)
           }
         }
       })
diff --git a/packages/opencode/test/bus/bus-integration.test.ts b/packages/opencode/test/bus/bus-integration.test.ts
index e42bd5299..8b3306449 100644
--- a/packages/opencode/test/bus/bus-integration.test.ts
+++ b/packages/opencode/test/bus/bus-integration.test.ts
@@ -2,6 +2,7 @@ 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 { AppRuntime } from "../../src/effect/app-runtime"
 import { Instance } from "../../src/project/instance"
 import { tmpdir } from "../fixture/fixture"
 
@@ -11,6 +12,21 @@ function withInstance(directory: string, fn: () => Promise<void>) {
   return Instance.provide({ directory, fn })
 }
 
+function publish<D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) {
+  return AppRuntime.runPromise(Bus.Service.use((bus) => bus.publish(def, properties)))
+}
+
+function subscribe<D extends BusEvent.Definition>(
+  def: D,
+  callback: (event: { type: D["type"]; properties: z.infer<D["properties"]> }) => unknown,
+) {
+  return AppRuntime.runSync(Bus.Service.use((bus) => bus.subscribeCallback(def, callback)))
+}
+
+function subscribeAll(callback: (event: any) => unknown) {
+  return AppRuntime.runSync(Bus.Service.use((bus) => bus.subscribeAllCallback(callback)))
+}
+
 describe("Bus integration: acquireRelease subscriber pattern", () => {
   afterEach(() => Instance.disposeAll())
 
@@ -19,19 +35,19 @@ describe("Bus integration: acquireRelease subscriber pattern", () => {
     const received: number[] = []
 
     await withInstance(tmp.path, async () => {
-      const unsub = Bus.subscribe(TestEvent, (evt) => {
+      const unsub = 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 publish(TestEvent, { value: 1 })
+      await 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 publish(TestEvent, { value: 3 })
       await Bun.sleep(10)
 
       expect(received).toEqual([1, 2])
@@ -45,12 +61,12 @@ describe("Bus integration: acquireRelease subscriber pattern", () => {
     const OtherEvent = BusEvent.define("test.other", z.object({ value: z.number() }))
 
     await withInstance(tmp.path, async () => {
-      Bus.subscribeAll((evt) => {
+      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 publish(TestEvent, { value: 10 })
+      await publish(OtherEvent, { value: 20 })
       await Bun.sleep(10)
     })
 
@@ -66,7 +82,7 @@ describe("Bus integration: acquireRelease subscriber pattern", () => {
     let disposed = false
 
     await withInstance(tmp.path, async () => {
-      Bus.subscribeAll((evt) => {
+      subscribeAll((evt) => {
         if (evt.type === Bus.InstanceDisposed.type) {
           disposed = true
           return
@@ -74,7 +90,7 @@ describe("Bus integration: acquireRelease subscriber pattern", () => {
         received.push(evt.properties.value)
       })
       await Bun.sleep(10)
-      await Bus.publish(TestEvent, { value: 1 })
+      await publish(TestEvent, { value: 1 })
       await Bun.sleep(10)
     })
 
diff --git a/packages/opencode/test/bus/bus.test.ts b/packages/opencode/test/bus/bus.test.ts
index 3df179787..59cf777c3 100644
--- a/packages/opencode/test/bus/bus.test.ts
+++ b/packages/opencode/test/bus/bus.test.ts
@@ -2,6 +2,7 @@ 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 { AppRuntime } from "../../src/effect/app-runtime"
 import { Instance } from "../../src/project/instance"
 import { tmpdir } from "../fixture/fixture"
 
@@ -14,6 +15,21 @@ function withInstance(directory: string, fn: () => Promise<void>) {
   return Instance.provide({ directory, fn })
 }
 
+function publish<D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) {
+  return AppRuntime.runPromise(Bus.Service.use((bus) => bus.publish(def, properties)))
+}
+
+function subscribe<D extends BusEvent.Definition>(
+  def: D,
+  callback: (event: { type: D["type"]; properties: z.infer<D["properties"]> }) => unknown,
+) {
+  return AppRuntime.runSync(Bus.Service.use((bus) => bus.subscribeCallback(def, callback)))
+}
+
+function subscribeAll(callback: (event: any) => unknown) {
+  return AppRuntime.runSync(Bus.Service.use((bus) => bus.subscribeAllCallback(callback)))
+}
+
 describe("Bus", () => {
   afterEach(() => Instance.disposeAll())
 
@@ -23,10 +39,10 @@ describe("Bus", () => {
       const received: number[] = []
 
       await withInstance(tmp.path, async () => {
-        Bus.subscribe(TestEvent.Ping, (evt) => {
+        subscribe(TestEvent.Ping, (evt) => {
           received.push(evt.properties.value)
         })
-        await Bus.publish(TestEvent.Ping, { value: 42 })
+        await publish(TestEvent.Ping, { value: 42 })
         await Bun.sleep(10)
       })
 
@@ -38,13 +54,13 @@ describe("Bus", () => {
       const received: number[] = []
 
       await withInstance(tmp.path, async () => {
-        Bus.subscribe(TestEvent.Ping, (evt) => {
+        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 })
+        await publish(TestEvent.Ping, { value: 42 })
+        await publish(TestEvent.Ping, { value: 99 })
         // Give subscriber time to process
         await Bun.sleep(10)
       })
@@ -57,12 +73,12 @@ describe("Bus", () => {
       const pings: number[] = []
 
       await withInstance(tmp.path, async () => {
-        Bus.subscribe(TestEvent.Ping, (evt) => {
+        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 publish(TestEvent.Pong, { message: "hello" })
+        await publish(TestEvent.Ping, { value: 1 })
         await Bun.sleep(10)
       })
 
@@ -73,7 +89,7 @@ describe("Bus", () => {
       await using tmp = await tmpdir()
 
       await withInstance(tmp.path, async () => {
-        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await publish(TestEvent.Ping, { value: 1 })
       })
     })
   })
@@ -84,15 +100,15 @@ describe("Bus", () => {
       const received: number[] = []
 
       await withInstance(tmp.path, async () => {
-        const unsub = Bus.subscribe(TestEvent.Ping, (evt) => {
+        const unsub = subscribe(TestEvent.Ping, (evt) => {
           received.push(evt.properties.value)
         })
         await Bun.sleep(10)
-        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await publish(TestEvent.Ping, { value: 1 })
         await Bun.sleep(10)
         unsub()
         await Bun.sleep(10)
-        await Bus.publish(TestEvent.Ping, { value: 2 })
+        await publish(TestEvent.Ping, { value: 2 })
         await Bun.sleep(10)
       })
 
@@ -106,10 +122,10 @@ describe("Bus", () => {
       const received: string[] = []
 
       await withInstance(tmp.path, async () => {
-        Bus.subscribeAll((evt) => {
+        subscribeAll((evt) => {
           received.push(evt.type)
         })
-        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await publish(TestEvent.Ping, { value: 1 })
         await Bun.sleep(10)
       })
 
@@ -121,12 +137,12 @@ describe("Bus", () => {
       const received: string[] = []
 
       await withInstance(tmp.path, async () => {
-        Bus.subscribeAll((evt) => {
+        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 publish(TestEvent.Ping, { value: 1 })
+        await publish(TestEvent.Pong, { message: "hi" })
         await Bun.sleep(10)
       })
 
@@ -142,14 +158,14 @@ describe("Bus", () => {
       const b: number[] = []
 
       await withInstance(tmp.path, async () => {
-        Bus.subscribe(TestEvent.Ping, (evt) => {
+        subscribe(TestEvent.Ping, (evt) => {
           a.push(evt.properties.value)
         })
-        Bus.subscribe(TestEvent.Ping, (evt) => {
+        subscribe(TestEvent.Ping, (evt) => {
           b.push(evt.properties.value)
         })
         await Bun.sleep(10)
-        await Bus.publish(TestEvent.Ping, { value: 7 })
+        await publish(TestEvent.Ping, { value: 7 })
         await Bun.sleep(10)
       })
 
@@ -166,26 +182,26 @@ describe("Bus", () => {
       const receivedB: number[] = []
 
       await withInstance(tmpA.path, async () => {
-        Bus.subscribe(TestEvent.Ping, (evt) => {
+        subscribe(TestEvent.Ping, (evt) => {
           receivedA.push(evt.properties.value)
         })
         await Bun.sleep(10)
       })
 
       await withInstance(tmpB.path, async () => {
-        Bus.subscribe(TestEvent.Ping, (evt) => {
+        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 publish(TestEvent.Ping, { value: 1 })
         await Bun.sleep(10)
       })
 
       await withInstance(tmpB.path, async () => {
-        await Bus.publish(TestEvent.Ping, { value: 2 })
+        await publish(TestEvent.Ping, { value: 2 })
         await Bun.sleep(10)
       })
 
@@ -200,11 +216,11 @@ describe("Bus", () => {
       const received: string[] = []
 
       await withInstance(tmp.path, async () => {
-        Bus.subscribeAll((evt) => {
+        subscribeAll((evt) => {
           received.push(evt.type)
         })
         await Bun.sleep(10)
-        await Bus.publish(TestEvent.Ping, { value: 1 })
+        await publish(TestEvent.Ping, { value: 1 })
         await Bun.sleep(10)
       })
 
diff --git a/packages/opencode/test/effect/legacy-boundaries.test.ts b/packages/opencode/test/effect/legacy-boundaries.test.ts
index af732a4ca..ec66f7adb 100644
--- a/packages/opencode/test/effect/legacy-boundaries.test.ts
+++ b/packages/opencode/test/effect/legacy-boundaries.test.ts
@@ -331,6 +331,23 @@ test("Config service does not expose Promise facades", async () => {
   for (const facade of facades) expect(text).not.toContain(facade)
 })
 
+test("Bus and Automation services do not expose runtime facades", async () => {
+  const services = {
+    "bus/index.ts": [
+      "export async function publish",
+      "export function subscribe",
+      "export function subscribeAll",
+    ],
+    "automation/index.ts": ["const automationRuntime = makeRuntime"],
+  }
+
+  for (const [file, facades] of Object.entries(services)) {
+    const text = await readFile(path.join(srcRoot, file), "utf8")
+    expect(text).not.toMatch(/\bfrom\s+["']@\/effect\/run-service["']/)
+    for (const facade of facades) expect(text).not.toContain(facade)
+  }
+})
+
 test("Plugin and Skill services do not expose Promise facades", async () => {
   const services = {
     "plugin/index.ts": ["export async function trigger", "export async function list", "export async function init"],
diff --git a/packages/opencode/test/file/watcher.test.ts b/packages/opencode/test/file/watcher.test.ts
index abe1cf171..291e26442 100644
--- a/packages/opencode/test/file/watcher.test.ts
+++ b/packages/opencode/test/file/watcher.test.ts
@@ -10,6 +10,7 @@ import { FileWatcher } from "../../src/file/watcher"
 import { Git } from "../../src/git"
 import { Instance } from "../../src/project/instance"
 import { shouldRunNativeWatcherTests } from "./native-watcher-ci-guard"
+import { subscribeBus } from "../lib/bus"
 
 // Native @parcel/watcher bindings aren't reliably available in CI (missing on Linux, flaky on Windows)
 const describeWatcher = shouldRunNativeWatcherTests(FileWatcher.hasNativeBinding) ? describe : describe.skip
@@ -34,6 +35,7 @@ function withWatcher<E>(directory: string, body: Effect.Effect<void, E>) {
     directory,
     fn: async () => {
       const layer: Layer.Layer<FileWatcher.Service, never, never> = FileWatcher.layer.pipe(
+        Layer.provide(Bus.defaultLayer),
         Layer.provide(Config.defaultLayer),
         Layer.provide(Git.defaultLayer),
         Layer.provide(watcherConfigLayer),
@@ -53,7 +55,7 @@ function withWatcher<E>(directory: string, body: Effect.Effect<void, E>) {
 function listen(directory: string, check: (evt: WatcherEvent) => boolean, hit: (evt: WatcherEvent) => void) {
   let done = false
 
-  const unsub = Bus.subscribe(FileWatcher.Event.Updated, (evt) => {
+  const unsub = subscribeBus(FileWatcher.Event.Updated, (evt) => {
     if (done) return
     if (!check(evt.properties)) return
     hit(evt.properties)
@@ -86,7 +88,7 @@ function waitRescan(check: (evt: RescanEvent) => boolean) {
     const deferred = yield* Deferred.make<RescanEvent>()
     const cleanup = yield* Effect.sync(() => {
       let done = false
-      const unsub = Bus.subscribe(FileWatcher.Event.Rescan, (evt) => {
+      const unsub = subscribeBus(FileWatcher.Event.Rescan, (evt) => {
         if (done) return
         if (!check(evt.properties)) return
         done = true
diff --git a/packages/opencode/test/lib/bus.ts b/packages/opencode/test/lib/bus.ts
new file mode 100644
index 000000000..ea5bacdbe
--- /dev/null
+++ b/packages/opencode/test/lib/bus.ts
@@ -0,0 +1,19 @@
+import z from "zod"
+import { AppRuntime } from "../../src/effect/app-runtime"
+import { Bus } from "../../src/bus"
+import { BusEvent } from "../../src/bus/bus-event"
+
+export function publishBus<D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) {
+  return AppRuntime.runPromise(Bus.Service.use((bus) => bus.publish(def, properties)))
+}
+
+export function subscribeBus<D extends BusEvent.Definition>(
+  def: D,
+  callback: (event: { type: D["type"]; properties: z.infer<D["properties"]> }) => unknown,
+) {
+  return AppRuntime.runSync(Bus.Service.use((bus) => bus.subscribeCallback(def, callback)))
+}
+
+export function subscribeAllBus(callback: (event: any) => unknown) {
+  return AppRuntime.runSync(Bus.Service.use((bus) => bus.subscribeAllCallback(callback)))
+}
diff --git a/packages/opencode/test/lsp/client.test.ts b/packages/opencode/test/lsp/client.test.ts
index 98fec4f11..f5d91fa34 100644
--- a/packages/opencode/test/lsp/client.test.ts
+++ b/packages/opencode/test/lsp/client.test.ts
@@ -1,4 +1,5 @@
 import { describe, expect, test, beforeEach } from "bun:test"
+import { Effect } from "effect"
 import path from "path"
 import { LSPClient } from "../../src/lsp/client"
 import { LSPServer } from "../../src/lsp/server"
@@ -16,6 +17,11 @@ function spawnFakeServer() {
   }
 }
 
+const fakeBus = {
+  publish: () => Effect.void,
+  subscribeCallback: () => Effect.succeed(() => {}),
+}
+
 describe("LSPClient interop", () => {
   beforeEach(async () => {
     await Log.init({ print: true })
@@ -28,6 +34,7 @@ describe("LSPClient interop", () => {
       directory: process.cwd(),
       fn: () =>
         LSPClient.create({
+          bus: fakeBus,
           serverID: "fake",
           server: handle as unknown as LSPServer.Handle,
           root: process.cwd(),
@@ -52,6 +59,7 @@ describe("LSPClient interop", () => {
       directory: process.cwd(),
       fn: () =>
         LSPClient.create({
+          bus: fakeBus,
           serverID: "fake",
           server: handle as unknown as LSPServer.Handle,
           root: process.cwd(),
@@ -76,6 +84,7 @@ describe("LSPClient interop", () => {
       directory: process.cwd(),
       fn: () =>
         LSPClient.create({
+          bus: fakeBus,
           serverID: "fake",
           server: handle as unknown as LSPServer.Handle,
           root: process.cwd(),
diff --git a/packages/opencode/test/mcp/lifecycle.test.ts b/packages/opencode/test/mcp/lifecycle.test.ts
index fd259eaca..67cb464b3 100644
--- a/packages/opencode/test/mcp/lifecycle.test.ts
+++ b/packages/opencode/test/mcp/lifecycle.test.ts
@@ -248,7 +248,7 @@ beforeEach(() => {
 
 // Import after mocks
 const { MCP } = await import("../../src/mcp/index")
-const { Bus } = await import("../../src/bus/index")
+const { subscribeBus } = await import("../lib/bus")
 const { Instance } = await import("../../src/project/instance")
 const { NotFoundError } = await import("../../src/storage/db")
 const { tmpdir } = await import("../fixture/fixture")
@@ -617,7 +617,7 @@ test("tool change notification reaches the instance bus from a detached callback
 
       await MCPFacade.add("notify-server", { type: "local", command: ["echo", "test"] })
 
-      unsubscribe = Bus.subscribe(MCP.ToolsChanged, (event) => {
+      unsubscribe = subscribeBus(MCP.ToolsChanged, (event) => {
         received.push(event.properties.server)
       })
 
diff --git a/packages/opencode/test/mcp/oauth-browser.test.ts b/packages/opencode/test/mcp/oauth-browser.test.ts
index d30f33e0f..f6d94f186 100644
--- a/packages/opencode/test/mcp/oauth-browser.test.ts
+++ b/packages/opencode/test/mcp/oauth-browser.test.ts
@@ -100,7 +100,7 @@ beforeEach(() => {
 
 // Import modules after mocking
 const { MCP } = await import("../../src/mcp/index")
-const { Bus } = await import("../../src/bus")
+const { subscribeBus } = await import("../lib/bus")
 const { McpOAuthCallback } = await import("../../src/mcp/oauth-callback")
 const { Instance } = await import("../../src/project/instance")
 const { tmpdir } = await import("../fixture/fixture")
@@ -134,7 +134,7 @@ test("BrowserOpenFailed event is published when open() throws", async () => {
       openShouldFail = true
 
       const events: Array<{ mcpName: string; url: string }> = []
-      const unsubscribe = Bus.subscribe(MCP.BrowserOpenFailed, (evt) => {
+      const unsubscribe = subscribeBus(MCP.BrowserOpenFailed, (evt) => {
         events.push(evt.properties)
       })
 
@@ -185,7 +185,7 @@ test("BrowserOpenFailed event is NOT published when open() succeeds", async () =
       openShouldFail = false
 
       const events: Array<{ mcpName: string; url: string }> = []
-      const unsubscribe = Bus.subscribe(MCP.BrowserOpenFailed, (evt) => {
+      const unsubscribe = subscribeBus(MCP.BrowserOpenFailed, (evt) => {
         events.push(evt.properties)
       })
 
diff --git a/packages/opencode/test/permission/next.test.ts b/packages/opencode/test/permission/next.test.ts
index 2da45eb5e..c3edb7b02 100644
--- a/packages/opencode/test/permission/next.test.ts
+++ b/packages/opencode/test/permission/next.test.ts
@@ -2,7 +2,6 @@ import { afterEach, test, expect } from "bun:test"
 import fs from "node:fs/promises"
 import os from "os"
 import path from "node:path"
-import { Bus } from "../../src/bus"
 import { AppRuntime } from "../../src/effect/app-runtime"
 import { Permission } from "../../src/permission"
 import { fromDeniedRule, isPermanentDeleteRule, permanentDeleteSuggestions } from "../../src/permission/diagnostic"
@@ -10,6 +9,7 @@ import { PermissionID } from "../../src/permission/schema"
 import { Instance } from "../../src/project/instance"
 import { tmpdir } from "../fixture/fixture"
 import { MessageID, SessionID } from "../../src/session/schema"
+import { subscribeBus } from "../lib/bus"
 import { NotFoundError } from "../../src/storage/db"
 
 afterEach(async () => {
@@ -1093,7 +1093,7 @@ test("ask - publishes asked event", async () => {
     directory: tmp.path,
     fn: async () => {
       let seen: Permission.Request | undefined
-      const unsub = Bus.subscribe(Permission.Event.Asked, (event) => {
+      const unsub = subscribeBus(Permission.Event.Asked, (event) => {
         seen = event.properties
       })
 
@@ -1402,7 +1402,7 @@ test("reply - publishes replied event", async () => {
             reply: Permission.Reply
           }
         | undefined
-      const unsub = Bus.subscribe(Permission.Event.Replied, (event) => {
+      const unsub = subscribeBus(Permission.Event.Replied, (event) => {
         seen = event.properties
       })
 
diff --git a/packages/opencode/test/pty/pty-session.test.ts b/packages/opencode/test/pty/pty-session.test.ts
index c87f38461..d2105d48d 100644
--- a/packages/opencode/test/pty/pty-session.test.ts
+++ b/packages/opencode/test/pty/pty-session.test.ts
@@ -1,6 +1,5 @@
 import { describe, expect, test } from "bun:test"
 import { Effect } from "effect"
-import { Bus } from "../../src/bus"
 import { AppRuntime } from "../../src/effect/app-runtime"
 import { Instance } from "../../src/project/instance"
 import { Pty } from "../../src/pty"
@@ -8,6 +7,7 @@ import type { PtyID } from "../../src/pty/schema"
 import { tmpdir } from "../fixture/fixture"
 import { setTimeout as sleep } from "node:timers/promises"
 import { existsSync } from "node:fs"
+import { subscribeBus } from "../lib/bus"
 
 const wait = async (fn: () => boolean, ms = 5000) => {
   const end = Date.now() + ms
@@ -37,9 +37,9 @@ describe("pty", () => {
       fn: async () => {
         const log: Array<{ type: "created" | "exited" | "deleted"; id: PtyID }> = []
         const off = [
-          Bus.subscribe(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
-          Bus.subscribe(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
-          Bus.subscribe(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
+          subscribeBus(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
+          subscribeBus(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
+          subscribeBus(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
         ]
 
         let id: PtyID | undefined
@@ -74,9 +74,9 @@ describe("pty", () => {
       fn: async () => {
         const log: Array<{ type: "created" | "exited" | "deleted"; id: PtyID }> = []
         const off = [
-          Bus.subscribe(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
-          Bus.subscribe(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
-          Bus.subscribe(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
+          subscribeBus(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
+          subscribeBus(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
+          subscribeBus(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
         ]
 
         let id: PtyID | undefined
diff --git a/packages/opencode/test/server/automation-routes.test.ts b/packages/opencode/test/server/automation-routes.test.ts
index 8adf2633c..2d062fb4e 100644
--- a/packages/opencode/test/server/automation-routes.test.ts
+++ b/packages/opencode/test/server/automation-routes.test.ts
@@ -7,7 +7,7 @@ import { Log } from "@opencode-ai/core/util/log"
 import { Automation, AutomationID } from "../../src/automation"
 import { AutomationRunTable } from "../../src/automation/automation.sql"
 import { AutomationScheduler } from "../../src/automation/scheduler"
-import { Bus } from "../../src/bus"
+import { GlobalBus } from "../../src/bus/global"
 import { Instance } from "../../src/project/instance"
 import { ProjectID } from "../../src/project/schema"
 import { Server } from "../../src/server/server"
@@ -26,6 +26,20 @@ import { tmpdir } from "../fixture/fixture"
 
 void Log.init({ print: false })
 
+function subscribeAutomationEvent<D extends { type: string; properties: { parse(input: unknown): any } }>(
+  def: D,
+  callback: (event: { type: D["type"]; properties: ReturnType<D["properties"]["parse"]> }) => unknown,
+  options?: { directory?: string },
+) {
+  const listener = (event: { directory?: string; payload?: { type?: string; properties?: unknown } }) => {
+    if (options?.directory && event.directory !== options.directory) return
+    if (event.payload?.type !== def.type) return
+    callback({ type: def.type, properties: def.properties.parse(event.payload.properties) })
+  }
+  GlobalBus.on("event", listener)
+  return () => GlobalBus.off("event", listener)
+}
+
 const previousSkipAutomationModelValidation = process.env.OPENCODE_SKIP_AUTOMATION_MODEL_VALIDATION
 
 beforeAll(() => {
@@ -667,10 +681,10 @@ describe("automation routes", () => {
       const releaseSettle = deferred<void>()
       const publication = deferred<string>()
       let publishedID: string | undefined
-      const unsubscribe = Bus.subscribe(Automation.Event.DefinitionUpdated, (event) => {
+      const unsubscribe = subscribeAutomationEvent(Automation.Event.DefinitionUpdated, (event) => {
         publishedID = event.properties.id
         publication.resolve(event.properties.id)
-      })
+      }, { directory: Instance.directory })
       AutomationScheduler.install({
         stop: () => undefined,
         settleOwner: async () => {
@@ -1222,9 +1236,9 @@ describe("automation routes", () => {
         body: JSON.stringify(recurringInput(projectID)),
       })
       const revisions: number[] = []
-      const unsubscribe = Bus.subscribe(Automation.Event.DefinitionUpdated, (event) => {
+      const unsubscribe = subscribeAutomationEvent(Automation.Event.DefinitionUpdated, (event) => {
         revisions.push(event.properties.revision)
-      })
+      }, { directory: Instance.directory })
 
       const empty = await json(app, `/automation/${created.id}`, {
         method: "PUT",
@@ -1298,9 +1312,9 @@ describe("automation routes", () => {
       })
       expect(created).not.toHaveProperty("variant")
       const updates: number[] = []
-      const unsubscribe = Bus.subscribe(Automation.Event.DefinitionUpdated, (event) => {
+      const unsubscribe = subscribeAutomationEvent(Automation.Event.DefinitionUpdated, (event) => {
         if (event.properties.id === created.id) updates.push(event.properties.revision)
-      })
+      }, { directory: Instance.directory })
       const noop = await json(app, `/automation/${created.id}`, {
         method: "PUT",
         headers: { "content-type": "application/json" },
@@ -1322,9 +1336,9 @@ describe("automation routes", () => {
         body: JSON.stringify(recurringInput(projectID)),
       })
       const revisions: number[] = []
-      const unsubscribe = Bus.subscribe(Automation.Event.DefinitionUpdated, (event) => {
+      const unsubscribe = subscribeAutomationEvent(Automation.Event.DefinitionUpdated, (event) => {
         revisions.push(event.properties.revision)
-      })
+      }, { directory: Instance.directory })
 
       const paused = await json(app, `/automation/${created.id}/pause`, { method: "POST" })
       const pausedAgain = await json(app, `/automation/${created.id}/pause`, { method: "POST" })
diff --git a/packages/opencode/test/server/automation-runner.test.ts b/packages/opencode/test/server/automation-runner.test.ts
index c80c6e194..ae9ff6d4f 100644
--- a/packages/opencode/test/server/automation-runner.test.ts
+++ b/packages/opencode/test/server/automation-runner.test.ts
@@ -7,7 +7,7 @@ import { Automation } from "../../src/automation"
 import { sessionPromptExecutor } from "../../src/automation/runner"
 import { AutomationRunTable } from "../../src/automation/automation.sql"
 import { AppRuntime } from "../../src/effect/app-runtime"
-import { Bus } from "../../src/bus"
+import { GlobalBus } from "../../src/bus/global"
 import { Database, eq } from "../../src/storage/db"
 import { Instance } from "../../src/project/instance"
 import { Project } from "../../src/project/project"
@@ -24,6 +24,20 @@ import { tmpdir } from "../fixture/fixture"
 const RUN_WAIT_TIMEOUT_MS = 10_000
 const runSession = <A>(fn: (svc: Session.Interface) => Effect.Effect<A>) => AppRuntime.runPromise(Session.Service.use(fn))
 
+function subscribeAutomationEvent<D extends { type: string; properties: { parse(input: unknown): any } }>(
+  def: D,
+  callback: (event: { type: D["type"]; properties: ReturnType<D["properties"]["parse"]> }) => unknown,
+  options?: { directory?: string },
+) {
+  const listener = (event: { directory?: string; payload?: { type?: string; properties?: unknown } }) => {
+    if (options?.directory && event.directory !== options.directory) return
+    if (event.payload?.type !== def.type) return
+    callback({ type: def.type, properties: def.properties.parse(event.payload.properties) })
+  }
+  GlobalBus.on("event", listener)
+  return () => GlobalBus.off("event", listener)
+}
+
 afterEach(async () => {
   await Instance.disposeAll()
 })
@@ -205,9 +219,9 @@ describe("automation runNow execution", () => {
       const definition = Automation.create(input(projectID))
       const sessionID = SessionID.descending()
       const runEvents: Automation.Run[] = []
-      const unsubscribeRun = Bus.subscribe(Automation.Event.RunUpdated, (event) => {
+      const unsubscribeRun = subscribeAutomationEvent(Automation.Event.RunUpdated, (event) => {
         if (event.properties.automationID === definition.id) runEvents.push(event.properties)
-      })
+      }, { directory: Instance.directory })
 
       await Automation.runNowExecuting(definition.id, {
         executor: async ({ run }) => {
@@ -408,9 +422,9 @@ describe("automation runNow execution", () => {
       const definition = Automation.create(input(projectID))
       const runEvents: Automation.Run[] = []
       const executorFinished = defer<void>()
-      const unsubscribeRun = Bus.subscribe(Automation.Event.RunUpdated, (event) => {
+      const unsubscribeRun = subscribeAutomationEvent(Automation.Event.RunUpdated, (event) => {
         if (event.properties.automationID === definition.id) runEvents.push(event.properties)
-      })
+      }, { directory: Instance.directory })
 
       await Automation.runNowExecuting(definition.id, {
         executor: async ({ run }) => {
@@ -439,9 +453,9 @@ describe("automation runNow execution", () => {
       const fresh = Automation.create(input(projectID, { context: "fresh" }))
 
       const deletedEvents: Automation.Tombstone[] = []
-      const unsubscribe = Bus.subscribe(Automation.Event.DefinitionDeleted, (event) => {
+      const unsubscribe = subscribeAutomationEvent(Automation.Event.DefinitionDeleted, (event) => {
         deletedEvents.push(event.properties)
-      })
+      }, { directory: Instance.directory })
 
       await Automation.deleteBySourceSession(sourceSessionID)
       unsubscribe()
@@ -494,9 +508,9 @@ describe("automation runNow execution", () => {
         sourceSessionID: SessionID.descending(),
       })
       const definitionEvents: Automation.Definition[] = []
-      const unsubscribeDefinition = Bus.subscribe(Automation.Event.DefinitionUpdated, (event) => {
+      const unsubscribeDefinition = subscribeAutomationEvent(Automation.Event.DefinitionUpdated, (event) => {
         definitionEvents.push(event.properties)
-      })
+      }, { directory: Instance.directory })
       let removed!: Awaited<ReturnType<typeof Automation.remove>>
 
       const initial = await Automation.runNowExecuting(definition.id, {
@@ -543,9 +557,9 @@ describe("automation runNow execution", () => {
       const started = Promise.withResolvers<void>()
       const release = Promise.withResolvers<void>()
       const runEvents: Automation.Run[] = []
-      const unsubscribeRun = Bus.subscribe(Automation.Event.RunUpdated, (event) => {
+      const unsubscribeRun = subscribeAutomationEvent(Automation.Event.RunUpdated, (event) => {
         if (event.properties.automationID === definition.id) runEvents.push(event.properties)
-      })
+      }, { directory: Instance.directory })
 
       const initial = await Automation.runNowExecuting(definition.id, {
         executor: async ({ run, signal }) => {
@@ -1147,10 +1161,10 @@ describe("automation runNow execution", () => {
         fn: async () => {
           const definition = Automation.create(input(Instance.project.id, { title: "Cancel before runner busy" }))
           const removed = Promise.withResolvers<Awaited<ReturnType<typeof Automation.remove>>>()
-          const unsubscribe = Bus.subscribe(Automation.Event.RunUpdated, (event) => {
+          const unsubscribe = subscribeAutomationEvent(Automation.Event.RunUpdated, (event) => {
             if (event.properties.automationID !== definition.id || event.properties.state !== "running") return
             void Automation.remove(definition.id).then(removed.resolve, removed.reject)
-          })
+          }, { directory: Instance.directory })
 
           const initial = await Automation.runNowExecuting(definition.id, { executor: sessionPromptExecutor })
           let result: Awaited<ReturnType<typeof Automation.remove>>
diff --git a/packages/opencode/test/server/automation-scheduler.test.ts b/packages/opencode/test/server/automation-scheduler.test.ts
index e5b8e9394..186307559 100644
--- a/packages/opencode/test/server/automation-scheduler.test.ts
+++ b/packages/opencode/test/server/automation-scheduler.test.ts
@@ -3,7 +3,7 @@ import { Effect, ManagedRuntime } from "effect"
 import { Automation } from "../../src/automation"
 import { internalTestHooks } from "../../src/automation/__test_hooks"
 import { AutomationScheduler } from "../../src/automation/scheduler"
-import { Bus } from "../../src/bus"
+import { GlobalBus } from "../../src/bus/global"
 import { Instance } from "../../src/project/instance"
 import { ProjectID } from "../../src/project/schema"
 import { trackActiveRun } from "../../src/session/lifecycle-provenance"
@@ -18,6 +18,14 @@ import { Flock } from "../../src/util/flock"
 const runtime = ManagedRuntime.make(Automation.defaultLayer)
 const automation = await runtime.runPromise(Effect.gen(function* () { return yield* Automation.Service }))
 
+function publishAutomationEvent<D extends { type: string }>(def: D, properties: unknown) {
+  GlobalBus.emit("event", {
+    directory: Instance.directory,
+    project: Instance.project.id,
+    payload: { type: def.type, properties },
+  })
+}
+
 afterEach(async () => {
   AutomationScheduler.stopProcess({ stopRuns: false })
   await Instance.disposeAll()
@@ -766,7 +774,7 @@ describe("automation scheduler", () => {
       const definition = Automation.create(oneshotInput(projectID, 1_000), { now: 0 })
 
       scheduler.reschedule(definition)
-      await Bus.publish(Automation.Event.DefinitionDeleted, { id: definition.id, deleted: true, revision: 2 })
+      publishAutomationEvent(Automation.Event.DefinitionDeleted, { id: definition.id, deleted: true, revision: 2 })
       await clock.advance(1_000)
 
       expect(calls).toEqual([])
diff --git a/packages/opencode/test/server/event-stream-routes.test.ts b/packages/opencode/test/server/event-stream-routes.test.ts
index da3ec2ae0..9f6844b2f 100644
--- a/packages/opencode/test/server/event-stream-routes.test.ts
+++ b/packages/opencode/test/server/event-stream-routes.test.ts
@@ -1,6 +1,5 @@
 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 { EventReplayStore } from "../../src/server/event-replay"
@@ -12,6 +11,7 @@ import {
 } from "../../src/server/instance/global"
 import type { AsyncQueue } from "../../src/util/queue"
 import { tmpdir } from "../fixture/fixture"
+import { publishBus } from "../lib/bus"
 
 type SseFrame = {
   event?: string
@@ -264,7 +264,7 @@ describe("SSE event routes", () => {
           expect(connected.id).toBeUndefined()
           expect(connected.data).toEqual({ type: "server.connected", properties: {} })
 
-          await Bus.publish(TestEvent, { value: 7 })
+          await publishBus(TestEvent, { value: 7 })
           const [event, heartbeat] = await reader.read(2)
           expectNoSseControlFields([event, heartbeat])
           expect(event.id).toBeUndefined()
diff --git a/packages/opencode/test/session/session.test.ts b/packages/opencode/test/session/session.test.ts
index 7a3762410..dbd56f493 100644
--- a/packages/opencode/test/session/session.test.ts
+++ b/packages/opencode/test/session/session.test.ts
@@ -3,7 +3,6 @@ import { Effect } from "effect"
 import fs from "fs/promises"
 import path from "path"
 import { Session as SessionCore } from "../../src/session"
-import { Bus } from "../../src/bus"
 import { Log } from "@opencode-ai/core/util/log"
 import { Instance } from "../../src/project/instance"
 import { MessageV2 } from "../../src/session/message-v2"
@@ -15,6 +14,7 @@ import { MessageTable, SessionTable } from "../../src/session/session.sql"
 import { ProjectTable } from "../../src/project/project.sql"
 import { canonicalDirectory } from "../../src/session/execution-context"
 import { AppRuntime } from "../../src/effect/app-runtime"
+import { subscribeBus } from "../lib/bus"
 
 
 const SessionNs = {
@@ -621,7 +621,7 @@ describe("session.created event", () => {
         let eventReceived = false
         let receivedInfo: SessionNs.Info | undefined
 
-        const unsub = Bus.subscribe(SessionNs.Event.Created, (event) => {
+        const unsub = subscribeBus(SessionNs.Event.Created, (event) => {
           eventReceived = true
           receivedInfo = event.properties.info as SessionNs.Info
         })
@@ -648,11 +648,11 @@ describe("session.created event", () => {
       fn: async () => {
         const events: string[] = []
 
-        const unsubCreated = Bus.subscribe(SessionNs.Event.Created, () => {
+        const unsubCreated = subscribeBus(SessionNs.Event.Created, () => {
           events.push("created")
         })
 
-        const unsubUpdated = Bus.subscribe(SessionNs.Event.Updated, () => {
+        const unsubUpdated = subscribeBus(SessionNs.Event.Updated, () => {
           events.push("updated")
         })
 
@@ -746,7 +746,7 @@ describe("step-finish token propagation via Bus event", () => {
           // is the mutable domain type. Cast bridges the two — safe because the
           // test only reads the value afterwards.
           let received: MessageV2.Part | undefined
-          const unsub = Bus.subscribe(MessageV2.Event.PartUpdated, (event) => {
+          const unsub = subscribeBus(MessageV2.Event.PartUpdated, (event) => {
             received = event.properties.part as MessageV2.Part
           })
 
diff --git a/packages/opencode/test/session/turn-change-aggregate.test.ts b/packages/opencode/test/session/turn-change-aggregate.test.ts
index c5be23c1b..1a6568d86 100644
--- a/packages/opencode/test/session/turn-change-aggregate.test.ts
+++ b/packages/opencode/test/session/turn-change-aggregate.test.ts
@@ -9,10 +9,10 @@ import { MessageID, PartID, SessionID } from "../../src/session/schema"
 import { ModelID, ProviderID } from "../../src/provider/schema"
 import { SessionTable } from "../../src/session/session.sql"
 import { Database, eq } from "../../src/storage/db"
-import { Bus } from "../../src/bus"
 import { AppRuntime } from "../../src/effect/app-runtime"
 import { tmpdir } from "../fixture/fixture"
 import { resetDatabase } from "../fixture/db"
+import { subscribeBus } from "../lib/bus"
 
 
 const SessionNs = {
@@ -326,7 +326,7 @@ describe("TurnChange aggregate union", () => {
         const userMessageID = await makeUser(session.id, "union-invalidate-uncaptured")
         const assistantID = await makeAssistant(session.id, userMessageID, "union-invalidate-uncaptured")
         const events: string[] = []
-        const unsubscribe = Bus.subscribe(SessionNs.Event.TurnChangeInvalidated, (event) => {
+        const unsubscribe = subscribeBus(SessionNs.Event.TurnChangeInvalidated, (event) => {
           events.push(event.properties.sessionID)
         })
         try {
@@ -351,7 +351,7 @@ describe("TurnChange aggregate union", () => {
         const userMessageID = await makeUser(session.id, "union-invalidate-captured")
         const assistantID = await makeAssistant(session.id, userMessageID, "union-invalidate-captured")
         const events: string[] = []
-        const unsubscribe = Bus.subscribe(SessionNs.Event.TurnChangeInvalidated, (event) => {
+        const unsubscribe = subscribeBus(SessionNs.Event.TurnChangeInvalidated, (event) => {
           events.push(event.properties.sessionID)
         })
         try {
diff --git a/packages/opencode/test/sync/index.test.ts b/packages/opencode/test/sync/index.test.ts
index 1fc050f74..926902930 100644
--- a/packages/opencode/test/sync/index.test.ts
+++ b/packages/opencode/test/sync/index.test.ts
@@ -2,7 +2,6 @@ import { describe, test, expect, beforeEach, afterEach, afterAll } from "bun:tes
 import { Cause, Effect, Exit } from "effect"
 import { tmpdir } from "../fixture/fixture"
 import z from "zod"
-import { Bus } from "../../src/bus"
 import { Instance } from "../../src/project/instance"
 import { SyncEvent } from "../../src/sync"
 import { Database } from "../../src/storage/db"
@@ -11,6 +10,7 @@ import { Identifier } from "../../src/id/id"
 import { Flag } from "@opencode-ai/core/flag/flag"
 import { initProjectors } from "../../src/server/projectors"
 import { testEffect } from "../lib/effect"
+import { GlobalBus } from "../../src/bus/global"
 
 const original = Flag.OPENCODE_EXPERIMENTAL_WORKSPACES
 const syncIt = testEffect(SyncEvent.defaultLayer)
@@ -114,10 +114,12 @@ describe("SyncEvent", () => {
           properties: { id: string; name: string }
         }> = []
         const received = new Promise<void>((resolve) => {
-          Bus.subscribeAll((event) => {
-            events.push(event)
+          const handler = (event: { payload: { type: string; properties: { id: string; name: string } } }) => {
+            GlobalBus.off("event", handler)
+            events.push(event.payload)
             resolve()
-          })
+          }
+          GlobalBus.on("event", handler)
         })
 
         SyncEvent.run(Created, { id: "evt_1", name: "test" })
diff --git a/packages/opencode/test/tool/automate-manage.test.ts b/packages/opencode/test/tool/automate-manage.test.ts
index 580dc9d30..e7d25d6ee 100644
--- a/packages/opencode/test/tool/automate-manage.test.ts
+++ b/packages/opencode/test/tool/automate-manage.test.ts
@@ -3,7 +3,7 @@ import { Effect, ManagedRuntime, Schema } from "effect"
 import { Automation } from "../../src/automation"
 import { AutomationRunTable } from "../../src/automation/automation.sql"
 import { AutomationScheduler } from "../../src/automation/scheduler"
-import { Bus } from "../../src/bus"
+import { GlobalBus } from "../../src/bus/global"
 import { Instance } from "../../src/project/instance"
 import { ProjectID } from "../../src/project/schema"
 import { MessageID, SessionID } from "../../src/session/schema"
@@ -20,6 +20,20 @@ const automation = await runtime.runPromise(Effect.gen(function* () {
   return yield* Automation.Service
 }))
 
+function subscribeAutomationEvent<D extends { type: string; properties: { parse(input: unknown): any } }>(
+  def: D,
+  callback: (event: { type: D["type"]; properties: ReturnType<D["properties"]["parse"]> }) => unknown,
+  options?: { directory?: string },
+) {
+  const listener = (event: { directory?: string; payload?: { type?: string; properties?: unknown } }) => {
+    if (options?.directory && event.directory !== options.directory) return
+    if (event.payload?.type !== def.type) return
+    callback({ type: def.type, properties: def.properties.parse(event.payload.properties) })
+  }
+  GlobalBus.on("event", listener)
+  return () => GlobalBus.off("event", listener)
+}
+
 function recurring(projectID: ProjectID, title: string): Automation.CreateInput {
   return {
     kind: "recurring",
@@ -171,9 +185,9 @@ describe("automate_manage tool", () => {
       fn: async () => {
         const asks: unknown[] = []
         const deletedEvents: Automation.Tombstone[] = []
-        const unsubscribe = Bus.subscribe(Automation.Event.DefinitionDeleted, (event) => {
+        const unsubscribe = subscribeAutomationEvent(Automation.Event.DefinitionDeleted, (event) => {
           deletedEvents.push(event.properties)
-        })
+        }, { directory: Instance.directory })
         installScheduler()
         const created = Automation.create(recurring(Instance.project.id, "Daily repo brief"), { now: 100 })
 
@@ -275,9 +289,9 @@ describe("automate_manage tool", () => {
         installScheduler()
         const asks: unknown[] = []
         const deletedEvents: Automation.Tombstone[] = []
-        const unsubscribe = Bus.subscribe(Automation.Event.DefinitionDeleted, (event) => {
+        const unsubscribe = subscribeAutomationEvent(Automation.Event.DefinitionDeleted, (event) => {
           deletedEvents.push(event.properties)
-        })
+        }, { directory: Instance.directory })
         const created = Automation.create(recurring(Instance.project.id, "Daily repo brief"), { now: 100 })
         const ctx = {
           ...toolContext(asks),