fix(peer): route Tauri events once

src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts

@@ -1,19 +1,29 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { isExpectedTauriRequestError, TauriTransportAdapter } from './tauri-adapter';
+import {
+  beginRuntimeSessionAttachment,
+  resetRuntimeSessionEventGateForTest,
+  RUNTIME_EVENT_CURSOR_KEY,
+  RUNTIME_EVENT_STREAM_ID_KEY,
+} from '@/infrastructure/peer-device/runtimeSessionEventGate';
+import { setActiveSurfaceDeviceId } from '@/infrastructure/peer-device/deviceSurfaceRouting';
 
 const invokeMock = vi.hoisted(() => vi.fn());
+const listenMock = vi.hoisted(() => vi.fn());
 
 vi.mock('@tauri-apps/api/core', () => ({
   invoke: invokeMock,
 }));
 
 vi.mock('@tauri-apps/api/event', () => ({
-  listen: vi.fn(),
+  listen: listenMock,
 }));
 
 describe('Tauri adapter expected errors', () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    resetRuntimeSessionEventGateForTest();
+    setActiveSurfaceDeviceId(null);
   });
 
   it('classifies optional get_config not found as expected', () => {
@@ -82,4 +92,118 @@ describe('Tauri adapter expected errors', () => {
     expect(timing.invokeDurationMs).toEqual(expect.any(Number));
     expect(timing.transportDurationMs).toEqual(expect.any(Number));
   });
+
+  it('routes a positioned Tauri event once across adapter instances', async () => {
+    const handlers = new Map<string, (event: { payload: unknown }) => void>();
+    const underlyingUnlisten = vi.fn();
+    listenMock.mockImplementation(async (event: string, handler: (event: { payload: unknown }) => void) => {
+      handlers.set(event, handler);
+      return underlyingUnlisten;
+    });
+
+    const firstAdapter = new TauriTransportAdapter();
+    const secondAdapter = new TauriTransportAdapter();
+    const first = vi.fn();
+    const second = vi.fn();
+    const unlistenFirst = firstAdapter.listen('agentic://dialog-turn-completed', first);
+    await firstAdapter.waitForListenerRegistrations();
+    const unlistenSecond = secondAdapter.listen('agentic://dialog-turn-completed', second);
+    await secondAdapter.waitForListenerRegistrations();
+
+    expect(listenMock).toHaveBeenCalledTimes(1);
+    handlers.get('agentic://dialog-turn-completed')?.({
+      payload: {
+        sessionId: 'session-1',
+        turnId: 'turn-1',
+        success: true,
+        [RUNTIME_EVENT_STREAM_ID_KEY]: 'runtime-1',
+        [RUNTIME_EVENT_CURSOR_KEY]: 14,
+      },
+    });
+
+    const expectedPayload = {
+      sessionId: 'session-1',
+      turnId: 'turn-1',
+      success: true,
+    };
+    expect(first).toHaveBeenCalledOnce();
+    expect(first).toHaveBeenCalledWith(expectedPayload);
+    expect(second).toHaveBeenCalledOnce();
+    expect(second).toHaveBeenCalledWith(expectedPayload);
+
+    unlistenFirst();
+    expect(underlyingUnlisten).not.toHaveBeenCalled();
+    unlistenSecond();
+    expect(underlyingUnlisten).toHaveBeenCalledOnce();
+  });
+
+  it('disconnects only the subscriptions owned by that adapter', async () => {
+    const handlers = new Map<string, (event: { payload: unknown }) => void>();
+    const underlyingUnlisten = vi.fn();
+    listenMock.mockImplementation(async (event: string, handler: (event: { payload: unknown }) => void) => {
+      handlers.set(event, handler);
+      return underlyingUnlisten;
+    });
+
+    const firstAdapter = new TauriTransportAdapter();
+    const secondAdapter = new TauriTransportAdapter();
+    const first = vi.fn();
+    const second = vi.fn();
+    firstAdapter.listen('account://login-state', first);
+    secondAdapter.listen('account://login-state', second);
+    await Promise.all([
+      firstAdapter.waitForListenerRegistrations(),
+      secondAdapter.waitForListenerRegistrations(),
+    ]);
+
+    await firstAdapter.disconnect();
+    expect(underlyingUnlisten).not.toHaveBeenCalled();
+
+    handlers.get('account://login-state')?.({ payload: { logged_in: true } });
+    expect(first).not.toHaveBeenCalled();
+    expect(second).toHaveBeenCalledOnce();
+    expect(second).toHaveBeenCalledWith({ logged_in: true });
+
+    await secondAdapter.disconnect();
+    expect(underlyingUnlisten).toHaveBeenCalledOnce();
+  });
+
+  it('releases a held event to the subscribers that owned it on arrival', async () => {
+    const handlers = new Map<string, (event: { payload: unknown }) => void>();
+    listenMock.mockImplementation(async (event: string, handler: (event: { payload: unknown }) => void) => {
+      handlers.set(event, handler);
+      return vi.fn();
+    });
+
+    const firstAdapter = new TauriTransportAdapter();
+    const first = vi.fn();
+    const unlistenFirst = firstAdapter.listen('agentic://text-chunk', first);
+    await firstAdapter.waitForListenerRegistrations();
+
+    const attachment = beginRuntimeSessionAttachment('local', 'session-1');
+    handlers.get('agentic://text-chunk')?.({
+      payload: {
+        sessionId: 'session-1',
+        text: 'held',
+        [RUNTIME_EVENT_STREAM_ID_KEY]: 'runtime-1',
+        [RUNTIME_EVENT_CURSOR_KEY]: 2,
+      },
+    });
+    unlistenFirst();
+
+    const secondAdapter = new TauriTransportAdapter();
+    const second = vi.fn();
+    const unlistenSecond = secondAdapter.listen('agentic://text-chunk', second);
+    await secondAdapter.waitForListenerRegistrations();
+
+    attachment.finish({ streamId: 'runtime-1', cursor: 1 });
+    expect(first).toHaveBeenCalledOnce();
+    expect(first).toHaveBeenCalledWith({
+      sessionId: 'session-1',
+      text: 'held',
+    });
+    expect(second).not.toHaveBeenCalled();
+
+    unlistenSecond();
+  });
 });

