From 20133b81937f7f2f4c14a5f95544c43adc7b4c74 Mon Sep 17 00:00:00 2001 From: Aneesh Date: Sun, 30 Aug 2026 15:10:46 +0000 Subject: [PATCH] fix(question): reject pending questions on abort and interrupt --- .../context/global-sync/event-reducer.test.ts | 42 +++++++++ .../src/context/global-sync/event-reducer.ts | 45 ++++++--- packages/core/src/question.ts | 51 +++++++++- packages/core/test/question.test.ts | 61 +++++++++++- packages/core/test/tool-question.test.ts | 1 + packages/opencode/src/question/index.ts | 46 ++++++++-- packages/opencode/src/session/processor.ts | 3 + packages/opencode/src/session/prompt.ts | 4 + .../opencode/test/question/question.test.ts | 92 +++++++++++++++++++ packages/tui/src/context/sync.tsx | 47 +++++++--- 10 files changed, 354 insertions(+), 38 deletions(-) diff --git a/packages/app/src/context/global-sync/event-reducer.test.ts b/packages/app/src/context/global-sync/event-reducer.test.ts index 06d536618e05..5a1f09028462 100644 --- a/packages/app/src/context/global-sync/event-reducer.test.ts +++ b/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([]) + }) }) diff --git a/packages/app/src/context/global-sync/event-reducer.ts b/packages/app/src/context/global-sync/event-reducer.ts index b84d5201cf55..442c5df98dfa 100644 --- a/packages/app/src/context/global-sync/event-reducer.ts +++ b/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": { diff --git a/packages/core/src/question.ts b/packages/core/src/question.ts index 79e0ea5e0384..dab6e35d6670 100644 --- a/packages/core/src/question.ts +++ b/packages/core/src/question.ts @@ -57,6 +57,7 @@ export interface Interface { readonly ask: (input: AskInput) => Effect.Effect, RejectedError> readonly reply: (input: ReplyInput) => Effect.Effect readonly reject: (requestID: ID) => Effect.Effect + readonly cancel: (sessionID: SessionSchema.ID) => Effect.Effect readonly list: () => Effect.Effect> } @@ -79,9 +80,19 @@ const layer = Layer.effect( const pending = new Map() 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 }) }), ) diff --git a/packages/core/test/question.test.ts b/packages/core/test/question.test.ts index 03d61a956507..cdd9d8c640eb 100644 --- a/packages/core/test/question.test.ts +++ b/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() 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) + }), + ) }) diff --git a/packages/core/test/tool-question.test.ts b/packages/core/test/tool-question.test.ts index d76e15388252..70e4549fa3b4 100644 --- a/packages/core/test/tool-question.test.ts +++ b/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"), }), ) diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index 8afc141072d5..6c8aa7574579 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -56,6 +56,7 @@ export interface Interface { answers: ReadonlyArray }) => Effect.Effect readonly reject: (requestID: QuestionID) => Effect.Effect + readonly cancel: (sessionID: SessionID) => Effect.Effect readonly list: () => Effect.Effect> } @@ -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 }) }), ) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 20aa8a8404d8..623a1d561edc 100644 --- a/packages/opencode/src/session/processor.ts +++ b/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, ], }) diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 0f85d44f209b..d73e4f545ac2 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -31,6 +31,7 @@ import { ConfigMarkdown } from "@/config/markdown" import { SessionSummary } from "./summary" import { NamedError } from "@opencode-ai/core/util/error" import { SessionProcessor } from "./processor" +import { Question } from "@/question" import { Tool } from "@/tool/tool" import { Permission } from "@/permission" import { SessionStatus } from "./status" @@ -140,6 +141,7 @@ const layer = Layer.effect( const events = yield* EventV2Bridge.Service const flags = yield* RuntimeFlags.Service const database = yield* Database.Service + const question = yield* Question.Service const { db } = database const ops = Effect.fn("SessionPrompt.ops")(function* () { return { @@ -152,6 +154,7 @@ const layer = Layer.effect( const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) { yield* Effect.logInfo("cancel", { "session.id": sessionID }) yield* state.cancel(sessionID) + yield* question.cancel(sessionID) }) const resolvePromptParts = Effect.fn("SessionPrompt.resolvePromptParts")(function* (template: string) { @@ -1625,6 +1628,7 @@ export const node = LayerNode.make({ EventV2Bridge.node, RuntimeFlags.node, Database.node, + Question.node, ], }) diff --git a/packages/opencode/test/question/question.test.ts b/packages/opencode/test/question/question.test.ts index 9adb1e0c9ff7..afc93f43fe6c 100644 --- a/packages/opencode/test/question/question.test.ts +++ b/packages/opencode/test/question/question.test.ts @@ -457,3 +457,95 @@ lifecycle.live("pending question rejects on instance reload", () => if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(Question.RejectedError) }), ) + +it.instance( + "ask - publishes question.rejected and removes from pending on fiber interrupt", + () => + Effect.gen(function* () { + const events = yield* EventV2Bridge.Service + const rejectedQueue = yield* Queue.unbounded() + const off = yield* events.listen((event) => { + if (event.type === Question.Event.Rejected.type) Queue.offerUnsafe(rejectedQueue, event.data as any) + return Effect.void + }) + yield* Effect.addFinalizer(() => off) + + const fiber = yield* askEffect({ + sessionID: SessionID.make("ses_interrupt"), + questions: [ + { + question: "Interrupt me?", + header: "Interrupt", + options: [{ label: "Option 1", description: "First" }], + }, + ], + }).pipe(Effect.forkScoped) + + expect(yield* waitForPending(1)).toHaveLength(1) + + yield* Fiber.interrupt(fiber) + + const rejectedEvent = yield* Queue.take(rejectedQueue).pipe(Effect.timeout("2 seconds")) + expect(rejectedEvent.sessionID).toBe(SessionID.make("ses_interrupt")) + + const pending = yield* listEffect + expect(pending.length).toBe(0) + }), + { git: true }, +) + +it.instance( + "cancel - rejects pending questions for specific session and publishes question.rejected", + () => + Effect.gen(function* () { + const question = yield* Question.Service + const events = yield* EventV2Bridge.Service + const rejectedQueue = yield* Queue.unbounded() + const off = yield* events.listen((event) => { + if (event.type === Question.Event.Rejected.type) Queue.offerUnsafe(rejectedQueue, event.data as any) + return Effect.void + }) + yield* Effect.addFinalizer(() => off) + + const fiber1 = yield* askEffect({ + sessionID: SessionID.make("ses_cancel_target"), + questions: [ + { + question: "Target session question?", + header: "Target", + options: [{ label: "A", description: "A" }], + }, + ], + }).pipe(Effect.forkScoped) + + const fiber2 = yield* askEffect({ + sessionID: SessionID.make("ses_cancel_other"), + questions: [ + { + question: "Other session question?", + header: "Other", + options: [{ label: "B", description: "B" }], + }, + ], + }).pipe(Effect.forkScoped) + + expect(yield* waitForPending(2)).toHaveLength(2) + + yield* question.cancel(SessionID.make("ses_cancel_target")) + + const rejectedEvent = yield* Queue.take(rejectedQueue).pipe(Effect.timeout("2 seconds")) + expect(rejectedEvent.sessionID).toBe(SessionID.make("ses_cancel_target")) + + const exit1 = yield* Fiber.await(fiber1) + expect(Exit.isFailure(exit1)).toBe(true) + if (Exit.isFailure(exit1)) expect(Cause.squash(exit1.cause)).toBeInstanceOf(Question.RejectedError) + + const remaining = yield* listEffect + expect(remaining.length).toBe(1) + expect(remaining[0].sessionID).toBe(SessionID.make("ses_cancel_other")) + + yield* rejectAll + expect((yield* Fiber.await(fiber2))._tag).toBe("Failure") + }), + { git: true }, +) diff --git a/packages/tui/src/context/sync.tsx b/packages/tui/src/context/sync.tsx index 71e050d11e68..4feb802d4260 100644 --- a/packages/tui/src/context/sync.tsx +++ b/packages/tui/src/context/sync.tsx @@ -378,20 +378,43 @@ export const { const parts = store.part[event.properties.part.messageID] if (!parts) { setStore("part", event.properties.part.messageID, [event.properties.part]) - break + } else { + const result = search(parts, event.properties.part.id, (part) => part.id) + if (result.found) { + setStore("part", event.properties.part.messageID, result.index, reconcile(event.properties.part)) + } else { + setStore( + "part", + event.properties.part.messageID, + produce((draft) => { + draft.splice(result.index, 0, event.properties.part) + }), + ) + } } - const result = search(parts, event.properties.part.id, (part) => part.id) - if (result.found) { - setStore("part", event.properties.part.messageID, result.index, reconcile(event.properties.part)) - break + if ( + event.properties.part.type === "tool" && + (event.properties.part.state.status === "error" || event.properties.part.state.status === "completed") && + store.question[event.properties.part.sessionID] + ) { + const questions = store.question[event.properties.part.sessionID]! + const toolCallID = event.properties.part.callID + const toolMessageID = event.properties.part.messageID + const indices = questions + .map((q, i) => (q.tool?.callID === toolCallID || q.tool?.messageID === toolMessageID ? i : -1)) + .filter((i) => i !== -1) + if (indices.length > 0) { + setStore( + "question", + event.properties.part.sessionID, + produce((draft) => { + for (let i = indices.length - 1; i >= 0; i--) { + draft.splice(indices[i], 1) + } + }), + ) + } } - setStore( - "part", - event.properties.part.messageID, - produce((draft) => { - draft.splice(result.index, 0, event.properties.part) - }), - ) break }