fix(question): reject pending questions on abort and interrupt

packages/app/src/context/global-sync/event-reducer.test.ts

@@ -613,4 +613,46 @@ describe("applyDirectoryEvent", () => {
     expect(pushes).toEqual(["/tmp"])
     expect(lspLoads).toBe(1)
   })
+
+  test("cleans up pending question when its tool part transitions to error or completed", () => {
+    const sessionID = "ses_test"
+    const messageID = "msg_1"
+    const callID = "call_question_1"
+    const questionWithTool = {
+      ...questionRequest("q_1", sessionID),
+      tool: { messageID, callID },
+    }
+    const [store, setStore] = createStore(
+      baseState({
+        question: { [sessionID]: [questionWithTool] },
+      }),
+    )
+
+    const toolPart = {
+      id: "part_tool_1",
+      sessionID,
+      messageID,
+      type: "tool",
+      tool: "question",
+      callID,
+      state: {
+        status: "error",
+        input: {},
+        error: "Tool execution aborted",
+        metadata: { interrupted: true },
+        time: { start: 1, end: 2 },
+      },
+    } as Part
+
+    applyDirectoryEvent({
+      event: { type: "message.part.updated", properties: { part: toolPart } },
+      store,
+      setStore,
+      push() {},
+      directory: "/tmp",
+      loadLsp() {},
+    })
+
+    expect(store.question[sessionID]).toEqual([])
+  })
 })

packages/app/src/context/global-sync/event-reducer.ts

