Add global bus wait helper for server tests (#25468)

packages/opencode/test/server/global-bus.ts

@@ -0,0 +1,34 @@
+import { GlobalBus, type GlobalEvent } from "@/bus/global"
+import { Cause, Effect } from "effect"
+
+export function waitGlobalBusEvent(input: {
+  timeout?: number
+  message?: string
+  predicate: (event: GlobalEvent) => boolean
+}) {
+  return Effect.callback<GlobalEvent, unknown>((resume) => {
+    const cleanup = () => GlobalBus.off("event", handler)
+
+    const handler = (event: GlobalEvent) => {
+      try {
+        if (!input.predicate(event)) return
+        cleanup()
+        resume(Effect.succeed(event))
+      } catch (error) {
+        cleanup()
+        resume(Effect.fail(error))
+      }
+    }
+
+    GlobalBus.on("event", handler)
+    return Effect.sync(cleanup)
+  }).pipe(
+    Effect.timeout(input.timeout ?? 10_000),
+    Effect.mapError((error) =>
+      Cause.isTimeoutError(error) ? new Error(input.message ?? "timed out waiting for global bus event") : error,
+    ),
+  )
+}
+
+export const waitGlobalBusEventPromise = (input: Parameters<typeof waitGlobalBusEvent>[0]) =>
+  Effect.runPromise(waitGlobalBusEvent(input))

packages/opencode/test/server/httpapi-config.test.ts

@@ -1,12 +1,11 @@
 import { afterEach, describe, expect, test } from "bun:test"
 import path from "path"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { GlobalBus } from "@/bus/global"
-import { Instance } from "../../src/project/instance"
 import { Server } from "../../src/server/server"
 import * as Log from "@opencode-ai/core/util/log"
 import { resetDatabase } from "../fixture/db"
 import { disposeAllInstances, tmpdir } from "../fixture/fixture"
+import { waitGlobalBusEventPromise } from "./global-bus"
 
 void Log.init({ print: false })
 
@@ -18,20 +17,9 @@ function app() {
 }
 
 async function waitDisposed(directory: string) {
-  return await new Promise<void>((resolve, reject) => {
+  await waitGlobalBusEventPromise({
-    const timer = setTimeout(() => {
+    message: "timed out waiting for instance disposal",
-      GlobalBus.off("event", onEvent)
+    predicate: (event) => event.payload.type === "server.instance.disposed" && event.directory === directory,
-      reject(new Error("timed out waiting for instance disposal"))
-    }, 10_000)
-
-    function onEvent(event: { directory?: string; payload: { type?: string } }) {
-      if (event.payload.type !== "server.instance.disposed" || event.directory !== directory) return
-      clearTimeout(timer)
-      GlobalBus.off("event", onEvent)
-      resolve()
-    }
-
-    GlobalBus.on("event", onEvent)
   })
 }
 

packages/opencode/test/server/httpapi-experimental.test.ts

@@ -1,7 +1,6 @@
 import { afterEach, describe, expect, test } from "bun:test"
 import { Effect } from "effect"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { GlobalBus } from "@/bus/global"
 import { Instance } from "../../src/project/instance"
 import { Server } from "../../src/server/server"
 import { ExperimentalPaths } from "../../src/server/routes/instance/httpapi/groups/experimental"
@@ -11,6 +10,7 @@ import * as Log from "@opencode-ai/core/util/log"
 import { Worktree } from "../../src/worktree"
 import { resetDatabase } from "../fixture/db"
 import { disposeAllInstances, tmpdir } from "../fixture/fixture"
+import { waitGlobalBusEventPromise } from "./global-bus"
 
 void Log.init({ print: false })
 
@@ -31,20 +31,9 @@ function createSession(input?: Session.CreateInput) {
 }
 
 async function waitReady(directory: string) {
-  return await new Promise<void>((resolve, reject) => {
+  await waitGlobalBusEventPromise({
-    const timer = setTimeout(() => {
+    message: "timed out waiting for worktree.ready",
-      GlobalBus.off("event", onEvent)
+    predicate: (event) => event.payload.type === Worktree.Event.Ready.type && event.directory === directory,
-      reject(new Error("timed out waiting for worktree.ready"))
-    }, 10_000)
-
-    function onEvent(event: { directory?: string; payload: { type?: string } }) {
-      if (event.payload.type !== Worktree.Event.Ready.type || event.directory !== directory) return
-      clearTimeout(timer)
-      GlobalBus.off("event", onEvent)
-      resolve()
-    }
-
-    GlobalBus.on("event", onEvent)
   })
 }
 

packages/opencode/test/server/httpapi-instance-context.test.ts

@@ -1,6 +1,5 @@
 import { NodeHttpServer, NodeServices } from "@effect/platform-node"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { GlobalBus } from "@/bus/global"
 import { describe, expect } from "bun:test"
 import { Effect, Fiber, Layer } from "effect"
 import { HttpClient, HttpClientRequest, HttpRouter, HttpServerResponse } from "effect/unstable/http"
@@ -20,6 +19,7 @@ import { instanceRouterMiddleware } from "../../src/server/routes/instance/httpa
 import { workspaceRouterMiddleware } from "../../src/server/routes/instance/httpapi/middleware/workspace-routing"
 import { resetDatabase } from "../fixture/db"
 import { disposeAllInstances, tmpdirScoped } from "../fixture/fixture"
+import { waitGlobalBusEvent } from "./global-bus"
 import { testEffect } from "../lib/effect"
 
 const testStateLayer = Layer.effectDiscard(
@@ -97,24 +97,10 @@ const serveProbe = (probePath: HttpRouter.PathInput = "/probe") =>
     Layer.build,
   )
 
-const waitDisposedEvent = Effect.promise(
+const waitDisposedEvent = waitGlobalBusEvent({
-  () =>
+  message: "timed out waiting for instance disposal",
-    new Promise<{ directory?: string; workspace?: string }>((resolve, reject) => {
+  predicate: (event) => event.payload.type === "server.instance.disposed",
-      const timer = setTimeout(() => {
+}).pipe(Effect.map((event) => ({ directory: event.directory, workspace: event.workspace })))
-        GlobalBus.off("event", onEvent)
-        reject(new Error("timed out waiting for instance disposal"))
-      }, 10_000)
-
-      function onEvent(event: { directory?: string; workspace?: string; payload: { type?: string } }) {
-        if (event.payload.type !== "server.instance.disposed") return
-        clearTimeout(timer)
-        GlobalBus.off("event", onEvent)
-        resolve({ directory: event.directory, workspace: event.workspace })
-      }
-
-      GlobalBus.on("event", onEvent)
-    }),
-)
 
 const serveDisposeProbe = () =>
   HttpRouter.serve(

packages/opencode/test/server/httpapi-instance.legacy.test.ts

@@ -1,12 +1,11 @@
 import { afterEach, describe, expect, test } from "bun:test"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { GlobalBus } from "@/bus/global"
-import { Instance } from "../../src/project/instance"
 import { Server } from "../../src/server/server"
 import { InstancePaths } from "../../src/server/routes/instance/httpapi/groups/instance"
 import * as Log from "@opencode-ai/core/util/log"
 import { resetDatabase } from "../fixture/db"
 import { disposeAllInstances, tmpdir } from "../fixture/fixture"
+import { waitGlobalBusEventPromise } from "./global-bus"
 
 void Log.init({ print: false })
 
@@ -18,20 +17,9 @@ function app() {
 }
 
 async function waitDisposed(directory: string) {
-  return await new Promise<void>((resolve, reject) => {
+  await waitGlobalBusEventPromise({
-    const timer = setTimeout(() => {
+    message: "timed out waiting for instance disposal",
-      GlobalBus.off("event", onEvent)
+    predicate: (event) => event.payload.type === "server.instance.disposed" && event.directory === directory,
-      reject(new Error("timed out waiting for instance disposal"))
-    }, 10_000)
-
-    function onEvent(event: { directory?: string; payload: { type?: string } }) {
-      if (event.payload.type !== "server.instance.disposed" || event.directory !== directory) return
-      clearTimeout(timer)
-      GlobalBus.off("event", onEvent)
-      resolve()
-    }
-
-    GlobalBus.on("event", onEvent)
   })
 }
 
@@ -117,13 +105,9 @@ describe("instance HttpApi", () => {
   test("serves instance dispose through Hono bridge", async () => {
     await using tmp = await tmpdir()
 
-    const disposed = new Promise<string | undefined>((resolve) => {
+    const disposed = waitGlobalBusEventPromise({
-      const onEvent = (event: { directory?: string; payload: { type?: string } }) => {
+      message: "timed out waiting for instance disposal",
-        if (event.payload.type !== "server.instance.disposed") return
+      predicate: (event) => event.payload.type === "server.instance.disposed",
-        GlobalBus.off("event", onEvent)
-        resolve(event.directory)
-      }
-      GlobalBus.on("event", onEvent)
     })
 
     const response = await app().request(InstancePaths.dispose, {
@@ -133,6 +117,6 @@ describe("instance HttpApi", () => {
 
     expect(response.status).toBe(200)
     expect(await response.json()).toBe(true)
-    expect(await disposed).toBe(tmp.path)
+    expect((await disposed).directory).toBe(tmp.path)
   })
 })

packages/opencode/test/server/httpapi-tui.test.ts

@@ -1,7 +1,6 @@
 import { afterEach, describe, expect, test } from "bun:test"
 import type { Context } from "hono"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { GlobalBus } from "../../src/bus/global"
 import { TuiEvent } from "../../src/cli/cmd/tui/event"
 import { SessionID } from "../../src/session/schema"
 import { Instance } from "../../src/project/instance"
@@ -12,6 +11,7 @@ import * as Log from "@opencode-ai/core/util/log"
 import { OpenApi } from "effect/unstable/httpapi"
 import { resetDatabase } from "../fixture/db"
 import { disposeAllInstances, tmpdir } from "../fixture/fixture"
+import { waitGlobalBusEventPromise } from "./global-bus"
 
 void Log.init({ print: false })
 
@@ -23,14 +23,9 @@ function app(experimental = true) {
 }
 
 function nextCommandExecute() {
-  return new Promise<unknown>((resolve) => {
+  return waitGlobalBusEventPromise({
-    const listener = (event: { payload: { type?: string; properties?: { command?: unknown } } }) => {
+    predicate: (event) => event.payload.type === TuiEvent.CommandExecute.type,
-      if (event.payload.type !== TuiEvent.CommandExecute.type) return
+  }).then((event) => event.payload.properties?.command)
-      GlobalBus.off("event", listener)
-      resolve(event.payload.properties?.command)
-    }
-    GlobalBus.on("event", listener)
-  })
 }
 
 async function expectTrue(path: string, headers: Record<string, string>, body?: unknown) {