smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge PR #8392: fix(lsp): isolate document overlays across sessions (@roboomp)

Browse Source
This commit is contained in:
can1357
2026-08-13 05:48:27 +02:00
parent d8924d055b d1278762f8
commit 0af778ea54
5 changed files with 101 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/coding-agent/CHANGELOG.md
+3
View File
@@ -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
packages/coding-agent/src/lsp/mux/protocol.ts
+12 -13
View File
@@ -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;
}
packages/coding-agent/src/lsp/mux/server.ts
+28 -79
View File
@@ -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<Session, SessionDocumentVersion>;
}
interface ForwardedRequest {
session?: Session;
originalId?: LspJsonRpcId;
@@ -92,7 +82,7 @@ class ServerInstance {
readonly key: string;
readonly proc: ptree.ChildProcess<"pipe">;
readonly sessions = new Set<Session>();
readonly documents = new Map<string, DocumentRecord>();
readonly documents = new Set<string>();
readonly diagnostics = new Map<string, DiagnosticsParams>();
readonly registrations: RegistrationBatch[] = [];
readonly progress = new Map<string | number, ProgressParams>();
@@ -188,7 +178,7 @@ function cloneParams<T>(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<string, ServerInstance>();
readonly #servers = new Set<ServerInstance>();
readonly #sessions = new Set<Session>();
#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<void> {
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<void> {
await this.#writeServer(server, message);
}
async #didClose(session: Session, server: ServerInstance, message: LspJsonRpcNotification): Promise<void> {
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<void> | 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();
packages/coding-agent/test/fixtures/fake-lsp-server.ts
Vendored
+5
View File
@@ -102,6 +102,11 @@ async function handleRequest(message: JsonRpcMessage): Promise<void> {
});
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;
packages/coding-agent/test/lsp-mux.test.ts
+53 -44
View File
@@ -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<string | null>("test/documentText", { uri }),
second.client.request<string | null>("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<PublishDiagnosticsParams>("textDocument/publishDiagnostics"),
second.client.nextNotification<PublishDiagnosticsParams>("textDocument/publishDiagnostics"),
]);
expect(firstPublish).toMatchObject({
const publication = await first.client.nextNotification<PublishDiagnosticsParams>(
"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<PublishDiagnosticsParams>(
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<PublishDiagnosticsParams>(
"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<string | null>("test/documentText", { uri })).toBe("replacement");
},
10_000,
);