@@ -321,20 +321,41 @@ export function applyDirectoryEvent(input: {
       const parts = input.store.part[part.messageID]
       if (!parts) {
         input.setStore("part", part.messageID, [part])
-        break
+      } else {
+        const result = Binary.search(parts, part.id, (item) => item.id)
+        if (result.found) {
+          input.setStore("part", part.messageID, result.index, reconcile(part))
+        } else {
+          input.setStore(
+            "part",
+            part.messageID,
+            produce((draft) => {
+              draft.splice(result.index, 0, part)
+            }),
+          )
+        }
       }
-      const result = Binary.search(parts, part.id, (item) => item.id)
-      if (result.found) {
-        input.setStore("part", part.messageID, result.index, reconcile(part))
-        break
+      if (
+        part.type === "tool" &&
+        (part.state.status === "error" || part.state.status === "completed") &&
+        input.store.question[part.sessionID]
+      ) {
+        const questions = input.store.question[part.sessionID]!
+        const indices = questions
+          .map((q, i) => (q.tool?.callID === part.callID || q.tool?.messageID === part.messageID ? i : -1))
+          .filter((i) => i !== -1)
+        if (indices.length > 0) {
+          input.setStore(
+            "question",
+            part.sessionID,
+            produce((draft) => {
+              for (let i = indices.length - 1; i >= 0; i--) {
+                draft.splice(indices[i], 1)
+              }
+            }),
+          )
+        }
       }
-      input.setStore(
-        "part",
-        part.messageID,
-        produce((draft) => {
-          draft.splice(result.index, 0, part)
-        }),
-      )
       break
     }
     case "message.part.removed": {

packages/core/src/question.ts

@@ -57,6 +57,7 @@ export interface Interface {
   readonly ask: (input: AskInput) => Effect.Effect<ReadonlyArray<Answer>, RejectedError>
   readonly reply: (input: ReplyInput) => Effect.Effect<void, NotFoundError>
   readonly reject: (requestID: ID) => Effect.Effect<void, NotFoundError>
+  readonly cancel: (sessionID: SessionSchema.ID) => Effect.Effect<void>
   readonly list: () => Effect.Effect<ReadonlyArray<Request>>
 }
 
@@ -79,9 +80,19 @@ const layer = Layer.effect(
     const pending = new Map<ID, Pending>()
 
     yield* Effect.addFinalizer(() =>
-      Effect.forEach(pending.values(), (item) => Deferred.fail(item.deferred, new RejectedError()), {
-        discard: true,
-      }).pipe(
+      Effect.forEach(
+        pending.values(),
+        (item) =>
+          events
+            .publish(Event.Rejected, {
+              sessionID: item.request.sessionID,
+              requestID: item.request.id,
+            })
+            .pipe(Effect.andThen(Deferred.fail(item.deferred, new RejectedError()).pipe(Effect.ignore))),
+        {
+          discard: true,
+        },
+      ).pipe(
         Effect.ensuring(
           Effect.sync(() => {
             pending.clear()
@@ -98,7 +109,21 @@ const layer = Layer.effect(
           const request: Request = { id, ...input }
           pending.set(id, { request, deferred })
           return yield* events.publish(Event.Asked, request).pipe(
-            Effect.andThen(restore(Deferred.await(deferred))),
+            Effect.andThen(
+              restore(Deferred.await(deferred)).pipe(
+                Effect.onInterrupt(() =>
+                  Effect.gen(function* () {
+                    if (!pending.has(id)) return
+                    pending.delete(id)
+                    yield* events.publish(Event.Rejected, {
+                      sessionID: input.sessionID,
+                      requestID: id,
+                    })
+                    yield* Deferred.fail(deferred, new RejectedError()).pipe(Effect.ignore)
+                  }),
+                ),
+              ),
+            ),
             Effect.ensuring(
               Effect.sync(() => {
                 pending.delete(id)
@@ -140,11 +165,27 @@ const layer = Layer.effect(
       ),
     )
 
+    const cancel = Effect.fn("QuestionV2.cancel")((sessionID: SessionSchema.ID) =>
+      Effect.uninterruptible(
+        Effect.gen(function* () {
+          for (const [id, item] of pending.entries()) {
+            if (item.request.sessionID !== sessionID) continue
+            pending.delete(id)
+            yield* events.publish(Event.Rejected, {
+              sessionID: item.request.sessionID,
+              requestID: item.request.id,
+            })
+            yield* Deferred.fail(item.deferred, new RejectedError()).pipe(Effect.ignore)
+          }
+        }),
+      ),
+    )
+
     const list = Effect.fn("QuestionV2.list")(function* () {
       return Array.from(pending.values(), (item) => item.request)
     })
 
-    return Service.of({ ask, reply, reject, list })
+    return Service.of({ ask, reply, reject, cancel, list })
   }),
 )
 

packages/core/test/question.test.ts

@@ -20,16 +20,18 @@ const question: QuestionV2.Info = {
 const waitForAsk = Effect.fn("QuestionV2Test.waitForAsk")(function* (
   service: QuestionV2.Interface,
   input: QuestionV2.AskInput,
+  scope?: Scope.Scope,
 ) {
   const events = yield* EventV2.Service
+  const testScope = scope ?? (yield* Scope.Scope)
   const asked = yield* Deferred.make<QuestionV2.Request>()
   const unsubscribe = yield* events.listen((event) =>
-    event.type === QuestionV2.Event.Asked.type
+    event.type === QuestionV2.Event.Asked.type && (event.data as QuestionV2.Request).sessionID === input.sessionID
       ? Deferred.succeed(asked, event.data as QuestionV2.Request).pipe(Effect.asVoid)
       : Effect.void,
   )
   yield* Effect.addFinalizer(() => unsubscribe)
-  const fiber = yield* service.ask(input).pipe(Effect.forkScoped)
+  const fiber = yield* service.ask(input).pipe(Effect.forkIn(testScope))
   return { fiber, request: yield* Deferred.await(asked) }
 })
 
@@ -111,4 +113,59 @@ describe("QuestionV2", () => {
       yield* Scope.close(secondScope, Exit.void)
     }),
   )
+
+  it.effect("publishes rejection and cleans up pending on fiber interrupt", () =>
+    Effect.gen(function* () {
+      const service = yield* QuestionV2.Service
+      const events = yield* EventV2.Service
+      const published: EventV2.Payload[] = []
+      const unsubscribe = yield* events.listen((event) =>
+        Effect.sync(() => {
+          if (event.type === QuestionV2.Event.Rejected.type) published.push(event)
+        }),
+      )
+      yield* Effect.addFinalizer(() => unsubscribe)
+      const { fiber, request } = yield* waitForAsk(service, { sessionID, questions: [question] })
+
+      yield* Fiber.interrupt(fiber)
+      expect(yield* service.list()).toEqual([])
+      expect(published.map((event) => event.data)).toEqual([{ sessionID, requestID: request.id }])
+    }),
+  )
+
+  it.effect("cancels pending requests for a specific session and publishes rejection", () =>
+    Effect.gen(function* () {
+      const service = yield* QuestionV2.Service
+      const events = yield* EventV2.Service
+      const published: EventV2.Payload[] = []
+      const unsubscribe = yield* events.listen((event) =>
+        Effect.sync(() => {
+          if (event.type === QuestionV2.Event.Rejected.type) published.push(event)
+        }),
+      )
+      yield* Effect.addFinalizer(() => unsubscribe)
+
+      const scope = yield* Scope.Scope
+      const targetSession = SessionV2.ID.make("ses_target")
+      const otherSession = SessionV2.ID.make("ses_other")
+
+      const target = yield* waitForAsk(service, { sessionID: targetSession, questions: [question] }, scope)
+      const other = yield* waitForAsk(service, { sessionID: otherSession, questions: [question] }, scope)
+
+      expect(yield* service.list()).toHaveLength(2)
+
+      yield* service.cancel(targetSession)
+
+      const exit = yield* Fiber.await(target.fiber)
+      expect(Exit.isFailure(exit)).toBe(true)
+      if (Exit.isFailure(exit)) expect(exit.cause.toString()).toContain("QuestionV2.RejectedError")
+
+      expect(published.map((event) => event.data)).toEqual([{ sessionID: targetSession, requestID: target.request.id }])
+
+      const remaining = yield* service.list()
+      expect(remaining).toEqual([other.request])
+
+      yield* service.reject(other.request.id)
+    }),
+  )
 })

packages/core/test/tool-question.test.ts

@@ -40,6 +40,7 @@ const question = Layer.succeed(
       }).pipe(Effect.andThen(reject ? Effect.fail(new QuestionV2.RejectedError()) : Effect.succeed([["Build"], []]))),
     reply: () => Effect.die("unused"),
     reject: () => Effect.die("unused"),
+    cancel: () => Effect.die("unused"),
     list: () => Effect.die("unused"),
   }),
 )

packages/opencode/src/question/index.ts

@@ -56,6 +56,7 @@ export interface Interface {
     answers: ReadonlyArray<Answer>
   }) => Effect.Effect<void, NotFoundError>
   readonly reject: (requestID: QuestionID) => Effect.Effect<void, NotFoundError>
+  readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
   readonly list: () => Effect.Effect<ReadonlyArray<Request>>
 }
 
@@ -74,7 +75,11 @@ const layer = Layer.effect(
         yield* Effect.addFinalizer(() =>
           Effect.gen(function* () {
             for (const item of state.pending.values()) {
-              yield* Deferred.fail(item.deferred, new RejectedError())
+              yield* events.publish(Event.Rejected, {
+                sessionID: item.info.sessionID,
+                requestID: item.info.id,
+              })
+              yield* Deferred.fail(item.deferred, new RejectedError()).pipe(Effect.ignore)
             }
             state.pending.clear()
           }),
@@ -103,11 +108,24 @@ const layer = Layer.effect(
       pending.set(id, { info, deferred })
       yield* events.publish(Event.Asked, info)
 
-      return yield* Effect.ensuring(
-        Deferred.await(deferred),
-        Effect.sync(() => {
-          pending.delete(id)
-        }),
+      return yield* Deferred.await(deferred).pipe(
+        Effect.onInterrupt(() =>
+          Effect.gen(function* () {
+            if (!pending.has(id)) return
+            pending.delete(id)
+            yield* Effect.logInfo("rejected on interrupt", { requestID: id, sessionID: input.sessionID })
+            yield* events.publish(Event.Rejected, {
+              sessionID: input.sessionID,
+              requestID: id,
+            })
+            yield* Deferred.fail(deferred, new RejectedError()).pipe(Effect.ignore)
+          }),
+        ),
+        Effect.ensuring(
+          Effect.sync(() => {
+            pending.delete(id)
+          }),
+        ),
       )
     })
 
@@ -147,12 +165,26 @@ const layer = Layer.effect(
       yield* Deferred.fail(existing.deferred, new RejectedError())
     })
 
+    const cancel = Effect.fn("Question.cancel")(function* (sessionID: SessionID) {
+      const pending = (yield* InstanceState.get(state)).pending
+      for (const [id, item] of pending.entries()) {
+        if (item.info.sessionID !== sessionID) continue
+        pending.delete(id)
+        yield* Effect.logInfo("rejected on cancel", { sessionID, requestID: id })
+        yield* events.publish(Event.Rejected, {
+          sessionID: item.info.sessionID,
+          requestID: item.info.id,
+        })
+        yield* Deferred.fail(item.deferred, new RejectedError()).pipe(Effect.ignore)
+      }
+    })
+
     const list = Effect.fn("Question.list")(function* () {
       const pending = (yield* InstanceState.get(state)).pending
       return Array.from(pending.values(), (x) => x.info)
     })
 
-    return Service.of({ ask, reply, reject, list })
+    return Service.of({ ask, reply, reject, cancel, list })
   }),
 )
 

packages/opencode/src/session/processor.ts

@@ -94,6 +94,7 @@ const layer = Layer.effect(
     const image = yield* Image.Service
     const events = yield* EventV2Bridge.Service
     const database = yield* Database.Service
+    const question = yield* Question.Service
 
     const create = Effect.fn("SessionProcessor.create")(function* (input: Input) {
       // Pre-capture snapshot before the LLM stream starts. The AI SDK
@@ -591,6 +592,7 @@ const layer = Layer.effect(
             },
           })
         }
+        yield* question.cancel(ctx.sessionID)
         ctx.toolcalls = {}
         ctx.assistantMessage.time.completed = Date.now()
         yield* session.updateMessage(ctx.assistantMessage)
@@ -712,6 +714,7 @@ export const node = LayerNode.make({
     Image.node,
     EventV2Bridge.node,
     Database.node,
+    Question.node,
   ],
 })