diff --git a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts b/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts index 0aceaf4f26..56906cfc39 100644 --- a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts +++ b/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 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 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 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(); + }); }); diff --git a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts b/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts index 563038a1ad..b6fb5cccdb 100644 --- a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts +++ b/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; + unlistenFn: UnlistenFn | null; + registrationPromise: Promise | 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(); + +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 { + const registration = listen(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) | null = null; private initPromise: Promise | null = null; private listenerRegistrationPromises = new Set>(); + private eventSubscriptions = new Set(); supportsSearchStreamEvents(): boolean { return true; @@ -125,69 +227,62 @@ export class TauriTransportAdapter implements ITransportAdapter { } listen(event: string, callback: (data: T) => void): () => void { - let unlistenFn: UnlistenFn | null = null; - let isUnlistened = false; - - const registration = listen(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 { + 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 { - 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; }