mirror of
https://github.com/anomalyco/opencode.git
synced 2026-07-28 14:11:49 -04:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| cff1c6f333 | |||
| f7493d41cb | |||
| f2f8efa411 |
@@ -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)
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user