fix(vscode): serialize per-view session opens to stop duplicated streaming (#3276)

Concurrent openSession/attachResumedSession calls for the same webview
both missed the sessions map before either wrapped the SDK session, so
one Session facade got two SessionRuntimes. The overwritten runtime
leaked and kept broadcasting, doubling every streamed delta and tool
call in the view. Queue opens, attaches, and detaches per webviewId so
the second caller sees the first one's runtime.

.changeset/fix-vscode-duplicated-stream-events.md

@@ -0,0 +1,5 @@
+---
+"kimi-code": patch
+---
+
+Fix streamed replies occasionally showing every character twice and tool calls appearing in duplicate.

apps/vscode/src/runtime/kimi-runtime.ts

@@ -56,6 +56,7 @@ export class KimiRuntime {
   private readonly log: KimiRuntimeOptions["log"];
   private readonly sessions = new Map<string, SessionRuntime>();
   private readonly sessionByView = new Map<string, string>();
+  private readonly viewChains = new Map<string, Promise<void>>();
   private closed = false;
 
   constructor(options: KimiRuntimeOptions) {
@@ -86,6 +87,10 @@ export class KimiRuntime {
   }
 
   async openSession(options: OpenSessionOptions): Promise<SessionRuntime> {
+    return this.serializeView(options.webviewId, () => this.openSessionInner(options));
+  }
+
+  private async openSessionInner(options: OpenSessionOptions): Promise<SessionRuntime> {
     this.ensureOpen();
     const current = this.getSessionForView(options.webviewId);
     const requestedId = options.sessionId ?? current?.id;
@@ -104,7 +109,7 @@ export class KimiRuntime {
     if (runtime !== undefined) {
       assertSessionWorkDir(runtime.session, options.workDir);
       await applySessionSettings(runtime.session, options, runtime.legacyApprovalFlags);
-      await this.detachView(options.webviewId);
+      await this.detachViewInner(options.webviewId);
     } else {
       const defaultApproval: LegacyApprovalFlags = { yolo: options.yoloMode, afk: false };
       const session =
@@ -127,7 +132,7 @@ export class KimiRuntime {
           await session.updateMetadata(legacyApprovalMetadata(approval));
         }
         await applySessionSettings(session, options, approval);
-        await this.detachView(options.webviewId);
+        await this.detachViewInner(options.webviewId);
         runtime = this.wrapSession(session, approval);
       } catch (error) {
         await session.close().catch((closeError: unknown) => {
@@ -147,6 +152,16 @@ export class KimiRuntime {
     webviewId: string,
     session: Session,
     defaultYoloMode = false,
+  ): Promise<SessionRuntime> {
+    return this.serializeView(webviewId, () =>
+      this.attachResumedSessionInner(webviewId, session, defaultYoloMode),
+    );
+  }
+
+  private async attachResumedSessionInner(
+    webviewId: string,
+    session: Session,
+    defaultYoloMode: boolean,
   ): Promise<SessionRuntime> {
     const existing = this.sessions.get(session.id);
     if (existing !== undefined && this.sessionByView.get(webviewId) === session.id) {
@@ -154,7 +169,7 @@ export class KimiRuntime {
       await existing.announceStatus(webviewId);
       return existing;
     }
-    await this.detachView(webviewId);
+    await this.detachViewInner(webviewId);
     let runtime = existing ?? this.sessions.get(session.id);
     if (runtime === undefined) {
       try {
@@ -185,6 +200,10 @@ export class KimiRuntime {
   }
 
   async detachView(webviewId: string): Promise<void> {
+    return this.serializeView(webviewId, () => this.detachViewInner(webviewId));
+  }
+
+  private async detachViewInner(webviewId: string): Promise<void> {
     const id = this.sessionByView.get(webviewId);
     if (id === undefined) return;
     this.sessionByView.delete(webviewId);
@@ -197,6 +216,23 @@ export class KimiRuntime {
     }
   }
 
+  // A view attaches to at most one session, so opens/detaches for one view
+  // must never overlap: concurrent callers that both miss `this.sessions`
+  // would wrap the same SDK session twice and double every streamed event.
+  private serializeView<T>(webviewId: string, work: () => Promise<T>): Promise<T> {
+    const prev = this.viewChains.get(webviewId) ?? Promise.resolve();
+    const run = prev.then(work, work);
+    const next = run.then(
+      () => undefined,
+      () => undefined,
+    );
+    this.viewChains.set(webviewId, next);
+    void next.finally(() => {
+      if (this.viewChains.get(webviewId) === next) this.viewChains.delete(webviewId);
+    });
+    return run;
+  }
+
   async closeSession(id: string): Promise<void> {
     const runtime = this.sessions.get(id);
     if (runtime === undefined) {

apps/vscode/test/kimi-runtime.test.ts

@@ -396,6 +396,91 @@ describe("Kimi runtime (owns shared SDK sessions for Webviews)", () => {
     expect(boundary.handlerInstallations).toEqual({ approval: 1, question: 1 });
   });
 
+  it("does not double-wrap the SDK session when two opens race for it", async () => {
+    const sdk = createFakeHarness();
+    const broadcasts: { event: string; data: unknown; webviewId?: string }[] = [];
+    const runtime = new KimiRuntime({
+      version: "0.6.0",
+      harness: sdk.harness,
+      broadcast: (event, data, webviewId) => {
+        broadcasts.push({ event, data, webviewId });
+      },
+      captureBaseline: () => undefined,
+      log: () => undefined,
+    });
+    const boundary = sdk.addSession("saved-1", "/workspace");
+
+    const [first, second] = await Promise.all([
+      runtime.openSession(openOptions({ sessionId: "saved-1" })),
+      runtime.openSession(openOptions({ sessionId: "saved-1" })),
+    ]);
+
+    expect(second).toBe(first);
+    expect(boundary.subscriptionCount()).toBe(1);
+
+    boundary.emit({
+      type: "assistant.delta",
+      sessionId: "saved-1",
+      agentId: "main",
+      turnId: 1,
+      delta: "Hello",
+    });
+
+    const parts = broadcasts.filter(
+      ({ data }) => (data as { type?: string }).type === "ContentPart",
+    );
+    expect(parts).toHaveLength(1);
+  });
+
+  it("coalesces two concurrent new-session opens for one view onto one session", async () => {
+    const { runtime, sdk } = createRuntime();
+
+    const [first, second] = await Promise.all([
+      runtime.openSession(openOptions()),
+      runtime.openSession(openOptions()),
+    ]);
+
+    expect(second).toBe(first);
+    expect(sdk.createInputs).toHaveLength(1);
+    expect(first.subscribers).toEqual(["view-1"]);
+  });
+
+  it("does not double-wrap the SDK session when two attaches race for it", async () => {
+    const sdk = createFakeHarness();
+    const broadcasts: { event: string; data: unknown; webviewId?: string }[] = [];
+    const runtime = new KimiRuntime({
+      version: "0.6.0",
+      harness: sdk.harness,
+      broadcast: (event, data, webviewId) => {
+        broadcasts.push({ event, data, webviewId });
+      },
+      captureBaseline: () => undefined,
+      log: () => undefined,
+    });
+    const boundary = sdk.addSession("saved-1", "/workspace");
+
+    const [first, second] = await Promise.all([
+      runtime.attachResumedSession("view-1", boundary.session),
+      runtime.attachResumedSession("view-1", boundary.session),
+    ]);
+
+    expect(second).toBe(first);
+    expect(boundary.subscriptionCount()).toBe(1);
+
+    boundary.emit({
+      type: "assistant.delta",
+      sessionId: "saved-1",
+      agentId: "main",
+      turnId: 1,
+      delta: "Hello",
+    });
+
+    const parts = broadcasts.filter(
+      ({ data }) => (data as { type?: string }).type === "ContentPart",
+    );
+    expect(parts).toHaveLength(1);
+  });
+
   it("preserves the resumed session's model instead of reapplying the configured default", async () => {
     const { runtime, sdk } = createRuntime();
     const session = sdk.addSession("saved-1", "/workspace", { model: "old-model" });