fix(vscode): stop mid-turn core errors from corrupting the active turn

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 =

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<PromptResult> {
     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<PromptResult>((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,
     });
   }
 

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);

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>) => 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<void> = 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<void>((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<void>((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<string, unknown>)["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<void>((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<string, unknown>)["terminal"]).toBeUndefined();
+    expect(opened.isBusy).toBe(false);
+  });
 });

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);
+    });
+  });
+});

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<ChatState>((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();