Compare commits

...

3 Commits

Author SHA1 Message Date
Simon Klee cff1c6f333 cli: recover pending questions from plan_exit tool
The question recovery path only matched the "question" tool, so
plan_exit prompts would get stuck when the question.asked event
was missed during reconnection. Extend recovery to match any
tool-owned question request and drop the tool-name guard in
syncQuestion so completions clear the footer correctly.
2026-05-18 15:23:34 +02:00
Simon Klee f7493d41cb tui: dismiss question when tool part ends
Questions remained visible after the underlying tool call
completed or errored because nothing removed them from the
store. Clear each question once its associated tool part
reaches a terminal state, and fix recovery to preserve
correct blocker priority across all active questions.
2026-05-18 10:09:06 +02:00
Simon Klee f2f8efa411 cli: fix question recovery matching wrong session
The recovery logic matched questions by checking if the list was
non-empty, which caused it to pick up stale questions from earlier
turns. When a re-ask fired, the wrong question could resolve the
blocker, leaving the real question undelivered and the process stuck.

Match questions by messageID and callID from the originating tool part
so only the correct question unblocks the prompt.

Fixes #27503
2026-05-18 10:09:06 +02:00
5 changed files with 215 additions and 33 deletions
@@ -360,14 +360,10 @@ function syncPermission(data: SessionData, part: ToolPart): FooterOutput | undef
}
}
// Question tool replies can complete without a matching question.replied event.
// When that happens, drop the recovered pending request tied to this tool call so
// the footer can return to the next blocker or to the prompt.
// Tool-owned question requests can complete without a matching question.replied
// event. When that happens, drop the recovered pending request tied to this tool
// call so the footer can return to the next blocker or to the prompt.
function syncQuestion(data: SessionData, part: ToolPart): FooterOutput | undefined {
if (part.tool !== "question") {
return undefined
}
if (part.state.status !== "completed" && part.state.status !== "error") {
return undefined
}
@@ -15,7 +15,7 @@
// The tick counter prevents stale idle events from resolving the wrong turn.
// We also re-check live session status before resolving an idle event so a
// delayed idle from an older turn cannot complete a newer busy turn.
import type { Event, GlobalEvent, OpencodeClient } from "@opencode-ai/sdk/v2"
import type { Event, GlobalEvent, OpencodeClient, ToolPart } from "@opencode-ai/sdk/v2"
import { Context, Deferred, Effect, Exit, Layer, Scope, Stream } from "effect"
import { makeRuntime } from "@/effect/run-service"
import {
@@ -505,7 +505,10 @@ function createLayer(input: StreamInput) {
state.footerView = current
}
const recoverQuestion = Effect.fn("RunStreamTransport.recoverQuestion")(function* (partID: string) {
const recoverQuestion = Effect.fn("RunStreamTransport.recoverQuestion")(function* (part: ToolPart) {
const partID = part.id
const matches = (request: SessionData["questions"][number]) =>
request.tool?.messageID === part.messageID && request.tool?.callID === part.callID
if (recovering.has(partID)) {
return
}
@@ -513,7 +516,7 @@ function createLayer(input: StreamInput) {
recovering.add(partID)
try {
while (!closed && !abort.signal.aborted && !input.footer.isClosed) {
if (state.data.questions.length > 0 || !state.data.tools.has(partID)) {
if (state.data.questions.some(matches) || !state.data.tools.has(partID)) {
return
}
@@ -521,11 +524,14 @@ function createLayer(input: StreamInput) {
Effect.map((item) => (item.data ?? []).filter((request) => request.sessionID === input.sessionID)),
Effect.orElseSucceed(() => []),
)
if (state.data.questions.length > 0 || !state.data.tools.has(partID)) {
if (state.data.questions.some(matches) || !state.data.tools.has(partID)) {
return
}
if (questions.length > 0) {
const matching = questions.filter(matches)
if (matching.length > 0) {
const active = new Set(questions.map((request) => request.id))
state.data.questions = state.data.questions.filter((request) => active.has(request.id))
bootstrapSessionData({
data: state.data,
messages: [],
@@ -535,9 +541,11 @@ function createLayer(input: StreamInput) {
for (const request of questions) {
seedBlocker(request.id)
}
const priority = Math.min(0, ...state.blockers.values()) - 1
for (const request of matching) state.blockers.set(request.id, priority)
input.trace?.write("question.recover", {
sessionID: input.sessionID,
requests: questions.map((request) => request.id),
requests: matching.map((request) => request.id),
})
syncFooter([])
return
@@ -783,11 +791,10 @@ function createLayer(input: StreamInput) {
event.type === "message.part.updated" &&
event.properties.part.sessionID === input.sessionID &&
event.properties.part.type === "tool" &&
event.properties.part.tool === "question" &&
event.properties.part.state.status === "running" &&
state.data.questions.length === 0
(event.properties.part.tool === "question" || event.properties.part.tool === "plan_exit") &&
event.properties.part.state.status === "running"
) {
yield* recoverQuestion(event.properties.part.id).pipe(
yield* recoverQuestion(event.properties.part).pipe(
Effect.forkIn(scope, { startImmediately: true }),
Effect.asVoid,
)
@@ -34,6 +34,14 @@ import path from "path"
import { useKV } from "./kv"
import { aggregateFailures } from "./aggregate-failures"
export function questionToolRequestIndex(requests: readonly QuestionRequest[] | undefined, part: Part) {
if (part.type !== "tool") return -1
if (part.state.status !== "completed" && part.state.status !== "error") return -1
return requests?.findIndex(
(request) => request.tool?.messageID === part.messageID && request.tool?.callID === part.callID,
) ?? -1
}
export const { use: useSync, provider: SyncProvider } = createSimpleContext({
name: "Sync",
init: () => {
@@ -304,23 +312,32 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
break
}
case "message.part.updated": {
const parts = store.part[event.properties.part.messageID]
const part = event.properties.part
const parts = store.part[part.messageID]
if (!parts) {
setStore("part", event.properties.part.messageID, [event.properties.part])
break
setStore("part", part.messageID, [part])
}
const result = Binary.search(parts, event.properties.part.id, (p) => p.id)
if (result.found) {
setStore("part", event.properties.part.messageID, result.index, reconcile(event.properties.part))
break
if (parts) {
const result = Binary.search(parts, part.id, (p) => p.id)
if (result.found) setStore("part", part.messageID, result.index, reconcile(part))
if (!result.found)
setStore(
"part",
part.messageID,
produce((draft) => {
draft.splice(result.index, 0, part)
}),
)
}
setStore(
"part",
event.properties.part.messageID,
produce((draft) => {
draft.splice(result.index, 0, event.properties.part)
}),
)
const index = questionToolRequestIndex(store.question[part.sessionID], part)
if (index !== -1)
setStore(
"question",
part.sessionID,
produce((draft) => {
draft.splice(index, 1)
}),
)
break
}
@@ -835,12 +835,17 @@ describe("run stream transport", () => {
callID: "call-question-1",
},
}
const other = {
...request,
id: "question-old",
tool: { messageID: "msg-old", callID: "call-question-old" },
}
const transport = await createSessionTransport({
sdk: sdk({
stream: src.stream,
questions: async () => {
questionCalls += 1
return ok(questionCalls > 1 ? [request] : [])
return ok(questionCalls === 1 ? [other] : [request])
},
promptAsync: async () => {
queueMicrotask(() => {
@@ -885,7 +890,9 @@ describe("run stream transport", () => {
const view = await waitFor(() => {
const item = ui.events.findLast((event) => event.type === "stream.view")
return item?.type === "stream.view" && item.view.type === "question" ? item.view : undefined
return item?.type === "stream.view" && item.view.type === "question" && item.view.request.id === request.id
? item.view
: undefined
})
expect(view).toEqual({
@@ -901,6 +908,7 @@ describe("run stream transport", () => {
},
})
const count = ui.events.length
src.push(
toolUpdated(
completedTool({
@@ -930,6 +938,121 @@ describe("run stream transport", () => {
view: { type: "prompt" },
})
expect(
ui.events.slice(count).findLast(
(event) => event.type === "stream.view" && event.view.type === "question" && event.view.request.id === other.id,
),
).toBeUndefined()
ctrl.abort()
await run
} finally {
src.close()
await transport.close()
}
})
test("recovers pending plan_exit questions from question.list when question.asked is missed", async () => {
const src = eventFeed()
const ui = footer()
let questionCalls = 0
const request = {
id: "question-plan-1",
sessionID: "session-1",
questions: [
{
question: "Plan is complete. Start implementing it now?",
header: "Build Agent",
options: [{ label: "Yes", description: "Switch to build agent and start implementing." }],
multiple: false,
},
],
tool: {
messageID: "msg-plan-1",
callID: "call-plan-exit-1",
},
}
const transport = await createSessionTransport({
sdk: sdk({
stream: src.stream,
questions: async () => {
questionCalls += 1
return ok(questionCalls === 1 ? [] : [request])
},
promptAsync: async () => {
queueMicrotask(() => {
src.push(busy())
src.push(assistant("msg-plan-1"))
src.push(
toolUpdated(
runningTool({
sessionID: "session-1",
messageID: "msg-plan-1",
id: "plan-exit-tool-1",
callID: "call-plan-exit-1",
tool: "plan_exit",
body: {},
}),
),
)
})
return ok(undefined)
},
}),
sessionID: "session-1",
thinking: true,
limits: () => ({}),
footer: ui.api,
})
const ctrl = new AbortController()
try {
const run = transport.runPromptTurn({
agent: undefined,
model: undefined,
variant: undefined,
prompt: { text: "hello", parts: [] },
files: [],
includeFiles: false,
signal: ctrl.signal,
})
expect(
await waitFor(() => {
const item = ui.events.findLast((event) => event.type === "stream.view")
return item?.type === "stream.view" && item.view.type === "question" ? item.view : undefined
}),
).toEqual({
type: "question",
request,
})
src.push(
toolUpdated(
completedTool({
sessionID: "session-1",
messageID: "msg-plan-1",
id: "plan-exit-tool-1",
callID: "call-plan-exit-1",
tool: "plan_exit",
body: {},
output: "User approved switching to build agent.",
metadata: {},
}),
),
)
expect(
await waitFor(() => {
const item = ui.events.findLast((event) => event.type === "stream.view")
return item?.type === "stream.view" && item.view.type === "prompt" ? item : undefined
}),
).toEqual({
type: "stream.view",
view: { type: "prompt" },
})
ctrl.abort()
await run
} finally {
@@ -0,0 +1,39 @@
import { describe, expect, test } from "bun:test"
import type { QuestionRequest, ToolPart } from "@opencode-ai/sdk/v2"
import { questionToolRequestIndex } from "@/cli/cmd/tui/context/sync"
const request = {
id: "question-new",
sessionID: "session-1",
questions: [],
tool: { messageID: "msg-new", callID: "call-new" },
} satisfies QuestionRequest
function part(status: "running" | "completed" | "error", tool = "question", callID = "call-new"): ToolPart {
return {
id: "part-new",
sessionID: "session-1",
messageID: "msg-new",
type: "tool",
callID,
tool,
state:
status === "running"
? { status, input: {}, time: { start: 1 } }
: status === "completed"
? { status, input: {}, output: "", title: "question", metadata: {}, time: { start: 1, end: 2 } }
: { status, input: {}, error: "Tool execution aborted", time: { start: 1, end: 2 } },
}
}
describe("tui sync", () => {
test("matches terminal tool-owned question requests", () => {
const stale = { ...request, id: "question-old", tool: { messageID: "msg-old", callID: "call-old" } }
expect(questionToolRequestIndex([stale, request], part("error"))).toBe(1)
expect(questionToolRequestIndex([stale, request], part("completed"))).toBe(1)
expect(questionToolRequestIndex([stale, request], part("completed", "plan_exit"))).toBe(1)
expect(questionToolRequestIndex([stale, request], part("running"))).toBe(-1)
expect(questionToolRequestIndex([stale, request], part("error", "bash", "call-other"))).toBe(-1)
})
})