refactor(opencode): retire bus automation facades

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,15 +1293,19 @@ 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,
       project: scope.projectID,
       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 {

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))
-  }
 }

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) {

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)))

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(() => {})
 }

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
     })