From 7c76bd6f2b832b795ecab9e36e4e43cf06308212 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=AF=E5=9F=BA=E9=AD=81?= <1412414664@qq.com> Date: Tue, 7 Jul 2026 16:43:44 +0800 Subject: [PATCH] fix(tui): settle stale replay work --- .../opencode/src/cli/cmd/run/session-data.ts | 21 +++++ .../src/cli/cmd/run/session-replay.test.ts | 94 +++++++++++++++++++ .../src/cli/cmd/run/session-replay.ts | 13 ++- .../src/cli/cmd/run/stream.transport.ts | 10 +- 4 files changed, 136 insertions(+), 2 deletions(-) create mode 100644 packages/opencode/src/cli/cmd/run/session-replay.test.ts diff --git a/packages/opencode/src/cli/cmd/run/session-data.ts b/packages/opencode/src/cli/cmd/run/session-data.ts index 03951ec4c9e1..485138084864 100644 --- a/packages/opencode/src/cli/cmd/run/session-data.ts +++ b/packages/opencode/src/cli/cmd/run/session-data.ts @@ -54,6 +54,7 @@ type SessionCommit = StreamCommit // // - ids: parts and error keys we've already committed (dedup guard) // - tools: tool parts we've emitted a "start" for but not yet completed +// - toolPart: latest full tool payload for active tool parts // - call: tool call inputs, keyed by msg:call, for enriching permission views // - role: message ID → "assistant" | "user", learned from message.updated // - msg: part ID → message ID @@ -74,6 +75,7 @@ export type SessionData = { announced: boolean ids: Set tools: Set + toolPart: Map call: Map shell: Map permissions: PermissionRequest[] @@ -112,6 +114,7 @@ export function createSessionData( announced: false, ids: new Set(), tools: new Set(), + toolPart: new Map(), call: new Map(), shell: new Map(), permissions: [], @@ -758,6 +761,21 @@ export function flushInterrupted(data: SessionData, commits: SessionCommit[]) { data.ids.add(partID) drop(data, partID) } + + for (const partID of data.tools) { + const part = data.toolPart.get(partID) + data.tools.delete(partID) + data.toolPart.delete(partID) + if (!part || data.ids.has(partID)) { + continue + } + + data.ids.add(partID) + commits.push({ + ...failTool(part, "Tool execution interrupted"), + interrupted: true, + }) + } } // The main reducer. Takes one SDK event and returns scrollback commits and @@ -930,6 +948,7 @@ export function reduceSessionData(input: SessionDataInput): SessionDataOutput { } if (part.state.status === "running") { + data.toolPart.set(part.id, part) if (data.ids.has(part.id)) { return out(data, commits, view) } @@ -946,6 +965,7 @@ export function reduceSessionData(input: SessionDataInput): SessionDataOutput { const seen = data.tools.has(part.id) const mode = toolView(part.tool) data.tools.delete(part.id) + data.toolPart.delete(part.id) if (data.ids.has(part.id)) { return out(data, commits, view) } @@ -982,6 +1002,7 @@ export function reduceSessionData(input: SessionDataInput): SessionDataOutput { if (part.state.status === "error") { const seen = data.tools.has(part.id) data.tools.delete(part.id) + data.toolPart.delete(part.id) if (data.ids.has(part.id)) { return out(data, commits, view) } diff --git a/packages/opencode/src/cli/cmd/run/session-replay.test.ts b/packages/opencode/src/cli/cmd/run/session-replay.test.ts new file mode 100644 index 000000000000..1f975976b22c --- /dev/null +++ b/packages/opencode/src/cli/cmd/run/session-replay.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, test } from "bun:test" +import { replaySession } from "./session-replay" +import type { SessionMessages } from "./session.shared" + +const sessionID = "ses_replay_interrupted" +const userID = "msg_user" +const assistantID = "msg_assistant" + +const messages: SessionMessages = [ + { + info: { + id: userID, + sessionID, + role: "user", + time: { created: 1 }, + agent: "build", + model: { providerID: "provider", modelID: "model" }, + }, + parts: [{ id: "prt_user", sessionID, messageID: userID, type: "text", text: "check health" }], + }, + { + info: { + id: assistantID, + sessionID, + role: "assistant", + time: { created: 2 }, + agent: "build", + parentID: userID, + modelID: "model", + providerID: "provider", + mode: "build", + path: { cwd: "/project", root: "/project" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + }, + parts: [ + { + id: "prt_tool", + sessionID, + messageID: assistantID, + type: "tool", + callID: "call_health", + tool: "bash", + state: { + status: "running", + input: { command: "bun ops prod health" }, + title: "bun ops prod health", + time: { start: 3 }, + }, + }, + ], + }, +] + +describe("replaySession", () => { + test("keeps replayed active work running by default", () => { + const replay = replaySession({ + messages, + permissions: [], + questions: [], + thinking: true, + limits: {}, + }) + + expect(replay.patch?.phase).toBe("running") + expect(replay.commits).not.toContainEqual( + expect.objectContaining({ + partID: "prt_tool", + interrupted: true, + }), + ) + }) + + test("settles replayed active work when the session is no longer running", () => { + const replay = replaySession({ + messages, + permissions: [], + questions: [], + thinking: true, + limits: {}, + settleActive: true, + }) + + expect(replay.patch?.phase).toBe("idle") + expect(replay.commits).toContainEqual( + expect.objectContaining({ + kind: "tool", + phase: "final", + partID: "prt_tool", + interrupted: true, + }), + ) + }) +}) diff --git a/packages/opencode/src/cli/cmd/run/session-replay.ts b/packages/opencode/src/cli/cmd/run/session-replay.ts index 69a24f2719cb..378ce5cfc8dd 100644 --- a/packages/opencode/src/cli/cmd/run/session-replay.ts +++ b/packages/opencode/src/cli/cmd/run/session-replay.ts @@ -1,5 +1,11 @@ import type { Event, PermissionRequest, QuestionRequest } from "@opencode-ai/sdk/v2" -import { bootstrapSessionData, createSessionData, reduceSessionData, type SessionData } from "./session-data" +import { + bootstrapSessionData, + createSessionData, + flushInterrupted, + reduceSessionData, + type SessionData, +} from "./session-data" import { messagePrompt, type SessionMessages } from "./session.shared" import { messageTurnSummaryCommit } from "./turn-summary" import type { FooterPatch, LocalReplayRow, RunProvider, StreamCommit } from "./types" @@ -11,6 +17,7 @@ type ReplayInput = { thinking: boolean limits: Record providers?: RunProvider[] + settleActive?: boolean } type ReplayConfig = { @@ -253,6 +260,10 @@ export function replaySession(input: ReplayInput): SessionReplay { patch = mergePatch(patch, next.patch) } + if (input.settleActive) { + flushInterrupted(data, commits) + } + return { data, commits, diff --git a/packages/opencode/src/cli/cmd/run/stream.transport.ts b/packages/opencode/src/cli/cmd/run/stream.transport.ts index e4817f514dca..df273ae652a0 100644 --- a/packages/opencode/src/cli/cmd/run/stream.transport.ts +++ b/packages/opencode/src/cli/cmd/run/stream.transport.ts @@ -674,7 +674,7 @@ function createLayer(input: StreamInput) { }) const bootstrap = Effect.fn("RunStreamTransport.bootstrap")(function* () { - const [messagesList, children, permissions, questions] = yield* Effect.all( + const [messagesList, children, permissions, questions, statuses] = yield* Effect.all( [ messages( input.sessionID, @@ -700,6 +700,10 @@ function createLayer(input: StreamInput) { Effect.map((item) => item.data ?? []), Effect.orElseSucceed(() => []), ), + Effect.promise(() => input.sdk.session.status()).pipe( + Effect.map((item): Record => item.data ?? {}), + Effect.orElseSucceed((): Record => ({})), + ), ], { concurrency: "unbounded", @@ -708,6 +712,8 @@ function createLayer(input: StreamInput) { const sessionPermissions = permissions.filter((item) => item.sessionID === input.sessionID) const sessionQuestions = questions.filter((item) => item.sessionID === input.sessionID) + const status = statuses[input.sessionID]?.type + const settleActive = status !== "busy" && status !== "retry" const history = input.replay ? replaySession({ messages: messagesList, @@ -716,6 +722,7 @@ function createLayer(input: StreamInput) { thinking: input.thinking, limits: input.limits(), providers: input.providers?.(), + settleActive, }) : undefined const replay = @@ -727,6 +734,7 @@ function createLayer(input: StreamInput) { thinking: input.thinking, limits: input.limits(), providers: input.providers?.(), + settleActive, }) : history