From d8b921fe66e7cf5c247390bf301a83eaae1315b7 Mon Sep 17 00:00:00 2001 From: wsp Date: Tue, 18 Aug 2026 15:14:31 +0800 Subject: [PATCH] fix(peer): route Tauri events once The regression was introduced by Bob Lee in commit 14c4b5c7301818758d9a43b06672e5b912be3db9, which moved the stateful runtime-session cursor routing into every logical Tauri listener. The first subscriber advanced SessionStream, causing later subscribers to drop the same terminal event. Share each native Tauri listener across all adapter instances so surface and cursor routing runs once before logical subscriber fan-out. Keep subscription ownership isolated per adapter and preserve held-event delivery semantics. Add regressions for cross-adapter fan-out, adapter-specific disconnect, and attachment-held events. Refs: 14c4b5c7301818758d9a43b06672e5b912be3db9 --- .../api/adapters/tauri-adapter.test.ts | 126 ++++++++++- .../api/adapters/tauri-adapter.ts | 211 +++++++++++++----- 2 files changed, 278 insertions(+), 59 deletions(-) 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; }