From b3fda7b9858c2ff88bfc6afbbec3dc30de910ec8 Mon Sep 17 00:00:00 2001 From: Matthias Goergens Date: Tue, 11 Aug 2026 11:29:27 +0800 Subject: [PATCH 1/3] fix(vscode): report a busy session as turn.agent_busy, not an internal error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The busy rejections in SessionRuntime threw a plain Error, which emitError maps to code "internal" and the webview renders as "Internal error occurred." — alarming and wrong for what is simply "a response is already being generated". Throw KimiError with the existing turn.agent_busy code instead so the UI shows the mapped "A message is being sent. Please wait." with the real detail attached. Fixes #2796 --- .changeset/vscode-busy-error-message.md | 5 +++++ apps/vscode/src/runtime/session-runtime.ts | 5 +++-- apps/vscode/test/kimi-runtime.test.ts | 2 ++ apps/vscode/test/settings-store.test.ts | 6 +++--- 4 files changed, 13 insertions(+), 5 deletions(-) create mode 100644 .changeset/vscode-busy-error-message.md 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/apps/vscode/src/runtime/session-runtime.ts b/apps/vscode/src/runtime/session-runtime.ts index a3bd2b4615..4b1db8ff36 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, @@ -188,7 +189,7 @@ 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 }, ); @@ -225,7 +226,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/kimi-runtime.test.ts b/apps/vscode/test/kimi-runtime.test.ts index 6a86f7f2d1..a5cfdc856e 100644 --- a/apps/vscode/test/kimi-runtime.test.ts +++ b/apps/vscode/test/kimi-runtime.test.ts @@ -599,6 +599,8 @@ describe("Kimi runtime (owns shared SDK sessions for Webviews)", () => { 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, diff --git a/apps/vscode/test/settings-store.test.ts b/apps/vscode/test/settings-store.test.ts index 93d0decad8..b26590a29e 100644 --- a/apps/vscode/test/settings-store.test.ts +++ b/apps/vscode/test/settings-store.test.ts @@ -426,15 +426,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); From eaca5e6dcc7cb579d21de7ab43023cc15a66e711 Mon Sep 17 00:00:00 2001 From: Matthias Goergens Date: Tue, 11 Aug 2026 19:27:45 +0800 Subject: [PATCH 2/3] fix(vscode): queue or restore prompts bounced by a busy session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A prompt sent while the session was busy with a turn the webview lost track of (isStreaming resets to false on loadSession even when the session has a live turn — e.g. after a window reload) was rejected with a busy warning, and the text sat parked in pendingInput — invisible, not queued — until the other turn's terminal event finally restored it. The runtime now marks the busy rejection with reason "busy" and the bridge result carries bounced: true, so the webview can tell "this send never started a turn" apart from a normal turn end. On a bounce the message moves into the send queue (flushed by the live turn's terminal event) and the composer keeps streaming state; any other send that fails before its TurnBegin rolls the text straight back into the composer. A bounce during an exclusive operation (terminal rejection, no later terminal event) is left to the error path as before. --- .changeset/vscode-busy-bounce-queue.md | 5 + apps/vscode/src/handlers/chat.handler.ts | 6 +- apps/vscode/src/runtime/session-runtime.ts | 7 +- .../test/kimi-harness.integration.test.ts | 2 +- apps/vscode/test/kimi-runtime.test.ts | 4 +- apps/vscode/test/settings-store.test.ts | 141 ++++++++++++++++++ apps/vscode/webview-ui/src/services/bridge.ts | 2 +- .../webview-ui/src/stores/chat.store.ts | 65 +++++++- 8 files changed, 224 insertions(+), 8 deletions(-) create mode 100644 .changeset/vscode-busy-bounce-queue.md 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/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/runtime/session-runtime.ts b/apps/vscode/src/runtime/session-runtime.ts index 4b1db8ff36..0ff368d8d5 100644 --- a/apps/vscode/src/runtime/session-runtime.ts +++ b/apps/vscode/src/runtime/session-runtime.ts @@ -50,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 { @@ -193,7 +198,7 @@ export class SessionRuntime { "runtime", { terminal: this.hasActiveWork ? false : undefined }, ); - return { status: "failed" }; + return { status: "failed", reason: "busy" }; } let resolveCompletion!: (result: PromptResult) => void; 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 a5cfdc856e..6af3c3706e 100644 --- a/apps/vscode/test/kimi-runtime.test.ts +++ b/apps/vscode/test/kimi-runtime.test.ts @@ -592,7 +592,7 @@ 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); @@ -626,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 b26590a29e..c1f345d352 100644 --- a/apps/vscode/test/settings-store.test.ts +++ b/apps/vscode/test/settings-store.test.ts @@ -67,6 +67,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 +420,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); @@ -449,3 +452,141 @@ 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); + }); +}); diff --git a/apps/vscode/webview-ui/src/services/bridge.ts b/apps/vscode/webview-ui/src/services/bridge.ts index 04f90f274b..d740220b8e 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() { diff --git a/apps/vscode/webview-ui/src/stores/chat.store.ts b/apps/vscode/webview-ui/src/stores/chat.store.ts index 9ec7691860..1654db54e5 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(), @@ -349,6 +410,7 @@ export const useChatStore = create((set, get) => ({ isStreaming: false, isCompacting: false, handshakeReceived: false, + awaitingTurnBegin: false, draftMedia: [], lastStatus: null, tokenUsage: createEmptyTokenUsage(), @@ -476,6 +538,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 = []; }), From aab9d33b0ec1d717524b9f742b2cfe23e9367d8b Mon Sep 17 00:00:00 2001 From: Matthias Goergens Date: Tue, 11 Aug 2026 21:29:48 +0800 Subject: [PATCH 3/3] fix(vscode): restore streaming state when attaching to a live session loadSession reset isStreaming to false unconditionally, so a view that attached to a session with an in-flight turn (started in another view, or still running when the window reloaded) believed the session was idle. New input then took the send path and bounced off the busy runtime instead of queueing. LoadKimiSessionHistory now appends a synthetic turn_active marker when the runtime is busy, and the store keeps isStreaming set when a replay contains it: new input enqueues and is flushed by the live turn's terminal event. --- .changeset/vscode-reattach-streaming-state.md | 5 ++ apps/vscode/shared/bridge.ts | 3 + apps/vscode/src/handlers/session.handler.ts | 10 +++ apps/vscode/test/bridge-handler.test.ts | 66 +++++++++++++++++++ apps/vscode/test/settings-store.test.ts | 53 +++++++++++++++ apps/vscode/webview-ui/src/services/bridge.ts | 4 ++ .../webview-ui/src/stores/chat.store.ts | 26 +++++++- .../webview-ui/src/stores/event-handlers.ts | 7 ++ 8 files changed, 172 insertions(+), 2 deletions(-) create mode 100644 .changeset/vscode-reattach-streaming-state.md 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/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/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/settings-store.test.ts b/apps/vscode/test/settings-store.test.ts index c1f345d352..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(); @@ -590,3 +594,52 @@ describe("Webview send reply correlation", () => { 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 d740220b8e..1268562744 100644 --- a/apps/vscode/webview-ui/src/services/bridge.ts +++ b/apps/vscode/webview-ui/src/services/bridge.ts @@ -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 1654db54e5..eb9ce308ac 100644 --- a/apps/vscode/webview-ui/src/stores/chat.store.ts +++ b/apps/vscode/webview-ui/src/stores/chat.store.ts @@ -371,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) { @@ -385,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 () => { 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) {