diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index c8b44c0bd8..4b17502e2a 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -111,6 +111,11 @@ import { mapOmpRpcUiPermissionRequest, } from "./rpc-ui-permission-mapper.js"; import { DEFAULT_OMP_THINKING_LEVEL, mapOmpModel } from "./map-omp-model.js"; +import { + OmpMountNoticeLedger, + OmpMountNoticeLedgers, + parseOmpMountNotice, +} from "./mount-notice.js"; const OMP_PROVIDER = "omp"; const OMP_CORE_CAPABILITIES: AgentCapabilityFlags = { @@ -208,6 +213,10 @@ interface OmpAgentSessionOptions { noTurnScheduler?: OmpNoTurnScheduler; usagePollScheduler?: OmpUsagePollScheduler; providerIdleDeadlineMs?: number; + /** Shared by every session of the client so a relaunched agent keeps its ledger. */ + mountNotices: OmpMountNoticeLedgers; + /** The Paseo agent whose projection this session feeds; absent for direct client use. */ + agentId?: string; /** * When false (resumed sessions), replayed session events are dropped until * the first prompt or agent_start so history is not re-emitted as live @@ -620,6 +629,8 @@ class OmpHistorySession implements AgentSession { private readonly sessionFile: string, provider: AgentProvider, private readonly bridgedTools: ReadonlyMap, + private readonly mountNotices: OmpMountNoticeLedgers, + private readonly agentId: string | undefined, ) { this.provider = provider; } @@ -641,6 +652,8 @@ class OmpHistorySession implements AgentSession { } async *streamHistory(): AsyncGenerator { + // This history seeds a fresh projection that holds no live mount notices. + if (this.agentId) this.mountNotices.seedFromHistory(this.agentId, this); yield* streamOmpHistory({ sessionFile: this.sessionFile, provider: this.provider, @@ -718,6 +731,9 @@ export class OmpAgentSession implements AgentSession { private readonly subagentIndex = new OmpSubagentIndex(); private readonly subagentCardTracker: OmpSubagentCardTracker; private lastTodoItem: Extract | null = null; + private readonly mountNotices: OmpMountNoticeLedgers; + private readonly agentId: string | undefined; + private readonly sessionMountNotices = new OmpMountNoticeLedger(); private state: OmpSessionState; private currentModeId: string | null; private runtimeDead = false; @@ -753,6 +769,8 @@ export class OmpAgentSession implements AgentSession { this.currentModeId = options.currentModeId ?? null; this.logger = options.logger; this.live = options.live ?? true; + this.mountNotices = options.mountNotices; + this.agentId = options.agentId; this.providerIdleScheduler = options.providerIdleScheduler ?? createOmpProviderIdleScheduler(); this.providerIdleDeadlineMs = options.providerIdleDeadlineMs ?? OMP_PROVIDER_IDLE_DEADLINE_MS; this.noTurnScheduler = options.noTurnScheduler ?? createOmpNoTurnScheduler(); @@ -999,6 +1017,10 @@ 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); yield* streamOmpHistory({ sessionFile: this.state.sessionFile, runtimeSession: this.runtimeSession, @@ -1014,6 +1036,23 @@ export class OmpAgentSession implements AgentSession { } } + private mountNoticeLedger(): OmpMountNoticeLedger { + return this.agentId ? this.mountNotices.forAgent(this.agentId) : this.sessionMountNotices; + } + + /** + * Drops a mount notice whose devices already have that state in the projected timeline. + * A notice no subscriber can receive is not recorded, so it cannot hide a later one. + */ + private isRedundantMountNotice(event: OmpRuntimeEvent): boolean { + const notice = parseOmpMountNotice(event); + if (!notice || this.subscribers.size === 0) return false; + const ledger = this.mountNoticeLedger(); + if (!ledger.announces(notice)) return true; + ledger.recordDelivered(notice, this); + return false; + } + async getRuntimeInfo(): Promise { await this.refreshState(); return this.runtimeInfoFromState(); @@ -1170,6 +1209,9 @@ export class OmpAgentSession implements AgentSession { await this.runtimeSession.branch(target); await this.refreshState(); this.activeToolCalls.clear(); + // A conversation rewind replaces the whole projection, rows this session delivered included. + if (this.agentId) this.mountNotices.forget(this.agentId); + else this.sessionMountNotices.clear(); } async canStopRuntime(): Promise { @@ -1718,6 +1760,10 @@ export class OmpAgentSession implements AgentSession { } return true; } + if (this.isRedundantMountNotice(event)) { + this.logger.debug({ event }, "Dropped OMP mount notice already shown in this timeline"); + return true; + } const mappedEvent = mapOmpRuntimeEventToTimelineItem(event); if (!mappedEvent.handled) { return false; @@ -2246,6 +2292,7 @@ export class OmpAgentClient implements AgentClient { private readonly usagePollScheduler?: OmpUsagePollScheduler; private readonly providerIdleDeadlineMs?: number; private readonly runtime?: OmpRuntime; + private readonly mountNotices = new OmpMountNoticeLedgers(); constructor(options: OmpAgentClientOptions) { const runtimeSettings = mergeOmpRuntimeSettings( @@ -2346,6 +2393,8 @@ export class OmpAgentClient implements AgentClient { noTurnScheduler: this.noTurnScheduler, usagePollScheduler: this.usagePollScheduler, providerIdleDeadlineMs: this.providerIdleDeadlineMs, + mountNotices: this.mountNotices, + agentId: launchContext?.agentId, }); } catch (error) { await hostTools?.close(); @@ -2375,6 +2424,8 @@ export class OmpAgentClient implements AgentClient { sessionFile, this.provider, persistenceMetadata.bridgedTools, + this.mountNotices, + launchContext?.agentId, ); } @@ -2417,6 +2468,8 @@ export class OmpAgentClient implements AgentClient { noTurnScheduler: this.noTurnScheduler, usagePollScheduler: this.usagePollScheduler, providerIdleDeadlineMs: this.providerIdleDeadlineMs, + mountNotices: this.mountNotices, + agentId: launchContext?.agentId, live: false, }); } catch (error) { 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 new file mode 100644 index 0000000000..4c3f8abaa0 --- /dev/null +++ b/packages/server/src/server/agent/providers/omp/mount-notice.test.ts @@ -0,0 +1,473 @@ +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { setImmediate as waitForImmediate } from "node:timers/promises"; +import { afterEach, describe, expect, test } from "vitest"; + +import { createTestLogger } from "../../../../test-utils/test-logger.js"; +import type { + AgentPersistenceHandle, + AgentSession, + AgentStreamEvent, + AgentTimelineItem, +} from "../../agent-sdk-types.js"; +import { AgentManager, type AgentManagerEvent } from "../../agent-manager.js"; +import { OmpAgentClient } from "./agent.js"; +import { parseOmpMountNotice } from "./mount-notice.js"; +import type { OmpRuntimeEvent } from "./rpc-types.js"; +import { FakeOmp } from "./test-utils/fake-omp.js"; + +const AGENT_ID = "8b0c7a52-6f4e-4c4b-9f55-6a5f0f2f6b11"; +const BROWSER_AND_LINEAR = "xd://: mounted browser, mcp__linear__search"; + +type NoticeLevel = "info" | "warning" | "error"; + +function notice( + message: string, + options: { level?: NoticeLevel; source?: string | null } = {}, +): OmpRuntimeEvent { + const source = options.source === undefined ? "xdev" : options.source; + return { + type: "notice", + level: options.level ?? "info", + message, + ...(source === null ? {} : { source }), + }; +} + +function notifications(items: readonly AgentTimelineItem[]): string[] { + return items.flatMap((item) => (item.type === "notification" ? [item.message] : [])); +} + +const cleanups: Array<() => Promise> = []; + +afterEach(async () => { + for (const cleanup of cleanups.splice(0).toReversed()) await cleanup(); +}); + +async function writeSessionFile(): Promise { + const directory = await mkdtemp(join(tmpdir(), "paseo-omp-mount-notice-")); + cleanups.push(() => rm(directory, { recursive: true, force: true })); + const sessionFile = join(directory, "session.jsonl"); + const entries = [ + { type: "session", id: "root", parentId: null }, + { type: "message", id: "u1", parentId: "root", message: { role: "user", content: "hi" } }, + ]; + await writeFile(sessionFile, entries.map((entry) => JSON.stringify(entry)).join("\n"), "utf8"); + return sessionFile; +} + +function persistenceHandle(sessionFile: string): AgentPersistenceHandle { + return { + provider: "omp", + sessionId: "omp-session", + nativeHandle: sessionFile, + metadata: { cwd: process.cwd() }, + }; +} + +async function drain(history: AsyncGenerator): Promise { + for await (const _event of history) { + // Consumed the way the manager seeds a projection. + } +} + +async function closeQuietly(manager: AgentManager, agentId: string): Promise { + await manager.closeAgent(agentId).catch(() => undefined); +} + +/** One Paseo agent whose OMP process the daemon relaunches, as idle collection and reload do. */ +class MountNoticeHarness { + readonly omp = new FakeOmp(); + readonly client = new OmpAgentClient({ logger: createTestLogger(), runtime: this.omp }); + private readonly events: AgentStreamEvent[] = []; + private session: AgentSession | null = null; + + constructor(private readonly sessionFile: string) {} + + handle(): AgentPersistenceHandle { + return persistenceHandle(this.sessionFile); + } + + async relaunch(): Promise { + await this.session?.close(); + const session = await this.client.resumeSession(this.handle(), undefined, { + agentId: AGENT_ID, + }); + this.session = session; + session.subscribe((event) => this.events.push(event)); + cleanups.push(() => session.close()); + } + + async announce(event: OmpRuntimeEvent | string): Promise { + this.omp.latestSession().emit(typeof event === "string" ? notice(event) : event); + await waitForImmediate(); + } + + /** The current session seeds a fresh projection from provider history. */ + async seedFromHistory(): Promise { + if (!this.session) throw new Error("No OMP session"); + await drain(this.session.streamHistory()); + } + + async readArchivedHistory(agentId = AGENT_ID): Promise { + const history = await this.client.resumeSession( + this.handle(), + undefined, + { agentId }, + { + purpose: "history", + }, + ); + await drain(history.streamHistory()); + await history.close(); + } + + notifications(): string[] { + return notifications( + this.events.flatMap((event) => (event.type === "timeline" ? [event.item] : [])), + ); + } +} + +async function harness(): Promise { + const omp = new MountNoticeHarness(await writeSessionFile()); + await omp.relaunch(); + return omp; +} + +describe("parseOmpMountNotice", () => { + test("reads mounted and unmounted devices", () => { + expect(parseOmpMountNotice(notice("xd://: mounted a, b; unmounted c"))).toEqual({ + mounted: ["a", "b"], + unmounted: ["c"], + }); + }); + + test("accepts OMP builds that omit the notice source", () => { + expect(parseOmpMountNotice(notice("xd://: unmounted a", { source: null }))).toEqual({ + mounted: [], + unmounted: ["a"], + }); + }); + + test("leaves other notices alone", () => { + expect(parseOmpMountNotice(notice(BROWSER_AND_LINEAR, { source: "btw" }))).toBeNull(); + expect(parseOmpMountNotice(notice(BROWSER_AND_LINEAR, { level: "warning" }))).toBeNull(); + expect(parseOmpMountNotice(notice(BROWSER_AND_LINEAR, { level: "error" }))).toBeNull(); + expect(parseOmpMountNotice(notice("xd://: rescanned devices"))).toBeNull(); + expect(parseOmpMountNotice(notice("xd://: mounted a, "))).toBeNull(); + expect(parseOmpMountNotice(notice("Session saved"))).toBeNull(); + }); +}); + +describe("OMP xd:// mount notices", () => { + 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); + + await omp.relaunch(); + await omp.announce(BROWSER_AND_LINEAR); + await omp.relaunch(); + await omp.announce("xd://: mounted mcp__linear__search"); + await omp.announce("xd://: mounted browser"); + + expect(omp.notifications()).toEqual([BROWSER_AND_LINEAR]); + }); + + test("a changed mount set is shown as OMP reported it", async () => { + const omp = await harness(); + await omp.announce(BROWSER_AND_LINEAR); + + await omp.relaunch(); + const withGithub = "xd://: mounted browser, mcp__linear__search, mcp__github__search"; + await omp.announce(withGithub); + await omp.announce("xd://: unmounted browser"); + await omp.relaunch(); + await omp.announce("xd://: unmounted browser"); + + expect(omp.notifications()).toEqual([ + BROWSER_AND_LINEAR, + withGithub, + "xd://: unmounted browser", + ]); + }); + + test("a device mounted again after an unmount is shown", async () => { + const omp = await harness(); + await omp.announce(BROWSER_AND_LINEAR); + await omp.announce("xd://: unmounted browser"); + await omp.announce("xd://: mounted browser"); + + expect(omp.notifications()).toEqual([ + BROWSER_AND_LINEAR, + "xd://: unmounted browser", + "xd://: mounted browser", + ]); + }); + + test("identical ordinary notices are all shown", async () => { + const omp = await harness(); + await omp.announce(notice("Session saved", { source: "session-persistence" })); + await omp.relaunch(); + await omp.announce(notice("Session saved", { source: "session-persistence" })); + await omp.announce(notice(BROWSER_AND_LINEAR, { source: "btw-history" })); + await omp.announce(notice(BROWSER_AND_LINEAR, { source: "btw-history" })); + + expect(omp.notifications()).toEqual([ + "Session saved", + "Session saved", + BROWSER_AND_LINEAR, + BROWSER_AND_LINEAR, + ]); + }); + + test("identical xd:// warnings and errors are all shown", async () => { + const omp = await harness(); + await omp.announce(BROWSER_AND_LINEAR); + for (const level of ["warning", "error"] as const) { + await omp.announce(notice(BROWSER_AND_LINEAR, { level })); + await omp.relaunch(); + await omp.announce(notice(BROWSER_AND_LINEAR, { level })); + } + + expect(omp.notifications()).toEqual([ + BROWSER_AND_LINEAR, + BROWSER_AND_LINEAR, + BROWSER_AND_LINEAR, + BROWSER_AND_LINEAR, + BROWSER_AND_LINEAR, + ]); + }); + + test("a projection seeded from history drops mounts earlier sessions showed", async () => { + const omp = await harness(); + await omp.announce(BROWSER_AND_LINEAR); + + await omp.relaunch(); + await omp.seedFromHistory(); + await omp.announce(BROWSER_AND_LINEAR); + + expect(omp.notifications()).toEqual([BROWSER_AND_LINEAR, BROWSER_AND_LINEAR]); + }); + + test("a projection seeded from history keeps mounts its own session already showed", async () => { + const omp = await harness(); + await omp.announce(BROWSER_AND_LINEAR); + await omp.seedFromHistory(); + + await omp.relaunch(); + await omp.announce(BROWSER_AND_LINEAR); + + expect(omp.notifications()).toEqual([BROWSER_AND_LINEAR]); + }); + + test("reading the agent's archived history drops the mounts it no longer shows", async () => { + const omp = await harness(); + await omp.announce(BROWSER_AND_LINEAR); + + await omp.readArchivedHistory(); + await omp.relaunch(); + await omp.announce(BROWSER_AND_LINEAR); + + expect(omp.notifications()).toEqual([BROWSER_AND_LINEAR, BROWSER_AND_LINEAR]); + }); + + test("reading another agent's history of the same session file keeps this agent's mounts", async () => { + const omp = await harness(); + await omp.announce(BROWSER_AND_LINEAR); + + await omp.readArchivedHistory("2d1e4f7a-0c3b-4d8e-a9f6-5b7c8d9e0f12"); + await omp.relaunch(); + await omp.announce(BROWSER_AND_LINEAR); + + expect(omp.notifications()).toEqual([BROWSER_AND_LINEAR]); + }); + + test("a notice nobody received does not hide the next one", async () => { + const omp = new MountNoticeHarness(await writeSessionFile()); + const unobserved = await omp.client.resumeSession(omp.handle(), undefined, { + agentId: AGENT_ID, + }); + cleanups.push(() => unobserved.close()); + await omp.announce(BROWSER_AND_LINEAR); + await unobserved.close(); + + await omp.relaunch(); + await omp.announce(BROWSER_AND_LINEAR); + + expect(omp.notifications()).toEqual([BROWSER_AND_LINEAR]); + }); + + test("an approval mode change keeps the mount set the timeline already shows", async () => { + const omp = new FakeOmp(); + const client = new OmpAgentClient({ logger: createTestLogger(), runtime: omp }); + const session = await client.createSession({ provider: "omp", cwd: process.cwd() }); + cleanups.push(() => session.close()); + const items: AgentTimelineItem[] = []; + session.subscribe((event) => { + if (event.type === "timeline") items.push(event.item); + }); + omp.latestSession().emit(notice(BROWSER_AND_LINEAR)); + await waitForImmediate(); + + await session.setMode("ask"); + omp.latestSession().emit(notice(BROWSER_AND_LINEAR)); + await waitForImmediate(); + + expect(omp.recordedLaunches).toHaveLength(2); + expect(notifications(items)).toEqual([BROWSER_AND_LINEAR]); + }); +}); + +/** A daemon AgentManager driving one OMP agent through FakeOmp. */ +class DaemonTimelineHarness { + readonly omp = new FakeOmp(); + readonly manager = new AgentManager({ + clients: { omp: new OmpAgentClient({ logger: createTestLogger(), runtime: this.omp }) }, + logger: createTestLogger(), + }); + readonly live: Array<{ + epoch: string | undefined; + seq: number | undefined; + item: AgentTimelineItem; + }> = []; + + constructor(readonly agentId: string) { + cleanups.push(() => closeQuietly(this.manager, agentId)); + this.manager.subscribe( + (event: AgentManagerEvent) => { + if (event.type === "agent_stream" && event.event.type === "timeline") { + this.live.push({ epoch: event.epoch, seq: event.seq, item: event.event.item }); + } + }, + { agentId, replayState: false }, + ); + } + + /** What the agent loader does when a chat opens an agent the daemon has not loaded. */ + async load(handle: AgentPersistenceHandle): Promise { + await this.manager.resumeAgentFromPersistence(handle, undefined, this.agentId); + await this.manager.hydrateTimelineFromProvider(this.agentId); + } + + async announce(event: OmpRuntimeEvent | string): Promise { + this.omp.latestSession().emit(typeof event === "string" ? notice(event) : event); + await waitForImmediate(); + } + + async prompt(text: string): Promise { + await this.manager.appendTimelineItem(this.agentId, { type: "user_message", text }); + } + + fetchAll() { + return this.manager.fetchTimeline(this.agentId, { direction: "tail", limit: 0 }); + } + + fetchedNotifications(): string[] { + return notifications(this.fetchAll().rows.map((row) => row.item)); + } +} + +describe("OMP mount notices in the daemon timeline", () => { + test("live delivery and fetched history agree after idle collection relaunches OMP", async () => { + const daemon = new DaemonTimelineHarness(AGENT_ID); + await daemon.load(persistenceHandle(await writeSessionFile())); + await daemon.announce(BROWSER_AND_LINEAR); + await daemon.prompt("first prompt"); + const handle = daemon.manager.getAgent(AGENT_ID)?.persistence; + if (!handle) throw new Error("OMP agent has no persistence handle"); + const epoch = daemon.fetchAll().epoch; + const historyRows = daemon.fetchAll().rows.length - 2; + + await daemon.manager.closeAgent(AGENT_ID); + await daemon.load(handle); + await daemon.announce(BROWSER_AND_LINEAR); + await daemon.announce(notice("Session saved", { source: null })); + await daemon.prompt("second prompt"); + await daemon.manager.flush(); + + const fetched = daemon.fetchAll(); + expect(daemon.omp.recordedLaunches.at(-1)?.session).toBe(handle.nativeHandle); + expect(fetched.epoch).toBe(epoch); + expect(fetched.rows.slice(historyRows).map((row) => row.item)).toEqual([ + { type: "notification", level: "info", message: BROWSER_AND_LINEAR }, + expect.objectContaining({ type: "user_message", text: "first prompt" }), + { type: "notification", level: "info", message: "Session saved" }, + expect.objectContaining({ type: "user_message", text: "second prompt" }), + ]); + const liveRows = daemon.live.filter((entry) => entry.epoch === epoch); + expect(liveRows.map((entry) => entry.item)).toEqual( + fetched.rows.slice(historyRows).map((row) => row.item), + ); + expect(liveRows.map((entry) => entry.seq)).toEqual( + fetched.rows.slice(historyRows).map((row) => row.seqEnd), + ); + const seqs = fetched.rows.map((row) => row.seqEnd); + expect(seqs).toEqual(seqs.map((_seq, index) => index + 1)); + expect(fetched.hasOlder).toBe(false); + expect(fetched.hasNewer).toBe(false); + + const afterFirstPrompt = daemon.manager.fetchTimeline(AGENT_ID, { + direction: "after", + cursor: { epoch, seq: historyRows + 2 }, + limit: 0, + }); + expect(afterFirstPrompt.gap).toBe(false); + expect(afterFirstPrompt.rows.map((row) => row.item)).toEqual( + fetched.rows.slice(historyRows + 2).map((row) => row.item), + ); + }); + + test("a mount notice that lands before history seeding still suppresses the next relaunch", async () => { + const daemon = new DaemonTimelineHarness(AGENT_ID); + const handle = persistenceHandle(await writeSessionFile()); + await daemon.manager.resumeAgentFromPersistence(handle, undefined, AGENT_ID); + await daemon.announce(BROWSER_AND_LINEAR); + await daemon.manager.hydrateTimelineFromProvider(AGENT_ID); + + await daemon.manager.closeAgent(AGENT_ID); + await daemon.load(handle); + await daemon.announce(BROWSER_AND_LINEAR); + + expect(daemon.fetchedNotifications()).toEqual([BROWSER_AND_LINEAR]); + }); + + test("a conversation rewind starts a projection that shows the mount set once", async () => { + const daemon = new DaemonTimelineHarness(AGENT_ID); + const handle = persistenceHandle(await writeSessionFile()); + await daemon.load(handle); + await daemon.announce(BROWSER_AND_LINEAR); + const userMessage = daemon + .fetchAll() + .rows.map((row) => row.item) + .find((item) => item.type === "user_message"); + if (userMessage?.type !== "user_message" || !userMessage.messageId) { + throw new Error("History has no user message to rewind to"); + } + + await daemon.manager.rewind(AGENT_ID, userMessage.messageId, "conversation"); + expect(daemon.fetchedNotifications()).toEqual([]); + await daemon.manager.closeAgent(AGENT_ID); + await daemon.load(handle); + await daemon.announce(BROWSER_AND_LINEAR); + + expect(daemon.fetchedNotifications()).toEqual([BROWSER_AND_LINEAR]); + }); + + test("reloading from disk starts a projection that shows the mount set once", async () => { + const daemon = new DaemonTimelineHarness(AGENT_ID); + await daemon.load(persistenceHandle(await writeSessionFile())); + await daemon.announce(BROWSER_AND_LINEAR); + const epoch = daemon.fetchAll().epoch; + + await daemon.manager.reloadAgentSession(AGENT_ID, undefined, { rehydrateFromDisk: true }); + await daemon.manager.hydrateTimelineFromProvider(AGENT_ID); + await daemon.announce(BROWSER_AND_LINEAR); + await daemon.manager.reloadAgentSession(AGENT_ID); + await daemon.announce(BROWSER_AND_LINEAR); + + expect(daemon.fetchAll().epoch).not.toBe(epoch); + expect(daemon.fetchedNotifications()).toEqual([BROWSER_AND_LINEAR]); + }); +}); diff --git a/packages/server/src/server/agent/providers/omp/mount-notice.ts b/packages/server/src/server/agent/providers/omp/mount-notice.ts new file mode 100644 index 0000000000..2c3bfd758e --- /dev/null +++ b/packages/server/src/server/agent/providers/omp/mount-notice.ts @@ -0,0 +1,132 @@ +import { OmpNoticeEventSchema } from "./rpc-types.js"; + +// OMP 18.x announces device mounts as an info notice `xd://: mounted a, b; unmounted c` with +// source `xdev`. Each OMP process diffs against the tools it had at startup, so every relaunch of +// the same session (idle collection, reload, mode change) announces again every MCP server that +// connects after OMP's startup wait. A relaunch never reports a device that went missing while +// no process ran, and has no RPC that lists mounted devices, so a device that vanishes and later +// remounts in a new process looks identical to one that was present all along. +const MOUNT_NOTICE_PREFIX = "xd://: "; +const MOUNT_NOTICE_SOURCE = "xdev"; +const MOUNT_NOTICE_LEVEL = "info"; +const MOUNTED_PREFIX = "mounted "; +const UNMOUNTED_PREFIX = "unmounted "; + +export interface OmpMountNotice { + mounted: readonly string[]; + unmounted: readonly string[]; +} + +function parseNames(list: string): string[] | null { + const names = list.split(",").map((name) => name.trim()); + return names.every((name) => name.length > 0) ? names : null; +} + +/** Reads an OMP `notice` event as an xd:// mount change, or null for any other notice. */ +export function parseOmpMountNotice(event: unknown): OmpMountNotice | null { + const parsed = OmpNoticeEventSchema.safeParse(event); + if (!parsed.success) return null; + const { level, message, source } = parsed.data; + if (level !== MOUNT_NOTICE_LEVEL) return null; + if (source !== undefined && source !== MOUNT_NOTICE_SOURCE) return null; + if (!message.startsWith(MOUNT_NOTICE_PREFIX)) return null; + + const mounted: string[] = []; + const unmounted: string[] = []; + for (const segment of message.slice(MOUNT_NOTICE_PREFIX.length).split(";")) { + const part = segment.trim(); + let target: string[]; + let names: string[] | null; + if (part.startsWith(UNMOUNTED_PREFIX)) { + target = unmounted; + names = parseNames(part.slice(UNMOUNTED_PREFIX.length)); + } else if (part.startsWith(MOUNTED_PREFIX)) { + target = mounted; + names = parseNames(part.slice(MOUNTED_PREFIX.length)); + } else { + return null; + } + if (!names) return null; + target.push(...names); + } + return mounted.length > 0 || unmounted.length > 0 ? { mounted, unmounted } : null; +} + +type OmpMountState = "mounted" | "unmounted"; + +/** The session that delivered a timeline row. A ledger keeps only rows its projection holds. */ +export type OmpMountNoticeDeliverer = object; + +interface OmpAnnouncedMount { + state: OmpMountState; + deliverer: OmpMountNoticeDeliverer; +} + +/** The last xd:// state one agent's projected timeline shows for each device. */ +export class OmpMountNoticeLedger { + private readonly announced = new Map(); + + /** Reports whether the notice changes any device state the projection shows. */ + announces(notice: OmpMountNotice): boolean { + return ( + notice.mounted.some((name) => this.announced.get(name)?.state !== "mounted") || + notice.unmounted.some((name) => this.announced.get(name)?.state !== "unmounted") + ); + } + + /** Records a row the deliverer added to the projection; it shows every device it names. */ + recordDelivered(notice: OmpMountNotice, deliverer: OmpMountNoticeDeliverer): void { + for (const name of notice.mounted) this.announced.set(name, { state: "mounted", deliverer }); + for (const name of notice.unmounted) { + this.announced.set(name, { state: "unmounted", deliverer }); + } + } + + /** + * Keeps only rows the deliverer added. A session streams provider history only to seed a fresh + * projection, which holds that history plus the rows this session already delivered into it. + */ + retainDeliveredBy(deliverer: OmpMountNoticeDeliverer): void { + for (const [name, mount] of this.announced) { + if (mount.deliverer !== deliverer) this.announced.delete(name); + } + } + + clear(): void { + this.announced.clear(); + } + + get size(): number { + return this.announced.size; + } +} + +/** + * Mount ledgers keyed by Paseo agent id. The agent's retained projection outlives one OMP + * process and one Paseo session, so its ledger must too. + */ +export class OmpMountNoticeLedgers { + private readonly byAgentId = new Map(); + + forAgent(agentId: string): OmpMountNoticeLedger { + let ledger = this.byAgentId.get(agentId); + if (!ledger) { + ledger = new OmpMountNoticeLedger(); + this.byAgentId.set(agentId, ledger); + } + return ledger; + } + + /** The agent's whole projection is about to be replaced from provider history. */ + forget(agentId: string): void { + this.byAgentId.delete(agentId); + } + + /** The agent's projection is being seeded from provider history by this deliverer. */ + seedFromHistory(agentId: string, deliverer: OmpMountNoticeDeliverer): void { + const ledger = this.byAgentId.get(agentId); + if (!ledger) return; + ledger.retainDeliveredBy(deliverer); + if (ledger.size === 0) this.byAgentId.delete(agentId); + } +}