From 6839ebc27dd55cc9bcef944061aa788e7fd0228e Mon Sep 17 00:00:00 2001 From: Matthias Goergens Date: Tue, 11 Aug 2026 11:39:53 +0800 Subject: [PATCH] fix(vscode): reuse the existing runtime when session opens race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two concurrent opens of the same session (sidebar + editor tab, or a reload overlapping a reattach) both passed the sessions.get check, resumed, and wrapped: wrapSession unconditionally overwrote the map entry, orphaning the first SessionRuntime with its event subscription still live. Both runtimes then adapted and broadcast every SDK event, and a view subscribed in both received each streamed part twice — visible as interleaved duplicated assistant text in the chat. wrapSession now returns the existing runtime when the session already has one. A resumed Session handle is inert until wrapped (its constructor registers nothing; listeners and approval/question handlers are only installed by the SessionRuntime constructor), so the loser's handle can simply be dropped — and must not be closed, since close() would tear down the shared engine session. Fixes #2799 --- .changeset/vscode-duplicate-session-wrap.md | 5 ++ apps/vscode/src/runtime/kimi-runtime.ts | 77 ++++++++++++++++--- apps/vscode/test/kimi-runtime.test.ts | 83 ++++++++++++++++++++- 3 files changed, 154 insertions(+), 11 deletions(-) create mode 100644 .changeset/vscode-duplicate-session-wrap.md diff --git a/.changeset/vscode-duplicate-session-wrap.md b/.changeset/vscode-duplicate-session-wrap.md new file mode 100644 index 0000000000..4698302b3b --- /dev/null +++ b/.changeset/vscode-duplicate-session-wrap.md @@ -0,0 +1,5 @@ +--- +"kimi-code": patch +--- + +Fix duplicated, interleaved assistant output (e.g. "TheThe roaring roaring") when two views race to open the same session. A concurrent open/attach used to wrap the session in a second `SessionRuntime` whose event subscription was never cleaned up, so every streamed part was broadcast twice; the later open now reuses the existing runtime. diff --git a/apps/vscode/src/runtime/kimi-runtime.ts b/apps/vscode/src/runtime/kimi-runtime.ts index d07af86f82..2a8a51cbfd 100644 --- a/apps/vscode/src/runtime/kimi-runtime.ts +++ b/apps/vscode/src/runtime/kimi-runtime.ts @@ -109,6 +109,7 @@ export class KimiRuntime { metadata: legacyApprovalMetadata(defaultApproval), }) : await this.harness.resumeSession({ id: requestedId, includeSubagents: true }); + let wrapped: { runtime: SessionRuntime; reused: boolean } | undefined; try { assertSessionWorkDir(session, options.workDir); const storedApproval = readLegacyApprovalFlags(session.summary?.metadata); @@ -120,11 +121,20 @@ export class KimiRuntime { } await applySessionSettings(session, options, approval); await this.detachView(options.webviewId); - runtime = this.wrapSession(session, approval); + wrapped = this.wrapSession(session, approval); + runtime = wrapped.runtime; + if (wrapped.reused) { + await this.reconcileWrappedApproval(session, runtime); + } } catch (error) { - await session.close().catch((closeError: unknown) => { - this.log("Failed to close a rejected session", closeError); - }); + // When the wrap was reused, `session` is the losing race handle: + // closing it would close the shared engine session out from under the + // surviving runtime and any active turn. + if (wrapped?.reused !== true) { + await session.close().catch((closeError: unknown) => { + this.log("Failed to close a rejected session", closeError); + }); + } throw error; } } @@ -149,6 +159,7 @@ export class KimiRuntime { await this.detachView(webviewId); let runtime = existing ?? this.sessions.get(session.id); if (runtime === undefined) { + let wrapped: { runtime: SessionRuntime; reused: boolean } | undefined; try { const storedApproval = readLegacyApprovalFlags(session.summary?.metadata); const restoredApproval = @@ -162,11 +173,20 @@ export class KimiRuntime { const status = await session.getStatus(); const permission = corePermissionForLegacyApproval(approval); if (status.permission !== permission) await session.setPermission(permission); - runtime = this.wrapSession(session, approval); + wrapped = this.wrapSession(session, approval); + runtime = wrapped.runtime; + if (wrapped.reused) { + await this.reconcileWrappedApproval(session, runtime); + } } catch (error) { - await session.close().catch((closeError: unknown) => { - this.log("Failed to close a rejected session", closeError); - }); + // When the wrap was reused, `session` is the losing race handle: + // closing it would close the shared engine session out from under the + // surviving runtime and any active turn. + if (wrapped?.reused !== true) { + await session.close().catch((closeError: unknown) => { + this.log("Failed to close a rejected session", closeError); + }); + } throw error; } } @@ -222,7 +242,27 @@ export class KimiRuntime { await this.harness.close(); } - private wrapSession(session: Session, legacyApproval: LegacyApprovalFlags): SessionRuntime { + private wrapSession( + session: Session, + legacyApproval: LegacyApprovalFlags, + ): { runtime: SessionRuntime; reused: boolean } { + // Two views can race opening the same session (sidebar + editor tab, or a + // reload overlapping a reattach): both pass the `sessions.get` check in + // openSession/attachResumedSession, both resume, and without this guard + // the later call would overwrite the earlier runtime here — orphaning it + // with its event subscription still live, so every streamed part reaches + // the shared view twice (interleaved duplicated text in the UI). A + // resumed Session handle is inert until wrapped (its constructor + // registers nothing), so the loser's handle can simply be dropped. + // + // The loser may already have pushed its own approval state (metadata, + // permission) onto the shared engine session; when `reused` is true, + // callers must run reconcileWrappedApproval so the engine and the + // surviving runtime agree again. + const existing = this.sessions.get(session.id); + if (existing !== undefined) { + return { runtime: existing, reused: true }; + } const runtime = new SessionRuntime({ session, legacyApproval, @@ -231,7 +271,24 @@ export class KimiRuntime { log: this.log, }); this.sessions.set(session.id, runtime); - return runtime; + return { runtime, reused: false }; + } + + /** + * Re-assert a surviving runtime's approval state on the engine session + * after wrapSession reused it: a racing open that lost may already have + * written its own (possibly different) approval flags to the session + * metadata and permission before reaching the reuse guard. + */ + private async reconcileWrappedApproval( + session: Session, + runtime: SessionRuntime, + ): Promise { + const flags = runtime.legacyApprovalFlags; + const status = await session.getStatus(); + const permission = corePermissionForLegacyApproval(flags); + if (status.permission !== permission) await session.setPermission(permission); + await session.updateMetadata(legacyApprovalMetadata(flags)); } private async readMigratedLegacyApproval( diff --git a/apps/vscode/test/kimi-runtime.test.ts b/apps/vscode/test/kimi-runtime.test.ts index 6a86f7f2d1..725e3b5bab 100644 --- a/apps/vscode/test/kimi-runtime.test.ts +++ b/apps/vscode/test/kimi-runtime.test.ts @@ -20,10 +20,14 @@ import type { SessionSummary, ThinkingEffort, } from "@moonshot-ai/kimi-code-sdk"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { Events } from "../shared/bridge"; import { KimiRuntime, type OpenSessionOptions } from "../src/runtime/kimi-runtime"; +import { + corePermissionForLegacyApproval, + legacyApprovalMetadata, +} from "../src/runtime/legacy-approval"; interface FakeSessionBoundary { readonly session: Session; @@ -579,6 +583,83 @@ describe("Kimi runtime (owns shared SDK sessions for Webviews)", () => { return { runtime, sdk, broadcasts }; } + it("deduplicates concurrent opens of the same session", async () => { + const { runtime, sdk, broadcasts } = createRecordingRuntime(); + const boundary = sdk.addSession("s1", "/workspace"); + + // Sidebar and editor tab racing to open the same session must end up on a + // single SessionRuntime: a second wrap would double-subscribe the event + // stream and broadcast every streamed part twice. + const [a, b] = await Promise.all([ + runtime.openSession(openOptions({ webviewId: "view-1", sessionId: "s1" })), + runtime.openSession(openOptions({ webviewId: "view-2", sessionId: "s1" })), + ]); + + expect(a).toBe(b); + expect(boundary.subscriptionCount()).toBe(1); + + broadcasts.length = 0; + boundary.emit({ + type: "assistant.delta", + agentId: "main", + sessionId: "s1", + delta: "hello", + } as unknown as Event); + + // One adapted ContentPart per subscribed view, never two. + const parts = broadcasts.filter( + ({ data }) => (data as { type?: string }).type === "ContentPart", + ); + expect(parts).toHaveLength(2); + }); + + it("reconciles approval state when a racing open loses with different settings", async () => { + const { runtime, sdk } = createRecordingRuntime(); + const boundary = sdk.addSession("s1", "/workspace"); + + // The loser of the wrap race may already have written its own yoloMode to + // the engine session before the reuse guard fired; the surviving + // runtime's flags must be what the session is left with. + const [a, b] = await Promise.all([ + runtime.openSession(openOptions({ webviewId: "view-1", sessionId: "s1", yoloMode: false })), + runtime.openSession(openOptions({ webviewId: "view-2", sessionId: "s1", yoloMode: true })), + ]); + + expect(a).toBe(b); + const expected = corePermissionForLegacyApproval(a.legacyApprovalFlags); + expect(boundary.setPermissions.at(-1)).toBe(expected); + expect(boundary.metadataUpdates.at(-1)).toEqual( + legacyApprovalMetadata(a.legacyApprovalFlags), + ); + }); + + it("does not close the shared session when the racing open's reconcile fails", async () => { + const { runtime, sdk } = createRecordingRuntime(); + const boundary = sdk.addSession("s1", "/workspace"); + + // The third metadata write is always the losing call's reconcile (both + // racers write once before wrapping; the winner writes nothing after). + const realUpdate = boundary.session.updateMetadata.bind(boundary.session); + let writes = 0; + vi.spyOn(boundary.session, "updateMetadata").mockImplementation(async (patch: JsonObject) => { + writes += 1; + if (writes === 3) throw new Error("transient metadata failure"); + return realUpdate(patch); + }); + + await expect( + Promise.all([ + runtime.openSession(openOptions({ webviewId: "view-1", sessionId: "s1", yoloMode: false })), + runtime.openSession(openOptions({ webviewId: "view-2", sessionId: "s1", yoloMode: true })), + ]), + ).rejects.toThrow("transient metadata failure"); + + // The losing handle's failure must not close the shared engine session + // out from under the surviving runtime. + expect(boundary.closeCount()).toBe(0); + expect(runtime.getSession("s1")).toBeDefined(); + }); + it("fails a reentrant prompt without disturbing the running turn", async () => { const { runtime, sdk, broadcasts } = createRecordingRuntime(); const opened = await runtime.openSession(openOptions());