test(omp): reproduce lost mount notices during rewind

A notice staged while rewind finishes is discarded by the manager but remembered by the mount ledger, hiding the next runtime announcement. This regression is intentionally red pending shared rewind event delivery.

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<string, OmpBridgedToolIdentity>,
+    private readonly mountNotices: OmpMountNoticeLedgers,
+    private readonly agentId: string | undefined,
   ) {
     this.provider = provider;
   }
@@ -641,6 +652,8 @@ class OmpHistorySession implements AgentSession {
   }
 
   async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
+    // 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,11 @@ export class OmpAgentSession implements AgentSession {
   private readonly subagentIndex = new OmpSubagentIndex();
   private readonly subagentCardTracker: OmpSubagentCardTracker;
   private lastTodoItem: Extract<AgentTimelineItem, { type: "todo" }> | null = null;
+  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;
@@ -753,6 +771,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 +1019,10 @@ export class OmpAgentSession implements AgentSession {
   }
 
   async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
+    // 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.mountNoticeDeliverer);
+    else this.sessionMountNotices.retainDeliveredBy(this.mountNoticeDeliverer);
     yield* streamOmpHistory({
       sessionFile: this.state.sessionFile,
       runtimeSession: this.runtimeSession,
@@ -1014,6 +1038,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.mountNoticeDeliverer);
+    return false;
+  }
+
   async getRuntimeInfo(): Promise<AgentRuntimeInfo> {
     await this.refreshState();
     return this.runtimeInfoFromState();
@@ -1170,6 +1211,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<boolean> {
@@ -1718,6 +1762,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 +2294,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 +2395,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 +2426,8 @@ export class OmpAgentClient implements AgentClient {
         sessionFile,
         this.provider,
         persistenceMetadata.bridgedTools,
+        this.mountNotices,
+        launchContext?.agentId,
       );
     }
 
@@ -2417,6 +2470,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) {

packages/server/src/server/agent/providers/omp/mount-notice.test.ts

@@ -0,0 +1,522 @@
+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";
+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<void>> = [];
+
+afterEach(async () => {
+  for (const cleanup of cleanups.splice(0).toReversed()) await cleanup();
+});
+
+async function writeSessionFile(): Promise<string> {
+  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<AgentStreamEvent>): Promise<void> {
+  for await (const _event of history) {
+    // Consumed the way the manager seeds a projection.
+  }
+}
+
+async function closeQuietly(manager: AgentManager, agentId: string): Promise<void> {
+  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<void> {
+    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<void> {
+    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<void> {
+    if (!this.session) throw new Error("No OMP session");
+    await drain(this.session.streamHistory());
+  }
+
+  async readArchivedHistory(agentId = AGENT_ID): Promise<void> {
+    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<MountNoticeHarness> {
+  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("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);
+
+    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,
+        // FakeOmp supplies the runtime; the manager still checks command availability.
+        runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
+      }),
+    },
+    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<void> {
+    await this.manager.resumeAgentFromPersistence(handle, undefined, this.agentId);
+    await this.manager.hydrateTimelineFromProvider(this.agentId);
+  }
+
+  async announce(event: OmpRuntimeEvent | string): Promise<void> {
+    this.omp.latestSession().emit(typeof event === "string" ? notice(event) : event);
+    await waitForImmediate();
+  }
+
+  async prompt(text: string): Promise<void> {
+    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("a mount staged near rewind completion is announced on the next runtime", async () => {
+    const daemon = new DaemonTimelineHarness(AGENT_ID);
+    const handle = persistenceHandle(await writeSessionFile());
+    await daemon.load(handle);
+    const user = daemon.fetchAll().rows.find((row) => row.item.type === "user_message")?.item;
+    if (user?.type !== "user_message" || !user.messageId) throw new Error("No rewind target");
+    const runtime = daemon.omp.latestSession();
+    const before = runtime.getStateRequestCount;
+    let release = runtime.holdStateRequests();
+    const rewind = daemon.manager.rewind(AGENT_ID, user.messageId, "conversation");
+    try {
+      await runtime.waitForStateRequests(before + 1);
+      release();
+      release = runtime.holdStateRequests();
+      await runtime.waitForStateRequests(before + 2);
+      await daemon.announce(BROWSER_AND_LINEAR);
+    } finally {
+      release();
+      await rewind;
+    }
+    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("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]);
+  });
+});

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<string, OmpAnnouncedMount>();
+
+  /** 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<string, OmpMountNoticeLedger>();
+
+  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);
+  }
+}

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();
+}