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

.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,14 +152,24 @@ 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) {
       existing.subscribe(webviewId);
       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" });