diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 8729b74a2..43c4d8810 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -81,6 +81,9 @@ ### Fixed - Fixed applied LSP `WorkspaceEdit`s leaving already-open language-server document overlays stale for later semantic requests ([#8372](https://github.com/can1357/oh-my-pi/issues/8372)). +### 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 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..3297e1c36 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) }); } @@ -691,26 +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) { - 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); + 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); @@ -722,7 +671,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..90d9c1cc3 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,36 +346,50 @@ 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", + "finishes orphan document closes before reusing a server", async () => { const first = await link(); - const second = await link(); - await Promise.all([initialize(first.client), initialize(second.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 initialize(first.client); + 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(async () => (await state(second.client)).didClose.includes(uri), "orphan didClose"); + await firstClosed; + + const second = await link(); + expect(second.connected.spawned).toBe(false); + 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, );