From 757066e965e97552f1b09342ac596ebb6f5101cb Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 13 Aug 2026 03:03:36 +0000 Subject: [PATCH 1/2] fix(lsp): isolated document overlays across sessions Assigned concurrent mux links separate language-server processes while retaining idle processes for later reuse. Added regression coverage for session-specific open-document content and updated mux lifecycle expectations. Fixes #8371 --- packages/coding-agent/CHANGELOG.md | 4 + packages/coding-agent/src/lsp/mux/protocol.ts | 25 ++--- packages/coding-agent/src/lsp/mux/server.ts | 103 +++++------------- .../test/fixtures/fake-lsp-server.ts | 5 + packages/coding-agent/test/lsp-mux.test.ts | 77 +++++++------ 5 files changed, 84 insertions(+), 130 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index b387e76fd..a4f7618eb 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed concurrent OMP sessions sharing one LSP backend overlay, so each session now receives semantic results for its own open-document content. ([#8371](https://github.com/can1357/oh-my-pi/issues/8371)) + ## [17.2.15] - 2026-08-12 ### Added diff --git a/packages/coding-agent/src/lsp/mux/protocol.ts b/packages/coding-agent/src/lsp/mux/protocol.ts index 91bd80764..cfd5ceaf8 100644 --- a/packages/coding-agent/src/lsp/mux/protocol.ts +++ b/packages/coding-agent/src/lsp/mux/protocol.ts @@ -2,9 +2,9 @@ * Cross-process contract for the broker-owned LSP mux daemon. * * One mux daemon runs per project scope (launched through the same daemon - * broker that owns the shared Chromium and `hub start` processes). It spawns - * each language server once and multiplexes every omp instance in the project - * onto that single server over a local socket. The link speaks plain + * broker that owns the shared Chromium and `hub start` processes). It assigns + * each concurrent OMP link its own language-server process, then retains idle + * processes briefly for reuse by later links. The link speaks plain * Content-Length-framed LSP JSON-RPC after a one-request handshake * ({@link MUX_CONNECT_METHOD}); everything below is shared by the worker * entry (`server.ts`), the client connector (`daemon.ts`), and tests. @@ -46,9 +46,9 @@ export function lspMuxEndpoint(projectDir: string, runtimeDir: string): string { /** * First (and only pre-LSP) request on a fresh connection: binds the link to - * one shared server instance, spawning it on first use. Params: - * {@link MuxConnectParams}, result: {@link MuxConnectResult}. After the - * response the link carries ordinary LSP traffic for that server. + * an idle server instance or spawns one. Params: {@link MuxConnectParams}, + * result: {@link MuxConnectResult}. After the response the link carries + * ordinary LSP traffic for that server. */ export const MUX_CONNECT_METHOD = "omp/muxConnect"; @@ -60,15 +60,14 @@ export const MUX_PING_METHOD = "omp/muxPing"; export const MUX_PING_RESULT = "pong"; /** - * Notification on a bound link: kill the shared server process (all sessions - * on it are disconnected). Sent by `lsp reload` when the generic reload path - * decides the server is wedged — a plain per-session `shutdown`/`exit` is - * intercepted by the mux and would leave the wedged server running for - * every other instance. + * Notification on a bound link: kill that link's server process. Sent by + * `lsp reload` when the generic reload path decides the server is wedged — + * a plain per-session `shutdown`/`exit` is intercepted by the mux so the + * process can linger for reuse. */ export const MUX_RESTART_METHOD = "omp/muxRestartServer"; -/** Handshake parameters identifying (and if needed spawning) a shared server. */ +/** Handshake parameters identifying a reusable server process. */ export interface MuxConnectParams { /** Executable to spawn (the client's `resolvedCommand ?? command`). */ command: string; @@ -86,7 +85,7 @@ export interface MuxConnectResult { key: string; /** True when this handshake spawned the server process. */ spawned: boolean; - /** Pid of the shared server process. */ + /** Pid of the server process assigned to this link. */ pid?: number; } diff --git a/packages/coding-agent/src/lsp/mux/server.ts b/packages/coding-agent/src/lsp/mux/server.ts index 0087ef6e3..6afb08c24 100644 --- a/packages/coding-agent/src/lsp/mux/server.ts +++ b/packages/coding-agent/src/lsp/mux/server.ts @@ -22,16 +22,6 @@ const SHUTDOWN_BUDGET_MS = 2_000; type RpcMessage = LspJsonRpcRequest | LspJsonRpcResponse | LspJsonRpcNotification; -interface SessionDocumentVersion { - clientVersion: number; - serverVersion: number; -} - -interface DocumentRecord { - serverVersion: number; - perSession: Map; -} - interface ForwardedRequest { session?: Session; originalId?: LspJsonRpcId; @@ -92,7 +82,7 @@ class ServerInstance { readonly key: string; readonly proc: ptree.ChildProcess<"pipe">; readonly sessions = new Set(); - readonly documents = new Map(); + readonly documents = new Set(); readonly diagnostics = new Map(); readonly registrations: RegistrationBatch[] = []; readonly progress = new Map(); @@ -188,7 +178,7 @@ function cloneParams(params: T): T { export class LspMuxServer { /** Called after the mux has had no connected sessions for its idle grace period. */ onIdle?: () => void; - readonly #servers = new Map(); + readonly #servers = new Set(); readonly #sessions = new Set(); #netServer?: net.Server; #endpoint?: string; @@ -204,7 +194,7 @@ export class LspMuxServer { /** Keys of currently live shared language-server children. */ get serverKeys(): string[] { - return [...this.#servers.keys()]; + return [...this.#servers].map(server => server.key); } /** Listen for Content-Length framed mux links at a Unix socket or named pipe. */ @@ -235,7 +225,7 @@ export class LspMuxServer { this.#shuttingDown = true; clearTimeout(this.#idleTimer); for (const session of [...this.#sessions]) session.socket.destroy(); - await Promise.all([...this.#servers.values()].map(server => this.#stopServer(server))); + await Promise.all([...this.#servers].map(server => this.#stopServer(server))); const listener = this.#netServer; this.#netServer = undefined; if (listener) { @@ -330,7 +320,7 @@ export class LspMuxServer { return; } const key = muxServerKey(params.command, params.cwd); - let server = this.#servers.get(key); + let server = [...this.#servers].find(candidate => candidate.key === key && candidate.sessions.size === 0); if (server && server.proc.exitCode !== null) { this.#serverExited(server); server = undefined; @@ -432,63 +422,26 @@ export class LspMuxServer { const params = parseDocumentParams(message.params); if (!params) return; const uri = params.textDocument.uri; - const clientVersion = params.textDocument.version; session.openUris.add(uri); - const existing = server.documents.get(uri); - if (!existing) { - server.documents.set(uri, { - serverVersion: clientVersion, - perSession: new Map([[session, { clientVersion, serverVersion: clientVersion }]]), - }); - await this.#writeServer(server, message); - return; - } - existing.serverVersion = Math.max(existing.serverVersion + 1, clientVersion); - existing.perSession.set(session, { clientVersion, serverVersion: existing.serverVersion }); - await this.#writeServer(server, { - jsonrpc: "2.0", - method: "textDocument/didChange", - params: { - textDocument: { uri, version: existing.serverVersion }, - contentChanges: [{ text: params.textDocument.text ?? "" }], - }, - }); + server.documents.add(uri); + await this.#writeServer(server, message); } - async #didChange(session: Session, server: ServerInstance, message: LspJsonRpcNotification): Promise { - const params = parseDocumentParams(message.params); - if (!params) return; - const uri = params.textDocument.uri; - const record = server.documents.get(uri); - if (!record) { - await this.#writeServer(server, message); - return; - } - const clientVersion = params.textDocument.version; - record.serverVersion = Math.max(record.serverVersion + 1, clientVersion); - record.perSession.set(session, { clientVersion, serverVersion: record.serverVersion }); - // Map each client's version stream into the one monotonically increasing server stream. - await this.#writeServer(server, { - ...message, - params: { ...params, textDocument: { ...params.textDocument, version: record.serverVersion } }, - }); + async #didChange(_session: Session, server: ServerInstance, message: LspJsonRpcNotification): Promise { + await this.#writeServer(server, message); } async #didClose(session: Session, server: ServerInstance, message: LspJsonRpcNotification): Promise { const uri = parseUri(message.params); if (!uri) return; session.openUris.delete(uri); - const record = server.documents.get(uri); - if (!record) return; - record.perSession.delete(session); - if (record.perSession.size > 0) return; server.documents.delete(uri); await this.#writeServer(server, message); } #spawnServer(key: string, params: MuxConnectParams): ServerInstance { const server = new ServerInstance(key, params); - this.#servers.set(key, server); + this.#servers.add(server); void this.#readServer(server); server.proc.exited.then( () => this.#serverExited(server), @@ -537,7 +490,7 @@ export class LspMuxServer { const params = parseDiagnostics(message.params); if (!params) return; server.diagnostics.set(params.uri, cloneParams(params)); - for (const session of server.sessions) if (session.initialized) this.#sendDiagnostics(session, server, params); + for (const session of server.sessions) if (session.initialized) this.#sendDiagnostics(session, params); return; } if (message.method === "$/progress") { @@ -641,13 +594,12 @@ export class LspMuxServer { void this.#writeServer(session.server, { ...message, id: pending.serverId }); } - #sendDiagnostics(session: Session, server: ServerInstance, params: DiagnosticsParams): void { - const rewritten = cloneParams(params); - const version = server.documents.get(params.uri)?.perSession.get(session); - if (params.version !== undefined && version && params.version === version.serverVersion) - rewritten.version = version.clientVersion; - else delete rewritten.version; - this.#sendSession(session, { jsonrpc: "2.0", method: "textDocument/publishDiagnostics", params: rewritten }); + #sendDiagnostics(session: Session, params: DiagnosticsParams): void { + this.#sendSession(session, { + jsonrpc: "2.0", + method: "textDocument/publishDiagnostics", + params: cloneParams(params), + }); } #replayState(session: Session, server: ServerInstance): void { @@ -660,7 +612,7 @@ export class LspMuxServer { params: cloneParams(batch), }); } - for (const params of server.diagnostics.values()) this.#sendDiagnostics(session, server, params); + for (const params of server.diagnostics.values()) this.#sendDiagnostics(session, params); for (const params of server.progress.values()) { this.#sendSession(session, { jsonrpc: "2.0", method: "$/progress", params: cloneParams(params) }); } @@ -693,17 +645,12 @@ export class LspMuxServer { if (server) { server.sessions.delete(session); for (const uri of session.openUris) { - const record = server.documents.get(uri); - if (!record) continue; - record.perSession.delete(session); - if (record.perSession.size === 0) { - server.documents.delete(uri); - await this.#writeServer(server, { - jsonrpc: "2.0", - method: "textDocument/didClose", - params: { textDocument: { uri } }, - }); - } + server.documents.delete(uri); + await this.#writeServer(server, { + jsonrpc: "2.0", + method: "textDocument/didClose", + params: { textDocument: { uri } }, + }); } for (const [muxId, pending] of server.pending) { if (pending.session !== session) continue; @@ -722,7 +669,7 @@ export class LspMuxServer { #serverExited(server: ServerInstance): void { server.stopping = true; - if (this.#servers.get(server.key) === server) this.#servers.delete(server.key); + this.#servers.delete(server); if (server.lingerTimer) clearTimeout(server.lingerTimer); server.pending.clear(); for (const session of [...server.sessions]) session.socket.destroy(); diff --git a/packages/coding-agent/test/fixtures/fake-lsp-server.ts b/packages/coding-agent/test/fixtures/fake-lsp-server.ts index 0d83a3ff0..70fecd7d2 100644 --- a/packages/coding-agent/test/fixtures/fake-lsp-server.ts +++ b/packages/coding-agent/test/fixtures/fake-lsp-server.ts @@ -102,6 +102,11 @@ async function handleRequest(message: JsonRpcMessage): Promise { }); break; } + case "test/documentText": { + const params = message.params as { uri: string }; + respond(id, documents.get(params.uri)?.text ?? null); + break; + } case "test/echo": respond(id, message.params); break; diff --git a/packages/coding-agent/test/lsp-mux.test.ts b/packages/coding-agent/test/lsp-mux.test.ts index 9bb18171b..a8dc7438e 100644 --- a/packages/coding-agent/test/lsp-mux.test.ts +++ b/packages/coding-agent/test/lsp-mux.test.ts @@ -211,24 +211,24 @@ describe("LspMuxServer", () => { } it.skipIf(process.platform === "win32")( - "spawns one server and caches its initialize result across links", + "spawns one server per concurrent link", async () => { const first = await link(); const second = await link(); expect(first.connected.spawned).toBe(true); - expect(second.connected.spawned).toBe(false); - expect(second.connected.pid).toBe(first.connected.pid); + expect(second.connected.spawned).toBe(true); + expect(second.connected.pid).not.toBe(first.connected.pid); const [firstInitialize, secondInitialize] = await Promise.all([ initialize(first.client), initialize(second.client), ]); - expect(firstInitialize).toEqual(secondInitialize); const firstInfo = firstInitialize.serverInfo as { version: string }; const secondInfo = secondInitialize.serverInfo as { version: string }; expect(firstInfo.version).toBe(String(first.connected.pid)); - expect(secondInfo.version).toBe(firstInfo.version); + expect(secondInfo.version).toBe(String(second.connected.pid)); expect((await state(first.client)).initializeCount).toBe(1); + expect((await state(second.client)).initializeCount).toBe(1); }, 10_000, ); @@ -260,61 +260,56 @@ describe("LspMuxServer", () => { ); it.skipIf(process.platform === "win32")( - "reference-counts opens and rewrites shared document versions", + "isolates open-document overlays between concurrent sessions", async () => { const first = await link(); const second = await link(); + expect(second.connected.pid).not.toBe(first.connected.pid); await Promise.all([initialize(first.client), initialize(second.client)]); const uri = "file:///shared.ts"; first.client.notify("textDocument/didOpen", { textDocument: { uri, languageId: "typescript", version: 1, text: "first" }, }); - await pollUntil(async () => (await state(first.client)).didOpen[uri] === 1, "first didOpen"); - second.client.notify("textDocument/didOpen", { textDocument: { uri, languageId: "typescript", version: 1, text: "second" }, }); - await pollUntil(async () => { - const snapshot = await state(first.client); - return snapshot.didOpen[uri] === 1 && (snapshot.didChange[uri]?.some(version => version >= 2) ?? false); - }, "second open converted to change"); - first.client.notify("textDocument/didClose", { textDocument: { uri } }); - await first.client.request("test/echo", { barrier: true }); - expect((await state(second.client)).didClose).not.toContain(uri); - second.client.notify("textDocument/didClose", { textDocument: { uri } }); - await pollUntil(async () => (await state(second.client)).didClose.includes(uri), "final didClose"); + await pollUntil(async () => { + const [seenByFirst, seenBySecond] = await Promise.all([ + first.client.request("test/documentText", { uri }), + second.client.request("test/documentText", { uri }), + ]); + return seenByFirst === "first" && seenBySecond === "second"; + }, "session-specific document contents"); }, 10_000, ); it.skipIf(process.platform === "win32")( - "broadcasts diagnostics and replays the cached publication to a new link", + "replays cached diagnostics when an idle server is reused", async () => { const first = await link(); - const second = await link(); - await Promise.all([initialize(first.client), initialize(second.client)]); + await initialize(first.client); const uri = "file:///diagnostics.ts"; first.client.notify("textDocument/didOpen", { textDocument: { uri, languageId: "typescript", version: 1, text: "x" }, }); - const [firstPublish, secondPublish] = await Promise.all([ - first.client.nextNotification("textDocument/publishDiagnostics"), - second.client.nextNotification("textDocument/publishDiagnostics"), - ]); - expect(firstPublish).toMatchObject({ + const publication = await first.client.nextNotification( + "textDocument/publishDiagnostics", + ); + expect(publication).toMatchObject({ uri, version: 1, diagnostics: [{ message: "fake", severity: 2, range: expect.any(Object) }], }); - expect(secondPublish).toMatchObject({ - uri, - diagnostics: [{ message: "fake", severity: 2, range: expect.any(Object) }], - }); - const third = await link(); - await initialize(third.client); - const replay = await third.client.nextNotification( + first.client.destroy(); + await pollUntil(() => Promise.resolve(server.sessionCount === 0), "first session close"); + const second = await link(); + expect(second.connected.spawned).toBe(false); + expect(second.connected.pid).toBe(first.connected.pid); + await initialize(second.client); + const replay = await second.client.nextNotification( "textDocument/publishDiagnostics", ); expect(replay).toMatchObject({ @@ -351,35 +346,39 @@ describe("LspMuxServer", () => { ); it.skipIf(process.platform === "win32")( - "restarts a shared server and disconnects every attached session", + "restarts only the calling session's server", async () => { const first = await link(); const second = await link(); await Promise.all([initialize(first.client), initialize(second.client)]); const firstClosed = first.client.waitForClose(); - const secondClosed = second.client.waitForClose(); first.client.notify(MUX_RESTART_METHOD); - await Promise.all([firstClosed, secondClosed]); + await firstClosed; + expect(await second.client.request<{ alive: boolean }>("test/echo", { alive: true })).toEqual({ alive: true }); const replacement = await link(); expect(replacement.connected.spawned).toBe(true); expect(replacement.connected.pid).not.toBe(first.connected.pid); + expect(replacement.connected.pid).not.toBe(second.connected.pid); }, 10_000, ); it.skipIf(process.platform === "win32")( - "closes orphaned documents when a session drops abruptly", + "closes orphaned documents before reusing an idle server", async () => { const first = await link(); - const second = await link(); - await Promise.all([initialize(first.client), initialize(second.client)]); + await initialize(first.client); const uri = "file:///orphan.ts"; first.client.notify("textDocument/didOpen", { textDocument: { uri, languageId: "typescript", version: 1, text: "orphan" }, }); - await pollUntil(async () => (await state(second.client)).didOpen[uri] === 1, "orphan didOpen"); + await pollUntil(async () => (await state(first.client)).didOpen[uri] === 1, "orphan didOpen"); first.client.destroy(); + await pollUntil(() => Promise.resolve(server.sessionCount === 0), "orphan session close"); + + const second = await link(); + expect(second.connected.spawned).toBe(false); await pollUntil(async () => (await state(second.client)).didClose.includes(uri), "orphan didClose"); }, 10_000, From d1278762f8ff76c4a3488c2f4742aec13347aec6 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 13 Aug 2026 03:15:29 +0000 Subject: [PATCH 2/2] fix(lsp): deferred idle reuse until cleanup Kept closing sessions attached to their server until orphan document closes and pending request cancellations are queued. Added coverage that reopens the last orphan URI after reuse and verifies the replacement overlay remains active. Fixes #8371 --- packages/coding-agent/src/lsp/mux/server.ts | 6 +++-- packages/coding-agent/test/lsp-mux.test.ts | 26 ++++++++++++++------- 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/packages/coding-agent/src/lsp/mux/server.ts b/packages/coding-agent/src/lsp/mux/server.ts index 6afb08c24..3297e1c36 100644 --- a/packages/coding-agent/src/lsp/mux/server.ts +++ b/packages/coding-agent/src/lsp/mux/server.ts @@ -643,21 +643,23 @@ export class LspMuxServer { this.#sessions.delete(session); const server = session.server; if (server) { - server.sessions.delete(session); + let cleanup: Promise | undefined; for (const uri of session.openUris) { server.documents.delete(uri); - await this.#writeServer(server, { + cleanup = this.#writeServer(server, { jsonrpc: "2.0", method: "textDocument/didClose", params: { textDocument: { uri } }, }); } + await cleanup; for (const [muxId, pending] of server.pending) { if (pending.session !== session) continue; pending.drop = true; await this.#writeServer(server, { jsonrpc: "2.0", method: "$/cancelRequest", params: { id: muxId } }); } server.initializeWaiters.delete(session); + server.sessions.delete(session); if (server.sessions.size === 0 && !server.stopping) { server.lingerTimer = setTimeout(() => { if (server.sessions.size === 0) void this.#stopServer(server); diff --git a/packages/coding-agent/test/lsp-mux.test.ts b/packages/coding-agent/test/lsp-mux.test.ts index a8dc7438e..90d9c1cc3 100644 --- a/packages/coding-agent/test/lsp-mux.test.ts +++ b/packages/coding-agent/test/lsp-mux.test.ts @@ -365,21 +365,31 @@ describe("LspMuxServer", () => { ); it.skipIf(process.platform === "win32")( - "closes orphaned documents before reusing an idle server", + "finishes orphan document closes before reusing a server", async () => { const first = await link(); await initialize(first.client); - const uri = "file:///orphan.ts"; - first.client.notify("textDocument/didOpen", { - textDocument: { uri, languageId: "typescript", version: 1, text: "orphan" }, - }); - await pollUntil(async () => (await state(first.client)).didOpen[uri] === 1, "orphan didOpen"); + const uris = Array.from({ length: 128 }, (_, index) => `file:///orphan-${index}.ts`); + for (const uri of uris) { + first.client.notify("textDocument/didOpen", { + textDocument: { uri, languageId: "typescript", version: 1, text: "orphan" }, + }); + } + await first.client.request("test/echo", { barrier: true }); + const firstClosed = first.client.waitForClose(); first.client.destroy(); - await pollUntil(() => Promise.resolve(server.sessionCount === 0), "orphan session close"); + await firstClosed; const second = await link(); expect(second.connected.spawned).toBe(false); - await pollUntil(async () => (await state(second.client)).didClose.includes(uri), "orphan didClose"); + const uri = uris.at(-1); + expect(uri).toBeDefined(); + await initialize(second.client); + second.client.notify("textDocument/didOpen", { + textDocument: { uri, languageId: "typescript", version: 1, text: "replacement" }, + }); + await second.client.request("test/echo", { barrier: true }); + expect(await second.client.request("test/documentText", { uri })).toBe("replacement"); }, 10_000, );