diff --git a/packages/opencode/src/cli/cmd/run/session-data.ts b/packages/opencode/src/cli/cmd/run/session-data.ts index d3d61dd15e..94b2baba36 100644 --- a/packages/opencode/src/cli/cmd/run/session-data.ts +++ b/packages/opencode/src/cli/cmd/run/session-data.ts @@ -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 } diff --git a/packages/opencode/src/cli/cmd/run/stream.transport.ts b/packages/opencode/src/cli/cmd/run/stream.transport.ts index 22240ebf56..504ff70faa 100644 --- a/packages/opencode/src/cli/cmd/run/stream.transport.ts +++ b/packages/opencode/src/cli/cmd/run/stream.transport.ts @@ -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, ) diff --git a/packages/opencode/src/cli/cmd/tui/context/sync.tsx b/packages/opencode/src/cli/cmd/tui/context/sync.tsx index 9f8a384f77..fc0103946c 100644 --- a/packages/opencode/src/cli/cmd/tui/context/sync.tsx +++ b/packages/opencode/src/cli/cmd/tui/context/sync.tsx @@ -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 } diff --git a/packages/opencode/test/cli/run/stream.transport.test.ts b/packages/opencode/test/cli/run/stream.transport.test.ts index 3358ae774d..5e3af37be0 100644 --- a/packages/opencode/test/cli/run/stream.transport.test.ts +++ b/packages/opencode/test/cli/run/stream.transport.test.ts @@ -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 { diff --git a/packages/opencode/test/cli/tui/sync.test.ts b/packages/opencode/test/cli/tui/sync.test.ts new file mode 100644 index 0000000000..06ad3efc21 --- /dev/null +++ b/packages/opencode/test/cli/tui/sync.test.ts @@ -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) + }) +})