From 798b546e73e47f62c3adef7040dc95b0757a082e Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:21:07 +0000 Subject: [PATCH] fix(omp): release closed sessions retained by mount notices Keep per-session notice identity without keeping the session and its runtime reachable from the client-wide ledger. Co-authored-by: luke --- .../src/server/agent/providers/omp/agent.ts | 8 ++- .../agent/providers/omp/mount-notice.test.ts | 24 +++++++- .../omp/test-utils/mount-notice-memory.ts | 60 +++++++++++++++++++ 3 files changed, 88 insertions(+), 4 deletions(-) create mode 100644 packages/server/src/server/agent/providers/omp/test-utils/mount-notice-memory.ts diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 4b17502e2a..22d7c27f81 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -734,6 +734,8 @@ export class OmpAgentSession implements AgentSession { private readonly mountNotices: OmpMountNoticeLedgers; private readonly agentId: string | undefined; private readonly sessionMountNotices = new OmpMountNoticeLedger(); + // The client-wide ledger outlives this session; retain identity without its runtime resources. + private readonly mountNoticeDeliverer = {}; private state: OmpSessionState; private currentModeId: string | null; private runtimeDead = false; @@ -1019,8 +1021,8 @@ export class OmpAgentSession implements AgentSession { async *streamHistory(): AsyncGenerator { // This history seeds a fresh projection: it keeps the mount notices this session already // delivered into it, and none from earlier sessions. - if (this.agentId) this.mountNotices.seedFromHistory(this.agentId, this); - else this.sessionMountNotices.retainDeliveredBy(this); + if (this.agentId) this.mountNotices.seedFromHistory(this.agentId, this.mountNoticeDeliverer); + else this.sessionMountNotices.retainDeliveredBy(this.mountNoticeDeliverer); yield* streamOmpHistory({ sessionFile: this.state.sessionFile, runtimeSession: this.runtimeSession, @@ -1049,7 +1051,7 @@ export class OmpAgentSession implements AgentSession { if (!notice || this.subscribers.size === 0) return false; const ledger = this.mountNoticeLedger(); if (!ledger.announces(notice)) return true; - ledger.recordDelivered(notice, this); + ledger.recordDelivered(notice, this.mountNoticeDeliverer); return false; } diff --git a/packages/server/src/server/agent/providers/omp/mount-notice.test.ts b/packages/server/src/server/agent/providers/omp/mount-notice.test.ts index 4c3f8abaa0..ce16cbd0c0 100644 --- a/packages/server/src/server/agent/providers/omp/mount-notice.test.ts +++ b/packages/server/src/server/agent/providers/omp/mount-notice.test.ts @@ -1,4 +1,6 @@ import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { execFileSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { setImmediate as waitForImmediate } from "node:timers/promises"; @@ -162,6 +164,19 @@ describe("parseOmpMountNotice", () => { }); describe("OMP xd:// mount notices", () => { + test("retaining announced mounts does not retain closed OMP sessions", () => { + const fixture = fileURLToPath(new URL("./test-utils/mount-notice-memory.ts", import.meta.url)); + const output = execFileSync(process.execPath, ["--expose-gc", "--import", "tsx", fixture], { + encoding: "utf8", + timeout: 20_000, + }); + expect(JSON.parse(output)).toEqual({ + closedWithoutNoticeRetained: false, + closedWithNoticeRetained: false, + noticesAfterRelaunch: 1, + }); + }); + test("a relaunched OMP process does not repeat a mount set the timeline already shows", async () => { const omp = await harness(); await omp.announce(BROWSER_AND_LINEAR); @@ -324,7 +339,14 @@ describe("OMP xd:// mount notices", () => { class DaemonTimelineHarness { readonly omp = new FakeOmp(); readonly manager = new AgentManager({ - clients: { omp: new OmpAgentClient({ logger: createTestLogger(), runtime: this.omp }) }, + clients: { + omp: new OmpAgentClient({ + logger: createTestLogger(), + runtime: this.omp, + // FakeOmp supplies the runtime; the manager still checks command availability. + runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } }, + }), + }, logger: createTestLogger(), }); readonly live: Array<{ diff --git a/packages/server/src/server/agent/providers/omp/test-utils/mount-notice-memory.ts b/packages/server/src/server/agent/providers/omp/test-utils/mount-notice-memory.ts new file mode 100644 index 0000000000..6ae2706976 --- /dev/null +++ b/packages/server/src/server/agent/providers/omp/test-utils/mount-notice-memory.ts @@ -0,0 +1,60 @@ +import { setImmediate as nextTask } from "node:timers/promises"; +import pino from "pino"; +import { OmpAgentClient } from "../agent.js"; +import { FakeOmp } from "./fake-omp.js"; + +if (!global.gc) throw new Error("Run with --expose-gc"); +const runtime = new FakeOmp(); +const client = new OmpAgentClient({ logger: pino({ level: "silent" }), runtime }); +const notices: string[] = []; +const notice = { + type: "notice", + level: "info", + source: "xdev", + message: "xd://: mounted browser", +} as const; +const agentId = "a7969ccf-b200-429a-a36a-e61ac01e59ab"; + +async function closedSession(announce: boolean, id: string) { + const session = await client.createSession( + { provider: "omp", cwd: process.cwd() }, + { agentId: id }, + ); + session.subscribe((event) => { + if (event.type === "timeline" && event.item.type === "notification") { + notices.push(event.item.message); + } + }); + if (announce) runtime.latestSession().emit(notice); + await nextTask(); + const persistence = session.describePersistence(); + if (!persistence) throw new Error("Session has no persistence handle"); + await session.close(); + return { reference: new WeakRef(session), persistence }; +} + +const withoutNotice = await closedSession(false, "cd39924e-3984-4d10-b2ec-8a4e4ba029f5"); +const withNotice = await closedSession(true, agentId); +// Move to another job before collecting: WeakRef targets survive their creation job. +for (let i = 0; i < 5; i++) { + await nextTask(); + global.gc(); +} +const retained = { + closedWithoutNoticeRetained: withoutNotice.reference.deref() !== undefined, + closedWithNoticeRetained: withNotice.reference.deref() !== undefined, +}; + +const resumed = await client.resumeSession(withNotice.persistence, undefined, { agentId }); +try { + resumed.subscribe((event) => { + if (event.type === "timeline" && event.item.type === "notification") { + notices.push(event.item.message); + } + }); + runtime.latestSession().emit(notice); + await nextTask(); + console.log(JSON.stringify({ ...retained, noticesAfterRelaunch: notices.length })); +} finally { + await resumed.close(); +}