| 1 | import { describe, expect, it, vi } from 'vitest' |
| 2 | import { TypedEventBus } from '../../../src/main/agent-runtime/events/bus' |
| 3 | import { RuntimeEventBridge } from '../../../src/main/ipc/runtime/event-bridge' |
| 4 | |
| 5 | const jobStarted = (overrides: Record<string, unknown> = {}) => ({ |
| 6 | type: 'job.started' as const, |
| 7 | payload: {}, |
| 8 | jobId: 'job-1', |
| 9 | domain: 'generation' as const, |
| 10 | owner: { sessionId: 'session-1' }, |
| 11 | audience: { kind: 'broadcast' as const }, |
| 12 | occurredAt: 1, |
| 13 | ...overrides |
| 14 | }) |
| 15 | |
| 16 | describe('TypedEventBus', () => { |
| 17 | it('filters by domain, owner and subscriber audience', () => { |
| 18 | const bus = new TypedEventBus() |
| 19 | const listener = vi.fn() |
| 20 | bus.subscribe({ domain: 'generation', owner: { sessionId: 'session-1' }, subscriberId: 'window-1' }, listener) |
| 21 | |
| 22 | bus.emit(jobStarted()) |
| 23 | bus.emit( |
| 24 | jobStarted({ audience: { kind: 'requester', subscriberId: 'window-2' } }) |
| 25 | ) |
| 26 | bus.emit( |
| 27 | jobStarted({ audience: { kind: 'requester', subscriberId: 'window-1' } }) |
| 28 | ) |
| 29 | bus.emit(jobStarted({ domain: 'image' })) |
| 30 | |
| 31 | expect(listener).toHaveBeenCalledTimes(2) |
| 32 | }) |
| 33 | |
| 34 | it('delivers owner-audience events only to matching owner subscribers', () => { |
| 35 | const bus = new TypedEventBus() |
| 36 | const matchingOwner = vi.fn() |
| 37 | const otherOwner = vi.fn() |
| 38 | const unscopedSubscriber = vi.fn() |
| 39 | bus.subscribe({ owner: { sessionId: 'session-1' }, subscriberId: 'window-1' }, matchingOwner) |
| 40 | bus.subscribe({ owner: { sessionId: 'session-2' }, subscriberId: 'window-2' }, otherOwner) |
| 41 | bus.subscribe({ subscriberId: 'legacy-broadcast' }, unscopedSubscriber) |
| 42 | |
| 43 | bus.emit(jobStarted({ audience: { kind: 'owner' } })) |
| 44 | |
| 45 | expect(matchingOwner).toHaveBeenCalledOnce() |
| 46 | expect(otherOwner).not.toHaveBeenCalled() |
| 47 | expect(unscopedSubscriber).not.toHaveBeenCalled() |
| 48 | }) |
| 49 | |
| 50 | it('isolates listener failures and supports unsubscribe', () => { |
| 51 | const listenerError = vi.fn() |
| 52 | const bus = new TypedEventBus({ onListenerError: listenerError }) |
| 53 | const healthyListener = vi.fn() |
| 54 | bus.subscribe({}, () => { |
| 55 | throw new Error('listener failed') |
| 56 | }) |
| 57 | const unsubscribe = bus.subscribe({}, healthyListener) |
| 58 | |
| 59 | expect(() => bus.emit(jobStarted())).not.toThrow() |
| 60 | expect(listenerError).toHaveBeenCalledTimes(1) |
| 61 | expect(healthyListener).toHaveBeenCalledTimes(1) |
| 62 | |
| 63 | unsubscribe() |
| 64 | bus.emit(jobStarted()) |
| 65 | expect(healthyListener).toHaveBeenCalledTimes(1) |
| 66 | }) |
| 67 | |
| 68 | it('removes a bridge subscriber cleanly', () => { |
| 69 | const bus = new TypedEventBus() |
| 70 | const bridge = new RuntimeEventBridge(bus) |
| 71 | const send = vi.fn() |
| 72 | const unregister = bridge.register({ subscriberId: 'window-1', send }) |
| 73 | |
| 74 | bus.emit(jobStarted()) |
| 75 | expect(send).toHaveBeenCalledTimes(1) |
| 76 | |
| 77 | unregister() |
| 78 | bus.emit(jobStarted()) |
| 79 | expect(send).toHaveBeenCalledTimes(1) |
| 80 | }) |
| 81 | |
| 82 | it('does not let a stale disposer unregister a replacement subscriber', () => { |
| 83 | const bus = new TypedEventBus() |
| 84 | const bridge = new RuntimeEventBridge(bus) |
| 85 | const firstSend = vi.fn() |
| 86 | const secondSend = vi.fn() |
| 87 | const firstUnregister = bridge.register({ subscriberId: 'window-1', send: firstSend }) |
| 88 | const secondUnregister = bridge.register({ subscriberId: 'window-1', send: secondSend }) |
| 89 | |
| 90 | firstUnregister() |
| 91 | bus.emit(jobStarted()) |
| 92 | |
| 93 | expect(firstSend).not.toHaveBeenCalled() |
| 94 | expect(secondSend).toHaveBeenCalledOnce() |
| 95 | secondUnregister() |
| 96 | }) |
| 97 | |
| 98 | it('removes a targeted webContents subscriber when it is destroyed', () => { |
| 99 | const bus = new TypedEventBus() |
| 100 | const bridge = new RuntimeEventBridge(bus) |
| 101 | const send = vi.fn() |
| 102 | let onDestroyed: (() => void) | undefined |
| 103 | const removeListener = vi.fn() |
| 104 | const unregister = bridge.registerWebContents({ |
| 105 | subscriberId: 'window-1', |
| 106 | webContents: { |
| 107 | id: 1, |
| 108 | isDestroyed: () => false, |
| 109 | send, |
| 110 | once: (_event, listener) => { |
| 111 | onDestroyed = listener |
| 112 | }, |
| 113 | removeListener |
| 114 | }, |
| 115 | translate: (event) => ({ channel: 'runtime:event', payload: event }) |
| 116 | }) |
| 117 | |
| 118 | bus.emit(jobStarted()) |
| 119 | expect(send).toHaveBeenCalledOnce() |
| 120 | |
| 121 | onDestroyed?.() |
| 122 | bus.emit(jobStarted()) |
| 123 | expect(send).toHaveBeenCalledOnce() |
| 124 | expect(removeListener).toHaveBeenCalledWith('destroyed', expect.any(Function)) |
| 125 | |
| 126 | unregister() |
| 127 | }) |
| 128 | |
| 129 | it('does not let a replaced webContents destroy callback remove the new subscriber', () => { |
| 130 | const bus = new TypedEventBus() |
| 131 | const bridge = new RuntimeEventBridge(bus) |
| 132 | const firstSend = vi.fn() |
| 133 | const secondSend = vi.fn() |
| 134 | let firstDestroyed: (() => void) | undefined |
| 135 | let secondDestroyed: (() => void) | undefined |
| 136 | |
| 137 | bridge.registerWebContents({ |
| 138 | subscriberId: 'window-1', |
| 139 | webContents: { |
| 140 | id: 1, |
| 141 | isDestroyed: () => false, |
| 142 | send: firstSend, |
| 143 | once: (_event, listener) => { |
| 144 | firstDestroyed = listener |
| 145 | }, |
| 146 | removeListener: vi.fn() |
| 147 | }, |
| 148 | translate: (event) => ({ channel: 'runtime:event', payload: event }) |
| 149 | }) |
| 150 | bridge.registerWebContents({ |
| 151 | subscriberId: 'window-1', |
| 152 | webContents: { |
| 153 | id: 2, |
| 154 | isDestroyed: () => false, |
| 155 | send: secondSend, |
| 156 | once: (_event, listener) => { |
| 157 | secondDestroyed = listener |
| 158 | }, |
| 159 | removeListener: vi.fn() |
| 160 | }, |
| 161 | translate: (event) => ({ channel: 'runtime:event', payload: event }) |
| 162 | }) |
| 163 | |
| 164 | firstDestroyed?.() |
| 165 | bus.emit(jobStarted()) |
| 166 | |
| 167 | expect(firstSend).not.toHaveBeenCalled() |
| 168 | expect(secondSend).toHaveBeenCalledOnce() |
| 169 | |
| 170 | secondDestroyed?.() |
| 171 | bus.emit(jobStarted()) |
| 172 | expect(secondSend).toHaveBeenCalledOnce() |
| 173 | }) |
| 174 | |
| 175 | it('translates typed generation chunks to the legacy channel for every live window', () => { |
| 176 | const bus = new TypedEventBus() |
| 177 | const bridge = new RuntimeEventBridge(bus) |
| 178 | const sent = vi.fn() |
| 179 | const sendFailure = vi.fn() |
| 180 | const onSendError = vi.fn() |
| 181 | |
| 182 | bridge.registerWindowBroadcast({ |
| 183 | subscriberId: 'legacy-generate-chunk', |
| 184 | windows: () => [ |
| 185 | { |
| 186 | id: 1, |
| 187 | isDestroyed: () => false, |
| 188 | webContents: { isDestroyed: () => false, send: sent } |
| 189 | }, |
| 190 | { |
| 191 | id: 2, |
| 192 | isDestroyed: () => false, |
| 193 | webContents: { |
| 194 | isDestroyed: () => false, |
| 195 | send: () => { |
| 196 | sendFailure() |
| 197 | throw new Error('renderer closed while sending') |
| 198 | } |
| 199 | } |
| 200 | }, |
| 201 | { |
| 202 | id: 3, |
| 203 | isDestroyed: () => true, |
| 204 | webContents: { isDestroyed: () => false, send: vi.fn() } |
| 205 | } |
| 206 | ], |
| 207 | translate: (event) => |
| 208 | event.type === 'generation.chunk' |
| 209 | ? { channel: 'generate:chunk', payload: event.payload } |
| 210 | : null, |
| 211 | onSendError |
| 212 | }) |
| 213 | |
| 214 | const chunk = { |
| 215 | type: 'run_completed' as const, |
| 216 | payload: { runId: 'run-1', totalPages: 1 } |
| 217 | } |
| 218 | bus.emit({ |
| 219 | type: 'generation.chunk', |
| 220 | payload: chunk, |
| 221 | jobId: 'run-1', |
| 222 | domain: 'generation', |
| 223 | owner: { sessionId: 'session-1' }, |
| 224 | audience: { kind: 'broadcast' }, |
| 225 | occurredAt: 1 |
| 226 | }) |
| 227 | |
| 228 | expect(sent).toHaveBeenCalledWith('generate:chunk', chunk) |
| 229 | expect(sendFailure).toHaveBeenCalledOnce() |
| 230 | expect(onSendError).toHaveBeenCalledWith( |
| 231 | expect.objectContaining({ windowId: 2, event: expect.objectContaining({ jobId: 'run-1' }) }) |
| 232 | ) |
| 233 | }) |
| 234 | }) |
| 235 |