test(question): wait on question events (#27124)

packages/opencode/test/question/question.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, expect } from "bun:test"
-import { Cause, Effect, Exit, Fiber, Layer } from "effect"
+import { Cause, Effect, Exit, Fiber, Layer, Queue } from "effect"
 import { Question } from "../../src/question"
 import { Instance } from "../../src/project/instance"
 import { InstanceRuntime } from "../../src/project/instance-runtime"
@@ -8,8 +8,11 @@ import { disposeAllInstances, provideInstance, reloadTestInstance, tmpdirScoped
 import { SessionID } from "../../src/session/schema"
 import { testEffect } from "../lib/effect"
 import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
+import { Bus } from "../../src/bus"
 
-const it = testEffect(Layer.mergeAll(Question.defaultLayer, CrossSpawnSpawner.defaultLayer))
+const it = testEffect(
+  Layer.mergeAll(Question.layer.pipe(Layer.provideMerge(Bus.layer)), CrossSpawnSpawner.defaultLayer),
+)
 
 const askEffect = Effect.fn("QuestionTest.ask")(function* (input: {
   sessionID: SessionID
@@ -44,15 +47,19 @@ const rejectAll = Effect.gen(function* () {
   yield* Effect.forEach(yield* listEffect, (req) => rejectEffect(req.id), { discard: true })
 })
 
-const waitForPending = (count: number) =>
+const waitForPending = Effect.fn("QuestionTest.waitForPending")(function* (count: number) {
-  Effect.gen(function* () {
+  const question = yield* Question.Service
-    for (let i = 0; i < 100; i++) {
+  const bus = yield* Bus.Service
-      const pending = yield* listEffect
+  const asked = yield* Queue.unbounded<void>()
-      if (pending.length === count) return pending
+  const off = yield* bus.subscribeCallback(Question.Event.Asked, () => Queue.offerUnsafe(asked, undefined))
-      yield* Effect.sleep("10 millis")
+  yield* Effect.addFinalizer(() => Effect.sync(off))
-    }
+
-    return yield* Effect.fail(new Error(`timed out waiting for ${count} pending question request(s)`))
+  for (;;) {
-  })
+    const pending = yield* question.list()
+    if (pending.length === count) return pending
+    yield* Queue.take(asked).pipe(Effect.timeout("2 seconds"))
+  }
+})
 
 it.instance(
   "ask - remains pending until answered",

packages/opencode/test/tool/question.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { Effect, Fiber, Layer } from "effect"
+import { Effect, Fiber, Layer, Queue } from "effect"
 import { QuestionTool } from "../../src/tool/question"
 import { Question } from "../../src/question"
 import { SessionID, MessageID } from "../../src/session/schema"
@@ -7,6 +7,7 @@ import { Agent } from "../../src/agent/agent"
 import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
 import { Truncate } from "@/tool/truncate"
 import { testEffect } from "../lib/effect"
+import { Bus } from "../../src/bus"
 
 const ctx = {
   sessionID: SessionID.make("ses_test-session"),
@@ -20,15 +21,25 @@ const ctx = {
 }
 
 const it = testEffect(
-  Layer.mergeAll(Question.defaultLayer, CrossSpawnSpawner.defaultLayer, Truncate.defaultLayer, Agent.defaultLayer),
+  Layer.mergeAll(
+    Question.layer.pipe(Layer.provideMerge(Bus.layer)),
+    CrossSpawnSpawner.defaultLayer,
+    Truncate.defaultLayer,
+    Agent.defaultLayer,
+  ),
 )
 
 const pending = Effect.fn("QuestionToolTest.pending")(function* (question: Question.Interface) {
+  const bus = yield* Bus.Service
+  const asked = yield* Queue.unbounded<void>()
+  const off = yield* bus.subscribeCallback(Question.Event.Asked, () => Queue.offerUnsafe(asked, undefined))
+  yield* Effect.addFinalizer(() => Effect.sync(off))
+
   for (;;) {
     const items = yield* question.list()
     const item = items[0]
     if (item) return item
-    yield* Effect.sleep("10 millis")
+    yield* Queue.take(asked).pipe(Effect.timeout("2 seconds"))
   }
 })