diff --git a/.changeset/vscode-busy-bounce-queue.md b/.changeset/vscode-busy-bounce-queue.md new file mode 100644 index 0000000000..094e067d1c --- /dev/null +++ b/.changeset/vscode-busy-bounce-queue.md @@ -0,0 +1,5 @@ +--- +"kimi-code": patch +--- + +Fix prompts vanishing when sent while the session is busy with a turn the chat lost track of (e.g. a live turn after a window reload). The send used to be rejected with "A message is being sent." and the text was parked invisibly until the other turn ended; a busy bounce now moves the message into the queue so it sends when the running turn finishes, and any other send that fails before its turn starts immediately restores the text into the composer. diff --git a/.changeset/vscode-busy-error-message.md b/.changeset/vscode-busy-error-message.md new file mode 100644 index 0000000000..b361b449cb --- /dev/null +++ b/.changeset/vscode-busy-error-message.md @@ -0,0 +1,5 @@ +--- +"kimi-code": patch +--- + +Stop reporting a busy chat session as "Internal error occurred." A prompt sent while a response is still being generated now surfaces as "A message is being sent. Please wait." (code `turn.agent_busy`) with the original detail attached, instead of the generic internal-error message that the plain `Error` was mapped to. diff --git a/.changeset/vscode-reattach-streaming-state.md b/.changeset/vscode-reattach-streaming-state.md new file mode 100644 index 0000000000..3dab07eaaa --- /dev/null +++ b/.changeset/vscode-reattach-streaming-state.md @@ -0,0 +1,5 @@ +--- +"kimi-code": patch +--- + +Keep the composer in streaming state when a chat is opened while its session has a live turn (e.g. a turn started in another view, or one still running when the window reloaded). The session history reply now carries a `turn_active` marker when the runtime is busy, and the store honors it — new input enqueues and is sent when the running turn finishes, instead of taking the send path and bouncing off the busy runtime. diff --git a/apps/vscode/shared/bridge.ts b/apps/vscode/shared/bridge.ts index d7f9bd88ae..0831db79d3 100644 --- a/apps/vscode/shared/bridge.ts +++ b/apps/vscode/shared/bridge.ts @@ -41,6 +41,7 @@ export const Methods = { GetKimiSessions: "getKimiSessions", GetAllKimiSessions: "getAllKimiSessions", + IsSessionBusy: "isSessionBusy", GetRegisteredWorkDirs: "getRegisteredWorkDirs", SetWorkDir: "setWorkDir", BrowseWorkDir: "browseWorkDir", @@ -194,6 +195,8 @@ function validateParams(method: RpcMethod, params: unknown): boolean { return isPlainObject(params) && (params["workDir"] === null || typeof params["workDir"] === "string"); case Methods.LoadKimiSessionHistory: return hasNonEmptyString(params, "kimiSessionId"); + case Methods.IsSessionBusy: + return hasNonEmptyString(params, "sessionId"); case Methods.DeleteKimiSession: return hasNonEmptyString(params, "sessionId"); case Methods.ForkKimiSession: diff --git a/apps/vscode/src/handlers/chat.handler.ts b/apps/vscode/src/handlers/chat.handler.ts index c5b21f89ad..a78c50f4e9 100644 --- a/apps/vscode/src/handlers/chat.handler.ts +++ b/apps/vscode/src/handlers/chat.handler.ts @@ -71,7 +71,7 @@ function prependSystemContext(content: string | ContentPart[], context: string): return copy; } -const streamChat: Handler = async (params, ctx) => { +const streamChat: Handler = async (params, ctx) => { if (!ctx.workDir) { emitPreflightError(ctx, "NO_WORKSPACE", "Please open a folder to start."); void vscode.window.showWarningMessage("Kimi: Please open a folder first.", "Open Folder").then((action) => { @@ -137,7 +137,9 @@ const streamChat: Handler = async (params, const systemContext = await buildSystemContext(runtime.id, ctx); try { const result = await runtime.prompt(prependSystemContext(params.content, systemContext)); - return { done: result.status === "finished" }; + // A busy bounce tells the webview the message never started a turn and is + // safe to queue for when the live turn ends. + return { done: result.status === "finished", bounced: result.reason === "busy" ? true : undefined }; } catch (error) { emitCaughtError(ctx, error, "runtime", runtime.id); return { done: false }; diff --git a/apps/vscode/src/handlers/session.handler.ts b/apps/vscode/src/handlers/session.handler.ts index 817d6960bf..628c0de878 100644 --- a/apps/vscode/src/handlers/session.handler.ts +++ b/apps/vscode/src/handlers/session.handler.ts @@ -41,6 +41,10 @@ export const sessionHandlers: Record> = { .map(toSessionInfo); }, + [Methods.IsSessionBusy]: async (params: { sessionId: string }, ctx): Promise<{ busy: boolean }> => { + return { busy: ctx.runtime.getSession(params.sessionId)?.isBusy ?? false }; + }, + [Methods.GetRegisteredWorkDirs]: async (_, ctx): Promise => { if (!ctx.workspaceRoot) return []; const sessions = await ctx.harness.listSessions(); @@ -172,6 +176,12 @@ export const sessionHandlers: Record> = { ctx.logError("Unable to show the file change warning", noticeError); }); } + if (runtime.isBusy) { + // The session has an in-flight turn (e.g. started by another view, or + // still running while this webview reloaded): tell the store, so the + // composer queues new input instead of bouncing off the busy runtime. + history.push({ type: "turn_active", payload: {}, _sessionId: runtime.id }); + } return history; }, diff --git a/apps/vscode/src/runtime/session-runtime.ts b/apps/vscode/src/runtime/session-runtime.ts index a3bd2b4615..0ff368d8d5 100644 --- a/apps/vscode/src/runtime/session-runtime.ts +++ b/apps/vscode/src/runtime/session-runtime.ts @@ -1,5 +1,6 @@ import { isKimiError, + KimiError, type ContentPart as SdkContentPart, type Event, type PromptInput, @@ -49,6 +50,11 @@ const ALREADY_GENERATING_MESSAGE = "A response is already being generated for th export interface PromptResult { readonly status: "finished" | "cancelled" | "failed"; + /** + * Present when the call was rejected because the session was already busy — + * the message never started a turn and can safely be queued for later. + */ + readonly reason?: "busy"; } interface SuppressedError { @@ -188,11 +194,11 @@ export class SessionRuntime { // such terminal event, so reject terminally: the caller's composer must // unlock rather than hang until the handshake timeout. this.emitError( - new Error(ALREADY_GENERATING_MESSAGE), + new KimiError("turn.agent_busy", ALREADY_GENERATING_MESSAGE), "runtime", { terminal: this.hasActiveWork ? false : undefined }, ); - return { status: "failed" }; + return { status: "failed", reason: "busy" }; } let resolveCompletion!: (result: PromptResult) => void; @@ -225,7 +231,7 @@ export class SessionRuntime { beginHostAction(input: string | LegacyContentPart[], forkable = false): number { this.ensureOpen(); if (this.isBusy) { - throw new Error(ALREADY_GENERATING_MESSAGE); + throw new KimiError("turn.agent_busy", ALREADY_GENERATING_MESSAGE); } const actionId = ++this.hostActionSequence; this.hostActionActive = true; diff --git a/apps/vscode/test/bridge-handler.test.ts b/apps/vscode/test/bridge-handler.test.ts index 51b203823c..0087d1e137 100644 --- a/apps/vscode/test/bridge-handler.test.ts +++ b/apps/vscode/test/bridge-handler.test.ts @@ -362,6 +362,72 @@ describe("Webview RPC boundary (validates requests before host dispatch)", () => expect(deleteBaseline).toHaveBeenCalledWith("session-2"); }); + it("appends a turn_active marker when the resumed session has a live turn", async () => { + const session = createResumedSession("session-1", root); + host.harness.resumeSession.mockResolvedValue(session as never); + + const first = await bridge.handle( + { + id: "rpc-1", + method: Methods.LoadKimiSessionHistory, + params: { kimiSessionId: "session-1" }, + }, + "view-1", + ); + expect(first).toEqual({ + id: "rpc-1", + result: expect.not.arrayContaining([ + expect.objectContaining({ type: "turn_active" }), + ]), + }); + + // A second attach while the runtime reports a live turn must say so. + const runtime = bridge.runtime.getSession("session-1")!; + vi.spyOn(runtime, "isBusy", "get").mockReturnValue(true); + + const second = await bridge.handle( + { + id: "rpc-2", + method: Methods.LoadKimiSessionHistory, + params: { kimiSessionId: "session-1" }, + }, + "view-1", + ); + expect(second).toEqual({ + id: "rpc-2", + result: expect.arrayContaining([ + expect.objectContaining({ type: "turn_active", _sessionId: "session-1" }), + ]), + }); + }); + + it("reports whether a session has live work", async () => { + const session = createResumedSession("session-1", root); + host.harness.resumeSession.mockResolvedValue(session as never); + await bridge.handle( + { + id: "rpc-1", + method: Methods.LoadKimiSessionHistory, + params: { kimiSessionId: "session-1" }, + }, + "view-1", + ); + + const idle = await bridge.handle( + { id: "rpc-2", method: Methods.IsSessionBusy, params: { sessionId: "session-1" } }, + "view-1", + ); + expect(idle).toEqual({ id: "rpc-2", result: { busy: false } }); + + const runtime = bridge.runtime.getSession("session-1")!; + vi.spyOn(runtime, "isBusy", "get").mockReturnValue(true); + const busy = await bridge.handle( + { id: "rpc-3", method: Methods.IsSessionBusy, params: { sessionId: "session-1" } }, + "view-1", + ); + expect(busy).toEqual({ id: "rpc-3", result: { busy: true } }); + }); + it("keeps conversation history available when its baseline snapshot disappears", async () => { const session = createResumedSession("session-1", root); host.harness.resumeSession.mockResolvedValueOnce(session as never); diff --git a/apps/vscode/test/kimi-harness.integration.test.ts b/apps/vscode/test/kimi-harness.integration.test.ts index ea53fae3d3..bb120e6c51 100644 --- a/apps/vscode/test/kimi-harness.integration.test.ts +++ b/apps/vscode/test/kimi-harness.integration.test.ts @@ -1122,7 +1122,7 @@ describe("VS Code Kimi harness integration (shares one in-process SDK home)", () const first = runtime.prompt("first message"); await blocked.started; - await expect(runtime.prompt("concurrent message")).resolves.toEqual({ status: "failed" }); + await expect(runtime.prompt("concurrent message")).resolves.toEqual({ status: "failed", reason: "busy" }); // The rejection surfaces as a mid-turn warning; the active turn is untouched. expect(runtime.isBusy).toBe(true); diff --git a/apps/vscode/test/kimi-runtime.test.ts b/apps/vscode/test/kimi-runtime.test.ts index 6a86f7f2d1..6af3c3706e 100644 --- a/apps/vscode/test/kimi-runtime.test.ts +++ b/apps/vscode/test/kimi-runtime.test.ts @@ -592,13 +592,15 @@ describe("Kimi runtime (owns shared SDK sessions for Webviews)", () => { boundary.emit({ type: "turn.started", agentId: "main", sessionId: opened.id, turnId: "t1" } as unknown as Event); expect(opened.isBusy).toBe(true); - await expect(opened.prompt("concurrent message")).resolves.toEqual({ status: "failed" }); + await expect(opened.prompt("concurrent message")).resolves.toEqual({ status: "failed", reason: "busy" }); // The rejection surfaces as a mid-turn warning; the active turn is untouched. expect(opened.isBusy).toBe(true); const busyWarning = broadcasts.find(({ data }) => (data as { type?: string }).type === "error"); expect(busyWarning?.data).toMatchObject({ type: "error", + code: "turn.agent_busy", + message: "A message is being sent. Please wait.", phase: "runtime", detail: "A response is already being generated for this session.", terminal: false, @@ -624,7 +626,7 @@ describe("Kimi runtime (owns shared SDK sessions for Webviews)", () => { // terminal so the caller's composer can unlock. expect(opened.isBusy).toBe(true); - await expect(opened.prompt("during fork")).resolves.toEqual({ status: "failed" }); + await expect(opened.prompt("during fork")).resolves.toEqual({ status: "failed", reason: "busy" }); const rejection = broadcasts.find(({ data }) => (data as { type?: string }).type === "error"); expect(rejection?.data).toMatchObject({ type: "error", phase: "runtime" }); expect((rejection?.data as Record)["terminal"]).toBeUndefined(); diff --git a/apps/vscode/test/settings-store.test.ts b/apps/vscode/test/settings-store.test.ts index 93d0decad8..89f4b27e9a 100644 --- a/apps/vscode/test/settings-store.test.ts +++ b/apps/vscode/test/settings-store.test.ts @@ -13,6 +13,7 @@ const boundary = vi.hoisted(() => ({ streamChat: vi.fn(), abortChat: vi.fn(), trackFiles: vi.fn(), + isSessionBusy: vi.fn(), toastError: vi.fn(), toastWarning: vi.fn(), })); @@ -23,6 +24,7 @@ vi.mock("@/services", () => ({ streamChat: boundary.streamChat, abortChat: boundary.abortChat, trackFiles: boundary.trackFiles, + isSessionBusy: boundary.isSessionBusy, }, })); vi.mock("@/components/ui/sonner", () => ({ @@ -57,6 +59,8 @@ beforeEach(() => { boundary.streamChat.mockResolvedValue({ done: false }); boundary.abortChat.mockReset(); boundary.abortChat.mockResolvedValue({ aborted: true }); + boundary.isSessionBusy.mockReset(); + boundary.isSessionBusy.mockResolvedValue({ busy: true }); boundary.trackFiles.mockReset(); boundary.toastError.mockReset(); boundary.toastWarning.mockReset(); @@ -67,6 +71,7 @@ beforeEach(() => { isStreaming: false, isCompacting: false, handshakeReceived: false, + awaitingTurnBegin: false, draftMedia: [], lastStatus: null, tokenUsage: { input_other: 0, output: 0, input_cache_read: 0, input_cache_creation: 0 }, @@ -419,6 +424,8 @@ describe("Webview thinking effort parity with the TUI", () => { describe("Webview mid-turn warnings", () => { it("shows a non-terminal error as a toast without unlocking the composer", async () => { + // A genuinely started turn keeps its bridge call open until the turn ends. + boundary.streamChat.mockImplementation(() => new Promise(() => {})); useChatStore.getState().sendMessage("first message"); useChatStore.getState().sendMessage("queued follow-up"); expect(useChatStore.getState().isStreaming).toBe(true); @@ -426,15 +433,15 @@ describe("Webview mid-turn warnings", () => { useChatStore.getState().processEvent({ type: "error", - code: "internal", - message: "Internal error occurred.", + code: "turn.agent_busy", + message: "A message is being sent. Please wait.", detail: "A response is already being generated for this session.", phase: "runtime", terminal: false, }); // The turn is still running: nothing unlocks, nothing flushes, nothing is retried. - expect(boundary.toastWarning).toHaveBeenCalledWith("Internal error occurred."); + expect(boundary.toastWarning).toHaveBeenCalledWith("A message is being sent. Please wait."); const state = useChatStore.getState(); expect(state.isStreaming).toBe(true); expect(state.queue).toHaveLength(1); @@ -449,3 +456,190 @@ describe("Webview mid-turn warnings", () => { }); }); }); + +describe("Webview send bounce handling", () => { + it("enqueues the message when the session bounced the send as busy", async () => { + boundary.streamChat.mockResolvedValue({ done: false, bounced: true }); + + useChatStore.getState().sendMessage("hold this"); + + await vi.waitFor(() => { + expect(useChatStore.getState().queue).toHaveLength(1); + }); + const state = useChatStore.getState(); + expect(state.queue[0]?.content).toBe("hold this"); + // The live (untracked) turn still owns the composer state: stays + // streaming so further input enqueues, and nothing is parked in + // pendingInput. + expect(state.isStreaming).toBe(true); + expect(state.pendingInput).toBeNull(); + expect(state.awaitingTurnBegin).toBe(false); + }); + + it("rolls the composer back when the send never started a turn", async () => { + boundary.streamChat.mockResolvedValue({ done: false }); + + useChatStore.getState().sendMessage("give it back"); + + await vi.waitFor(() => { + expect(useChatStore.getState().isStreaming).toBe(false); + }); + const state = useChatStore.getState(); + // pendingInput keeps the text so the composer restores it. + expect(state.pendingInput).toEqual({ content: "give it back", model: "plain" }); + expect(state.queue).toHaveLength(0); + expect(state.awaitingTurnBegin).toBe(false); + }); + + it("does not roll back a send whose turn already started", async () => { + let resolveSend!: (result: { done: boolean }) => void; + boundary.streamChat.mockImplementation( + () => new Promise<{ done: boolean }>((resolve) => { resolveSend = resolve; }), + ); + + useChatStore.getState().sendMessage("in flight"); + useChatStore.getState().processEvent({ + type: "TurnBegin", + payload: { user_input: "in flight" }, + }); + resolveSend({ done: true }); + + await vi.waitFor(() => { + expect(boundary.streamChat).toHaveBeenCalledTimes(1); + }); + const state = useChatStore.getState(); + expect(state.isStreaming).toBe(true); + expect(state.pendingInput).toEqual({ content: "in flight", model: "plain" }); + expect(state.queue).toHaveLength(0); + }); +}); + +describe("Webview send bounce ordering", () => { + it("resends the prompt when the busy turn completes before the bounce reply arrives", async () => { + let resolveFirst!: (result: { done: boolean; bounced: boolean }) => void; + boundary.streamChat + .mockImplementationOnce( + () => + new Promise<{ done: boolean; bounced: boolean }>((resolve) => { + resolveFirst = resolve; + }), + ) + .mockResolvedValue({ done: true }); + + useChatStore.getState().sendMessage("later"); + // The live turn ends before the bounce reply lands: its terminal event + // clears the parked input and the streaming state. + useChatStore.getState().processEvent({ type: "stream_complete", result: { status: "finished" } }); + resolveFirst({ done: false, bounced: true }); + + await vi.waitFor(() => { + expect(boundary.streamChat).toHaveBeenCalledTimes(2); + }); + const state = useChatStore.getState(); + expect(state.isStreaming).toBe(true); + expect(state.pendingInput).toEqual({ content: "later", model: "plain" }); + expect(state.queue).toHaveLength(0); + }); +}); + +describe("Webview send reply correlation", () => { + it("enqueues a bounced send even when another view's TurnBegin arrived first", async () => { + boundary.streamChat.mockResolvedValue({ done: false, bounced: true }); + + useChatStore.getState().sendMessage("from the losing view"); + // The winning view's TurnBegin is broadcast to every subscriber and + // clears our awaitingTurnBegin before our bounce reply arrives. + useChatStore.getState().processEvent({ + type: "TurnBegin", + payload: { user_input: "the winning view" }, + }); + + await vi.waitFor(() => { + expect(useChatStore.getState().queue).toHaveLength(1); + }); + expect(useChatStore.getState().queue[0]?.content).toBe("from the losing view"); + }); + + it("ignores a stale bridge reply once a newer send owns the composer", async () => { + let resolveFirst!: (result: { done: boolean }) => void; + boundary.streamChat + .mockImplementationOnce( + () => new Promise<{ done: boolean }>((resolve) => { resolveFirst = resolve; }), + ) + // The second send stays in flight for the whole test. + .mockImplementation(() => new Promise(() => {})); + + useChatStore.getState().sendMessage("first"); + useChatStore.getState().processEvent({ + type: "TurnBegin", + payload: { user_input: "first" }, + }); + useChatStore.getState().processEvent({ + type: "error", + code: "provider.api_error", + message: "Service temporarily unavailable.", + phase: "runtime", + }); + expect(useChatStore.getState().isStreaming).toBe(false); + + useChatStore.getState().sendMessage("second"); + // The first send's bridge reply only arrives now — it must not touch the + // second send's state. + resolveFirst({ done: false }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const state = useChatStore.getState(); + expect(state.isStreaming).toBe(true); + expect(state.pendingInput).toEqual({ content: "second", model: "plain" }); + expect(state.queue).toHaveLength(0); + }); +}); + +describe("Webview session load streaming state", () => { + it("keeps streaming state when the loaded session has an active turn", async () => { + await useChatStore.getState().loadSession("s1", [ + { type: "turn_active", payload: {}, _sessionId: "s1" }, + ]); + + expect(useChatStore.getState().isStreaming).toBe(true); + // New input enqueues instead of taking the send path. + useChatStore.getState().sendMessage("queued while live"); + expect(useChatStore.getState().queue).toHaveLength(1); + expect(boundary.streamChat).not.toHaveBeenCalled(); + }); + + it("resets streaming state when the loaded session has no active turn", async () => { + useChatStore.setState({ isStreaming: true }); + + await useChatStore.getState().loadSession("s1", []); + + expect(useChatStore.getState().isStreaming).toBe(false); + }); +}); + +describe("Webview session load revalidation", () => { + it("unlocks the composer when the live turn ended before the replay was applied", async () => { + boundary.isSessionBusy.mockResolvedValue({ busy: false }); + + await useChatStore.getState().loadSession("s1", [ + { type: "turn_active", payload: {}, _sessionId: "s1" }, + ]); + + await vi.waitFor(() => { + expect(useChatStore.getState().isStreaming).toBe(false); + }); + }); + + it("keeps streaming state when the session is still busy at apply time", async () => { + boundary.isSessionBusy.mockResolvedValue({ busy: true }); + + await useChatStore.getState().loadSession("s1", [ + { type: "turn_active", payload: {}, _sessionId: "s1" }, + ]); + + await vi.waitFor(() => { + expect(boundary.isSessionBusy).toHaveBeenCalledWith("s1"); + }); + expect(useChatStore.getState().isStreaming).toBe(true); + }); +}); diff --git a/apps/vscode/webview-ui/src/services/bridge.ts b/apps/vscode/webview-ui/src/services/bridge.ts index 04f90f274b..1268562744 100644 --- a/apps/vscode/webview-ui/src/services/bridge.ts +++ b/apps/vscode/webview-ui/src/services/bridge.ts @@ -184,7 +184,7 @@ class Bridge { } streamChat(content: string | ContentPart[], model: string, effort: string, planMode: boolean, sessionId?: string) { - return this.call<{ done: boolean }>(Methods.StreamChat, { content, model, effort, planMode, sessionId }); + return this.call<{ done: boolean; bounced?: boolean }>(Methods.StreamChat, { content, model, effort, planMode, sessionId }); } abortChat() { @@ -215,6 +215,10 @@ class Bridge { return this.call(Methods.GetAllKimiSessions); } + isSessionBusy(sessionId: string) { + return this.call<{ busy: boolean }>(Methods.IsSessionBusy, { sessionId }); + } + getRegisteredWorkDirs() { return this.call(Methods.GetRegisteredWorkDirs); } diff --git a/apps/vscode/webview-ui/src/stores/chat.store.ts b/apps/vscode/webview-ui/src/stores/chat.store.ts index 9ec7691860..eb9ce308ac 100644 --- a/apps/vscode/webview-ui/src/stores/chat.store.ts +++ b/apps/vscode/webview-ui/src/stores/chat.store.ts @@ -93,6 +93,8 @@ export interface ChatState { isStreaming: boolean; isCompacting: boolean; handshakeReceived: boolean; + /** True from send until the sent message's TurnBegin arrives. */ + awaitingTurnBegin: boolean; draftMedia: DraftMediaItem[]; lastStatus: StatusUpdate | null; tokenUsage: TokenUsage; @@ -125,6 +127,9 @@ export interface ChatState { } let handshakeTimer: ReturnType | null = null; +// Monotonic token identifying the latest send; bridge replies from earlier +// sends must not touch composer state they no longer own. +let sendGeneration = 0; function clearHandshakeTimer() { if (handshakeTimer) { @@ -144,6 +149,7 @@ function clearAllInlineErrors(draft: ChatState): void { function doSend(state: ChatState, content: string | ContentPart[], model: string) { const { sessionId, planMode } = state; const { thinkingEffort } = useSettingsStore.getState(); + const generation = ++sendGeneration; clearHandshakeTimer(); handshakeTimer = setTimeout(() => { @@ -161,6 +167,54 @@ function doSend(state: ChatState, content: string | ContentPart[], model: string void bridge .streamChat(content, model, thinkingEffort, planMode, sessionId ?? undefined) + .then((result) => { + // Ignore stale replies: a newer send owns the composer state now. + if (generation !== sendGeneration) { + return; + } + const s = useChatStore.getState(); + if (result.bounced === true) { + // Our send never started a turn. Note the TurnBegin that may have + // cleared awaitingTurnBegin is not necessarily ours: with the same + // session open in two views, the winning view's TurnBegin is + // broadcast to every subscriber, so key the branches on the parked + // input instead. + if (s.pendingInput !== null) { + if (s.isStreaming) { + // The session is busy with a turn this store lost track of (e.g. + // a live turn after a reload, or another view's): queue the + // message so it sends when that turn's terminal event flushes + // the queue, and keep the streaming state so further input + // enqueues too. + const pending = s.pendingInput; + useChatStore.setState({ awaitingTurnBegin: false, pendingInput: null }); + s.enqueue(pending.content, pending.model); + } else { + // A terminal error event already handled this send (e.g. a bounce + // during an exclusive operation) — keep the state it produced; + // the parked pendingInput drives the composer restore. + useChatStore.setState({ awaitingTurnBegin: false }); + } + } else { + // The busy turn ended before this reply arrived: its stream_complete + // already cleared the parked input, so the session is free now — + // send again instead of losing the prompt. + useChatStore.setState( + produce((draft: ChatState) => { + draft.isStreaming = true; + draft.handshakeReceived = false; + draft.awaitingTurnBegin = true; + draft.pendingInput = { content, model }; + }), + ); + doSend(useChatStore.getState(), content, model); + } + } else if (result.done === false && s.awaitingTurnBegin) { + // The send never started a turn: roll the text back into the composer + // (via pendingInput) instead of parking it until some later event. + useChatStore.setState({ isStreaming: false, awaitingTurnBegin: false }); + } + }) .catch((error: unknown) => { const detail = error instanceof Error ? error.message : String(error); useChatStore.getState().processEvent({ @@ -179,6 +233,7 @@ export const useChatStore = create((set, get) => ({ isStreaming: false, isCompacting: false, handshakeReceived: false, + awaitingTurnBegin: false, draftMedia: [], lastStatus: null, tokenUsage: createEmptyTokenUsage(), @@ -213,6 +268,7 @@ export const useChatStore = create((set, get) => ({ draft.draftMedia = []; draft.isStreaming = true; draft.handshakeReceived = false; + draft.awaitingTurnBegin = true; draft.pendingInput = { content, model: currentModel }; }), ); @@ -234,6 +290,7 @@ export const useChatStore = create((set, get) => ({ clearAllInlineErrors(draft); draft.isStreaming = true; draft.handshakeReceived = false; + draft.awaitingTurnBegin = true; const lastAssistant = draft.messages.at(-1); if (lastAssistant?.role === "assistant" && lastAssistant.inlineError) { draft.messages.pop(); @@ -258,7 +315,10 @@ export const useChatStore = create((set, get) => ({ return; } // Clear handshake timeout on receiving valid response - if (event.type === "TurnBegin" || event.type === "StepBegin" || event.type === "ContentPart") { + if (event.type === "TurnBegin") { + clearHandshakeTimer(); + set({ handshakeReceived: true, awaitingTurnBegin: false }); + } else if (event.type === "StepBegin" || event.type === "ContentPart") { clearHandshakeTimer(); set({ handshakeReceived: true }); } else if (event.type === "stream_complete" || event.type === "error") { @@ -295,6 +355,7 @@ export const useChatStore = create((set, get) => ({ isStreaming: false, isCompacting: false, handshakeReceived: false, + awaitingTurnBegin: false, draftMedia: [], lastStatus: null, tokenUsage: createEmptyTokenUsage(), @@ -310,7 +371,10 @@ export const useChatStore = create((set, get) => ({ get().processEvent(event); } - // All steps are finished when loading from history + // All steps are finished when loading from history. A turn_active marker + // means the session has an in-flight turn — keep the streaming state so + // new input enqueues instead of bouncing off the busy runtime. + const hasActiveTurn = events.some((event) => event.type === "turn_active"); set( produce((draft: ChatState) => { for (const msg of draft.messages) { @@ -324,12 +388,31 @@ export const useChatStore = create((set, get) => ({ } } } - draft.isStreaming = false; + draft.isStreaming = hasActiveTurn; draft.isCompacting = false; draft.pendingQuestion = null; }), ); useApprovalStore.getState().clearRequests(); + + if (hasActiveTurn) { + // The marker was sampled when the history was built; if the live turn + // ended while the replay was being applied, its terminal event was + // consumed by the pre-load state and no later one will come. Revalidate + // and converge: unlock the composer and flush anything queued. + void bridge + .isSessionBusy(sessionId) + .then(({ busy }) => { + if (busy) return; + const s = get(); + if (s.sessionId !== sessionId || !s.isStreaming) return; + set({ isStreaming: false }); + if (s.queue.length > 0) { + setTimeout(() => get().sendNextQueued(), 50); + } + }) + .catch(() => undefined); + } }, startNewConversation: async () => { @@ -349,6 +432,7 @@ export const useChatStore = create((set, get) => ({ isStreaming: false, isCompacting: false, handshakeReceived: false, + awaitingTurnBegin: false, draftMedia: [], lastStatus: null, tokenUsage: createEmptyTokenUsage(), @@ -476,6 +560,7 @@ export const useChatStore = create((set, get) => ({ draft.queue = rest; draft.isStreaming = true; draft.handshakeReceived = false; + draft.awaitingTurnBegin = true; draft.pendingInput = { content: next.content, model: next.model }; draft.draftMedia = []; }), diff --git a/apps/vscode/webview-ui/src/stores/event-handlers.ts b/apps/vscode/webview-ui/src/stores/event-handlers.ts index dc66b857e1..dd6290b065 100644 --- a/apps/vscode/webview-ui/src/stores/event-handlers.ts +++ b/apps/vscode/webview-ui/src/stores/event-handlers.ts @@ -289,6 +289,13 @@ function handleRuntimeError(draft: ChatState, code: string, message: string, det } const eventHandlers: Record = { + // Synthetic marker appended to a history replay when the session has an + // in-flight turn: the composer must treat the session as streaming (queue + // new input) even though this store never saw the turn start. + turn_active: (draft) => { + draft.isStreaming = true; + }, + // UI 事件 (Bridge 层) session_start: (draft, payload: { sessionId: string; model?: string }) => { if (payload.sessionId) {