From f34b2ecfb01dc194aaa3f209a00772991caad3f3 Mon Sep 17 00:00:00 2001 From: Grapedge Date: Thu, 27 Aug 2026 15:19:24 +0800 Subject: [PATCH] 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. --- .../fix-vscode-duplicated-stream-events.md | 5 ++ apps/vscode/src/runtime/kimi-runtime.ts | 42 ++++++++- apps/vscode/test/kimi-runtime.test.ts | 85 +++++++++++++++++++ 3 files changed, 129 insertions(+), 3 deletions(-) create mode 100644 .changeset/fix-vscode-duplicated-stream-events.md diff --git a/.changeset/fix-vscode-duplicated-stream-events.md b/.changeset/fix-vscode-duplicated-stream-events.md new file mode 100644 index 000000000..5bd16394d --- /dev/null +++ b/.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. diff --git a/apps/vscode/src/runtime/kimi-runtime.ts b/apps/vscode/src/runtime/kimi-runtime.ts index f3c6db4c6..e96af1316 100644 --- a/apps/vscode/src/runtime/kimi-runtime.ts +++ b/apps/vscode/src/runtime/kimi-runtime.ts @@ -56,6 +56,7 @@ export class KimiRuntime { private readonly log: KimiRuntimeOptions["log"]; private readonly sessions = new Map(); private readonly sessionByView = new Map(); + private readonly viewChains = new Map>(); private closed = false; constructor(options: KimiRuntimeOptions) { @@ -86,6 +87,10 @@ export class KimiRuntime { } async openSession(options: OpenSessionOptions): Promise { + return this.serializeView(options.webviewId, () => this.openSessionInner(options)); + } + + private async openSessionInner(options: OpenSessionOptions): Promise { 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 { + return this.serializeView(webviewId, () => + this.attachResumedSessionInner(webviewId, session, defaultYoloMode), + ); + } + + private async attachResumedSessionInner( + webviewId: string, + session: Session, + defaultYoloMode: boolean, ): Promise { 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 { + return this.serializeView(webviewId, () => this.detachViewInner(webviewId)); + } + + private async detachViewInner(webviewId: string): Promise { 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(webviewId: string, work: () => Promise): Promise { + 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 { const runtime = this.sessions.get(id); if (runtime === undefined) { diff --git a/apps/vscode/test/kimi-runtime.test.ts b/apps/vscode/test/kimi-runtime.test.ts index 00474c429..108e6fd49 100644 --- a/apps/vscode/test/kimi-runtime.test.ts +++ b/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" });