diff --git a/apps/vscode/shared/types.ts b/apps/vscode/shared/types.ts index 422aa5acc7..2f6c56d610 100644 --- a/apps/vscode/shared/types.ts +++ b/apps/vscode/shared/types.ts @@ -43,6 +43,12 @@ export interface StreamError { message: string; detail?: string; // 原始服务器错误信息 phase: ErrorPhase; + /** + * `false` marks a mid-turn warning: the turn is still running, so UIs must + * not treat it as turn-ending. Do not unlock the composer, offer Retry, or + * flush the queued messages for non-terminal errors. + */ + terminal?: boolean; } export type UIStreamEvent = diff --git a/apps/vscode/src/runtime/session-runtime.ts b/apps/vscode/src/runtime/session-runtime.ts index c1009930ff..45d60ac500 100644 --- a/apps/vscode/src/runtime/session-runtime.ts +++ b/apps/vscode/src/runtime/session-runtime.ts @@ -45,6 +45,8 @@ interface ActivePrompt { resolve: (result: PromptResult) => void; } +const ALREADY_GENERATING_MESSAGE = "A response is already being generated for this session."; + export interface PromptResult { readonly status: "finished" | "cancelled" | "failed"; } @@ -179,28 +181,40 @@ export class SessionRuntime { ): Promise { this.ensureOpen(); if (this.isBusy) { - throw new Error("A response is already being generated for this session."); + // A re-entrant turn request must never disturb the active turn — it fails + // only itself. When a turn or host action is running, its later terminal + // stream event unlocks every subscribed view, so a non-terminal warning + // is enough. An exclusive operation (e.g. fork materialization) emits no + // 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), + "runtime", + { terminal: this.hasActiveWork ? false : undefined }, + ); + return { status: "failed" }; } let resolveCompletion!: (result: PromptResult) => void; const completion = new Promise((resolve) => { resolveCompletion = resolve; }); - this.activePrompt = { + const active: ActivePrompt = { input, started: false, settled: false, resolve: resolveCompletion, }; + this.activePrompt = active; try { await action(); } catch (error) { - if (this.activePrompt !== undefined && !this.activePrompt.started) { - this.emitError(error, "preflight"); - this.settlePrompt({ status: "failed" }); - } else { - this.emitError(error, "runtime"); + // Only settle the prompt this call created. Once the event pipeline or a + // cancel has settled it, the failure was already reported — settling it + // again would misreport the active turn and could emit a duplicate error. + if (!active.settled) { + this.emitError(error, active.started ? "runtime" : "preflight"); this.settlePrompt({ status: "failed" }); } } @@ -211,7 +225,7 @@ export class SessionRuntime { beginHostAction(input: string | LegacyContentPart[], forkable = false): number { this.ensureOpen(); if (this.isBusy) { - throw new Error("A response is already being generated for this session."); + throw new Error(ALREADY_GENERATING_MESSAGE); } const actionId = ++this.hostActionSequence; this.hostActionActive = true; @@ -461,7 +475,15 @@ export class SessionRuntime { } if (adapted.event !== undefined) { - this.emitStreamEvent(adapted.event); + // Errors the core reports while the active turn keeps running (they are + // not followed by a terminal turn.ended) must not look turn-ending to the + // Webview — otherwise the UI unlocks mid-turn and the next send collides + // with the still-active prompt. + const wireEvent = + adapted.event.type === "error" && this.activePrompt?.started === true + ? { ...adapted.event, terminal: false as const } + : adapted.event; + this.emitStreamEvent(wireEvent); if (adapted.event.type === "error" && this.activePrompt !== undefined && !this.activePrompt.started) { this.settlePrompt({ status: "failed" }); } @@ -537,7 +559,7 @@ export class SessionRuntime { return suppressed.code === code && suppressed.message === message; } - private emitError(error: unknown, phase: ErrorPhase): void { + private emitError(error: unknown, phase: ErrorPhase, options?: { readonly terminal?: boolean }): void { const code = isKimiError(error) ? error.code : "internal"; const detail = error instanceof Error ? error.message : String(error); this.log(`Session ${phase} error`, error); @@ -548,6 +570,7 @@ export class SessionRuntime { detail, phase, _sessionId: this.session.id, + terminal: options?.terminal, }); } diff --git a/apps/vscode/test/kimi-harness.integration.test.ts b/apps/vscode/test/kimi-harness.integration.test.ts index 9798fa39b3..36c9505735 100644 --- a/apps/vscode/test/kimi-harness.integration.test.ts +++ b/apps/vscode/test/kimi-harness.integration.test.ts @@ -1054,6 +1054,26 @@ describe("VS Code Kimi harness integration (shares one in-process SDK home)", () expect(runtime.isBusy).toBe(false); }); + it("fails a prompt sent while a turn is running without disturbing the active turn", async () => { + const rig = await createRuntimeRig(); + const blocked = routeBlockedPrompt(rig.provider); + const runtime = await openRuntimeSession(rig); + const first = runtime.prompt("first message"); + await blocked.started; + + await expect(runtime.prompt("concurrent message")).resolves.toEqual({ status: "failed" }); + + // The rejection surfaces as a mid-turn warning; the active turn is untouched. + expect(runtime.isBusy).toBe(true); + expect(streamEvents(rig.broadcasts)).toContainEqual( + expect.objectContaining({ type: "error", terminal: false }), + ); + + blocked.release(); + await expect(first).resolves.toEqual({ status: "finished" }); + expect(runtime.isBusy).toBe(false); + }); + it("stops a running init command without surfacing its late result", async () => { const rig = await createRuntimeRig(); const blocked = routeBlockedPrompt(rig.provider); diff --git a/apps/vscode/test/kimi-runtime.test.ts b/apps/vscode/test/kimi-runtime.test.ts index 6ea9044679..8133c2916e 100644 --- a/apps/vscode/test/kimi-runtime.test.ts +++ b/apps/vscode/test/kimi-runtime.test.ts @@ -34,6 +34,8 @@ interface FakeSessionBoundary { readonly handlerInstallations: { approval: number; question: number }; readonly subscriptionCount: () => number; readonly closeCount: () => number; + readonly emit: (event: Event) => void; + readonly setPromptImpl: (impl: (input: string | PromptInput) => Promise) => void; } function createFakeSession( @@ -50,6 +52,7 @@ function createFakeSession( const handlerInstallations = { approval: 0, question: 0 }; let subscriptions = 0; let closes = 0; + let promptImpl: (input: string | PromptInput) => Promise = async () => {}; let status: SessionStatus = { model: initial.model ?? "kimi-test", thinkingEffort: initial.thinkingEffort ?? "off", @@ -88,7 +91,9 @@ function createFakeSession( listeners.add(listener); return () => listeners.delete(listener); }, - async prompt(_input: string | PromptInput) {}, + async prompt(input: string | PromptInput) { + await promptImpl(input); + }, async steer(_input: string | PromptInput) {}, async cancel() {}, async getStatus() { @@ -124,6 +129,12 @@ function createFakeSession( handlerInstallations, subscriptionCount: () => subscriptions, closeCount: () => closes, + emit: (event: Event) => { + for (const listener of [...listeners]) listener(event); + }, + setPromptImpl: (impl) => { + promptImpl = impl; + }, }; } @@ -551,4 +562,124 @@ describe("Kimi runtime (owns shared SDK sessions for Webviews)", () => { expect(runtime.getSession("foreign-1")).toBeUndefined(); expect(foreign.closeCount()).toBe(1); }); + + function createRecordingRuntime() { + const sdk = createFakeHarness(); + const broadcasts: Array<{ event: string; data: unknown }> = []; + const runtime = new KimiRuntime({ + version: "test", + harness: sdk.harness, + broadcast: (event, data) => { + broadcasts.push({ event, data }); + }, + captureBaseline: () => undefined, + log: () => undefined, + }); + return { runtime, sdk, broadcasts }; + } + + it("fails a reentrant prompt without disturbing the running turn", async () => { + const { runtime, sdk, broadcasts } = createRecordingRuntime(); + const opened = await runtime.openSession(openOptions()); + const boundary = sdk.sessions.get(opened.id)!; + + let releaseTurn!: () => void; + boundary.setPromptImpl(() => new Promise((resolve) => { + releaseTurn = resolve; + })); + const first = opened.prompt("first message"); + 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" }); + + // 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", + phase: "runtime", + detail: "A response is already being generated for this session.", + terminal: false, + }); + + boundary.emit({ type: "turn.ended", agentId: "main", sessionId: opened.id, turnId: "t1", reason: "completed" } as unknown as Event); + releaseTurn(); + await expect(first).resolves.toEqual({ status: "finished" }); + expect(opened.isBusy).toBe(false); + }); + + it("rejects a prompt during an exclusive operation with a terminal error", async () => { + const { runtime, broadcasts } = createRecordingRuntime(); + const opened = await runtime.openSession(openOptions()); + + let releaseExclusive!: () => void; + const exclusive = opened.runExclusiveAfterCancelling( + () => new Promise((resolve) => { + releaseExclusive = resolve; + }), + ); + // No active work means no later stream_complete: the rejection must be + // terminal so the caller's composer can unlock. + expect(opened.isBusy).toBe(true); + + await expect(opened.prompt("during fork")).resolves.toEqual({ status: "failed" }); + 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(); + + releaseExclusive(); + await exclusive; + expect(opened.isBusy).toBe(false); + }); + + it("marks a mid-turn core error as non-terminal until the turn ends", async () => { + const { runtime, sdk, broadcasts } = createRecordingRuntime(); + const opened = await runtime.openSession(openOptions()); + const boundary = sdk.sessions.get(opened.id)!; + + let releaseTurn!: () => void; + boundary.setPromptImpl(() => new Promise((resolve) => { + releaseTurn = resolve; + })); + const first = opened.prompt("first message"); + boundary.emit({ type: "turn.started", agentId: "main", sessionId: opened.id, turnId: "t1" } as unknown as Event); + + boundary.emit({ + type: "error", + agentId: "main", + sessionId: opened.id, + code: "records.write_failed", + message: "Failed to write agent records: EACCES", + } as unknown as Event); + + // The turn is still running: no settlement, no terminal error on the wire. + expect(opened.isBusy).toBe(true); + const warning = broadcasts.find(({ data }) => (data as { type?: string }).type === "error"); + expect(warning?.data).toMatchObject({ + type: "error", + code: "records.write_failed", + phase: "runtime", + terminal: false, + }); + + boundary.emit({ type: "turn.ended", agentId: "main", sessionId: opened.id, turnId: "t1", reason: "completed" } as unknown as Event); + releaseTurn(); + await expect(first).resolves.toEqual({ status: "finished" }); + expect(opened.isBusy).toBe(false); + }); + + it("keeps preflight failures terminal", async () => { + const { runtime, sdk, broadcasts } = createRecordingRuntime(); + const opened = await runtime.openSession(openOptions()); + const boundary = sdk.sessions.get(opened.id)!; + + boundary.setPromptImpl(() => Promise.reject(new Error("provider down"))); + + await expect(opened.prompt("hi")).resolves.toEqual({ status: "failed" }); + const failure = broadcasts.find(({ data }) => (data as { type?: string }).type === "error"); + expect(failure?.data).toMatchObject({ type: "error", phase: "preflight" }); + expect((failure?.data as Record)["terminal"]).toBeUndefined(); + expect(opened.isBusy).toBe(false); + }); }); diff --git a/apps/vscode/test/settings-store.test.ts b/apps/vscode/test/settings-store.test.ts index 1978e2df5e..b830e998b1 100644 --- a/apps/vscode/test/settings-store.test.ts +++ b/apps/vscode/test/settings-store.test.ts @@ -14,6 +14,7 @@ const boundary = vi.hoisted(() => ({ abortChat: vi.fn(), trackFiles: vi.fn(), toastError: vi.fn(), + toastWarning: vi.fn(), })); vi.mock("@/services", () => ({ @@ -25,7 +26,7 @@ vi.mock("@/services", () => ({ }, })); vi.mock("@/components/ui/sonner", () => ({ - toast: { error: boundary.toastError }, + toast: { error: boundary.toastError, warning: boundary.toastWarning }, })); import { @@ -58,6 +59,7 @@ beforeEach(() => { boundary.abortChat.mockResolvedValue({ aborted: true }); boundary.trackFiles.mockReset(); boundary.toastError.mockReset(); + boundary.toastWarning.mockReset(); useSettingsStore.getState().initModels(MODELS, "plain", false); useChatStore.setState({ sessionId: null, @@ -369,3 +371,36 @@ describe("Webview thinking effort parity with the TUI", () => { expect(boundary.saveConfig).not.toHaveBeenCalled(); }); }); + +describe("Webview mid-turn warnings", () => { + it("shows a non-terminal error as a toast without unlocking the composer", async () => { + useChatStore.getState().sendMessage("first message"); + useChatStore.getState().sendMessage("queued follow-up"); + expect(useChatStore.getState().isStreaming).toBe(true); + expect(useChatStore.getState().queue).toHaveLength(1); + + useChatStore.getState().processEvent({ + type: "error", + code: "internal", + message: "Internal error occurred.", + 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."); + const state = useChatStore.getState(); + expect(state.isStreaming).toBe(true); + expect(state.queue).toHaveLength(1); + expect(state.pendingInput).not.toBeNull(); + expect(state.messages.at(-1)?.inlineError).toBeUndefined(); + + // The genuine terminal still completes the turn and flushes the queue. + useChatStore.getState().processEvent({ type: "stream_complete", result: { status: "finished" } }); + expect(useChatStore.getState().isStreaming).toBe(false); + await vi.waitFor(() => { + expect(boundary.streamChat).toHaveBeenCalledTimes(2); + }); + }); +}); diff --git a/apps/vscode/webview-ui/src/stores/chat.store.ts b/apps/vscode/webview-ui/src/stores/chat.store.ts index 19f125ca55..9ec7691860 100644 --- a/apps/vscode/webview-ui/src/stores/chat.store.ts +++ b/apps/vscode/webview-ui/src/stores/chat.store.ts @@ -3,6 +3,7 @@ import { produce } from "immer"; import { bridge } from "@/services"; import { Content } from "@/lib/content"; import { useApprovalStore } from "./approval.store"; +import { toast } from "@/components/ui/sonner"; import { useSettingsStore } from "./settings.store"; import { processEvent } from "./event-handlers"; @@ -248,6 +249,14 @@ export const useChatStore = create((set, get) => ({ }, processEvent: (event) => { + // Mid-turn warnings (terminal === false) leave the turn, the composer, and + // the queued messages untouched — the engine is still streaming, so they + // are surfaced as a transient toast only. + if (event.type === "error" && "terminal" in event && event.terminal === false) { + clearHandshakeTimer(); + toast.warning(event.message); + return; + } // Clear handshake timeout on receiving valid response if (event.type === "TurnBegin" || event.type === "StepBegin" || event.type === "ContentPart") { clearHandshakeTimer();