src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts

@@ -11,6 +11,108 @@ import { sanitizeErrorForLog } from '../logSanitizer';
 
 const log = createLogger('TauriAdapter');
 
+interface SharedTauriEventListener {
+  subscriptions: Set<TauriEventSubscription>;
+  unlistenFn: UnlistenFn | null;
+  registrationPromise: Promise<void> | null;
+  closed: boolean;
+}
+
+interface TauriEventSubscription {
+  owner: TauriTransportAdapter;
+  event: string;
+  callback: (data: unknown) => void;
+  listener: SharedTauriEventListener;
+  active: boolean;
+}
+
+// The Tauri event bus is window-wide, while Peer Device Mode keeps multiple
+// transport adapters alive. Route each native event once before fan-out so a
+// positioned Session event cannot consume its cursor once per adapter.
+const sharedTauriEventListeners = new Map<string, SharedTauriEventListener>();
+
+function closeSharedTauriEventListener(
+  event: string,
+  shared: SharedTauriEventListener,
+): void {
+  if (shared.closed) {
+    return;
+  }
+  shared.closed = true;
+  if (sharedTauriEventListeners.get(event) === shared) {
+    sharedTauriEventListeners.delete(event);
+  }
+  if (shared.unlistenFn) {
+    try {
+      shared.unlistenFn();
+    } catch (error) {
+      log.error('Error while unlistening', sanitizeErrorForLog(error));
+    }
+    shared.unlistenFn = null;
+  }
+}
+
+function registerSharedTauriEventListener(
+  event: string,
+  shared: SharedTauriEventListener,
+): Promise<void> {
+  const registration = listen<unknown>(event, (e) => {
+    if (shared.closed) {
+      return;
+    }
+
+    // Capture the logical listeners that owned the event when it arrived. A
+    // Session read may hold delivery after accepting the write; those owners
+    // must still paint it when released, while later subscribers must not
+    // receive the native event retroactively.
+    const subscriptions = [...shared.subscriptions];
+
+    // Peer devices stay attached while the UI renders another device, so
+    // several product event streams share this bus. Only the rendered device
+    // surface may reach product listeners.
+    const route = routeSurfaceEvent(event, e.payload);
+    if (!route.deliver) {
+      return;
+    }
+
+    routeRuntimeSessionEvent(
+      surfaceIdForDevice(route.sourceDeviceId),
+      event,
+      route.payload,
+      payload => {
+        for (const subscription of subscriptions) {
+          try {
+            subscription.callback(payload);
+          } catch (error) {
+            log.error('Error in event listener callback', {
+              event,
+              error: sanitizeErrorForLog(error),
+            });
+          }
+        }
+      },
+    );
+  }).then(fn => {
+    if (shared.closed || sharedTauriEventListeners.get(event) !== shared) {
+      fn();
+    } else {
+      shared.unlistenFn = fn;
+    }
+  }).catch(error => {
+    log.error('Failed to listen event', { event, error: sanitizeErrorForLog(error) });
+    if (sharedTauriEventListeners.get(event) === shared) {
+      sharedTauriEventListeners.delete(event);
+    }
+    shared.closed = true;
+  }).finally(() => {
+    if (shared.registrationPromise === registration) {
+      shared.registrationPromise = null;
+    }
+  });
+  shared.registrationPromise = registration;
+  return registration;
+}
+
 export function isExpectedTauriRequestError(action: string, params: unknown, error: unknown): boolean {
   if (action !== 'get_config') {
     return false;
@@ -31,11 +133,11 @@ export function isExpectedTauriRequestError(action: string, params: unknown, err
 }
 
 export class TauriTransportAdapter implements ITransportAdapter {
-  private unlistenFunctions: UnlistenFn[] = [];
   private connected: boolean = false;
   private invokeFn: ((action: string, params?: any) => Promise<any>) | null = null;
   private initPromise: Promise<void> | null = null;
   private listenerRegistrationPromises = new Set<Promise<void>>();
+  private eventSubscriptions = new Set<TauriEventSubscription>();
 
   supportsSearchStreamEvents(): boolean {
     return true;
@@ -125,69 +227,62 @@ export class TauriTransportAdapter implements ITransportAdapter {
   }
 
   listen<T>(event: string, callback: (data: T) => void): () => void {
-    let unlistenFn: UnlistenFn | null = null;
-    let isUnlistened = false;
-
-    const registration = listen<T>(event, (e) => {
-      if (!isUnlistened) {
-        // Peer devices stay attached while the UI renders another device, so
-        // several product event streams share this bus. Only the rendered
-        // device surface may reach product listeners.
-        const route = routeSurfaceEvent(event, e.payload);
-        if (!route.deliver) {
-          return;
-        }
-        routeRuntimeSessionEvent(
-          surfaceIdForDevice(route.sourceDeviceId),
-          event,
-          route.payload,
-          payload => {
-            try {
-              callback(payload);
-            } catch (error) {
-              log.error('Error in event listener callback', {
-                event,
-                error: sanitizeErrorForLog(error),
-              });
-            }
-          },
-        );
-      }
-    }).then(fn => {
-      if (isUnlistened) {
-        fn();
-      } else {
-        unlistenFn = fn;
-        this.unlistenFunctions.push(fn);
-      }
-    }).catch(error => {
-      log.error('Failed to listen event', { event, error: sanitizeErrorForLog(error) });
-    }).finally(() => {
+    let shared = sharedTauriEventListeners.get(event);
+    let needsRegistration = false;
+    if (!shared) {
+      shared = {
+        subscriptions: new Set(),
+        unlistenFn: null,
+        registrationPromise: null,
+        closed: false,
+      };
+      sharedTauriEventListeners.set(event, shared);
+      needsRegistration = true;
+    }
+
+    const subscription: TauriEventSubscription = {
+      owner: this,
+      event,
+      callback: callback as (data: unknown) => void,
+      listener: shared,
+      active: true,
+    };
+    shared.subscriptions.add(subscription);
+    this.eventSubscriptions.add(subscription);
+
+    const registration = needsRegistration
+      ? registerSharedTauriEventListener(event, shared)
+      : shared.registrationPromise;
+    if (registration) {
+      this.trackListenerRegistration(registration);
+    }
+
+    return () => this.removeEventSubscription(subscription);
+  }
+
+  private trackListenerRegistration(registration: Promise<void>): void {
+    this.listenerRegistrationPromises.add(registration);
+    void registration.finally(() => {
       this.listenerRegistrationPromises.delete(registration);
     });
-    this.listenerRegistrationPromises.add(registration);
+  }
 
-    return () => {
-      isUnlistened = true;
-      if (unlistenFn) {
-        unlistenFn();
-        const index = this.unlistenFunctions.indexOf(unlistenFn);
-        if (index > -1) {
-          this.unlistenFunctions.splice(index, 1);
-        }
-      }
-    };
+  private removeEventSubscription(subscription: TauriEventSubscription): void {
+    if (!subscription.active || subscription.owner !== this) {
+      return;
+    }
+    subscription.active = false;
+    subscription.listener.subscriptions.delete(subscription);
+    this.eventSubscriptions.delete(subscription);
+    if (subscription.listener.subscriptions.size === 0) {
+      closeSharedTauriEventListener(subscription.event, subscription.listener);
+    }
   }
 
   async disconnect(): Promise<void> {
-    this.unlistenFunctions.forEach(fn => {
-      try {
-        fn();
-      } catch (error) {
-        log.error('Error while unlistening', sanitizeErrorForLog(error));
-      }
-    });
-    this.unlistenFunctions = [];
+    for (const subscription of [...this.eventSubscriptions]) {
+      this.removeEventSubscription(subscription);
+    }
     this.connected = false;
   }