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