diff --git a/apps/desktop/src/app/session/hooks/profile-rail-fresh-chat-owner.test.tsx b/apps/desktop/src/app/session/hooks/profile-rail-fresh-chat-owner.test.tsx index 48aa678ccb..a592fe0cf9 100644 --- a/apps/desktop/src/app/session/hooks/profile-rail-fresh-chat-owner.test.tsx +++ b/apps/desktop/src/app/session/hooks/profile-rail-fresh-chat-owner.test.tsx @@ -1,4 +1,4 @@ -import { registryBackendScopeKey } from '@hermes/shared' +import { type GatewayEvent, registryBackendScopeKey } from '@hermes/shared' import { useStore } from '@nanostores/react' import { act, cleanup, render, waitFor } from '@testing-library/react' import { useEffect, useMemo, useRef } from 'react' @@ -20,6 +20,7 @@ import { $newChatProfile, $newChatRoute, ensureGatewayAgent, + newSessionInProfile, selectProfile } from '@/store/profile' import { @@ -92,12 +93,16 @@ interface MockGateway { connectionState: string connect: Mock<(url: string) => Promise> close: Mock<() => void> - onEvent: Mock<() => () => void> - onState: Mock<() => () => void> + eventListeners: Set<(event: GatewayEvent) => void> + onEvent: Mock<(listener: (event: GatewayEvent) => void) => () => void> + onState: Mock<(listener: (state: 'closed' | 'open') => void) => () => void> request: GatewayRequestMock + stateListeners: Set<(state: 'closed' | 'open') => void> } const sockets: MockGateway[] = [] +let runtimeOwner: MockGateway | null = null +let registryOnEvent!: Mock<(event: GatewayEvent) => void> /** The port of the ONE socket allowed to mint (and then own) the runtime. */ let ownerPort = OMAR_PORT /** The ids the owner socket mints — per case, so one case's owner records @@ -108,9 +113,9 @@ let mintedStoredId = STORED_ID const sessionScoped = (params: unknown) => typeof (params as { session_id?: unknown } | undefined)?.session_id === 'string' -/** The owner socket (the registry entry homelab::omar, or the v1 omar socket - * for a legacy pick) answers; every other socket is a backend that never - * held the runtime, exactly as in the field. */ +/** The exact object that mints the runtime owns it. Endpoint equality is not + * ownership: a replacement WebSocket connected to the same URL has never held + * this in-memory runtime and must fail exactly like a different backend. */ function answer(socket: MockGateway, method: string, params: Record) { const isOmar = socket.connectUrl?.includes(`:${ownerPort}`) ?? false @@ -119,11 +124,19 @@ function answer(socket: MockGateway, method: string, params: Record ({ HermesGateway: class { connectUrl: null | string = null connectionState = 'closed' + eventListeners = new Set<(event: GatewayEvent) => void>() + stateListeners = new Set<(state: 'closed' | 'open') => void>() connect = vi.fn(async (url: string) => { this.connectUrl = url this.connectionState = 'open' @@ -164,9 +170,27 @@ vi.mock('@/hermes', async importOriginal => ({ }) close = vi.fn(() => { this.connectionState = 'closed' + this.stateListeners.forEach(listener => listener('closed')) + + if (runtimeOwner === (this as unknown as MockGateway)) { + this.eventListeners.forEach(listener => + listener({ + payload: { session_id: mintedRuntimeId, stored_session_id: mintedStoredId }, + type: 'session.reclaimed' + } as GatewayEvent) + ) + } + }) + onEvent = vi.fn((listener: (event: GatewayEvent) => void) => { + this.eventListeners.add(listener) + + return () => this.eventListeners.delete(listener) + }) + onState = vi.fn((listener: (state: 'closed' | 'open') => void) => { + this.stateListeners.add(listener) + + return () => this.stateListeners.delete(listener) }) - onEvent = vi.fn(() => () => {}) - onState = vi.fn(() => () => {}) constructor() { sockets.push(this as unknown as MockGateway) @@ -204,14 +228,27 @@ function installDesktop(): void { /** The remote primary. Session-scoped traffic here is the bug. */ function makePrimary(): MockGateway { + const eventListeners = new Set<(event: GatewayEvent) => void>() + const stateListeners = new Set<(state: 'closed' | 'open') => void>() + const primary: MockGateway = { connectUrl: 'ws://remote-primary:4242', connectionState: 'open', connect: vi.fn(), close: vi.fn(), - onEvent: vi.fn(() => () => {}), - onState: vi.fn(() => () => {}), - request: vi.fn(async (method: string, params: Record = {}) => answer(primary, method, params)) + eventListeners, + onEvent: vi.fn(listener => { + eventListeners.add(listener) + + return () => eventListeners.delete(listener) + }), + onState: vi.fn(listener => { + stateListeners.add(listener) + + return () => stateListeners.delete(listener) + }), + request: vi.fn(async (method: string, params: Record = {}) => answer(primary, method, params)), + stateListeners } return primary @@ -334,15 +371,17 @@ const omarScope = registryBackendScopeKey(SOURCE_ID, 'omar') describe('profile rail: a fresh Omar chat keeps its exact registry owner across turns (#94071)', () => { beforeEach(() => { sockets.length = 0 + runtimeOwner = null ownerPort = OMAR_PORT mintedRuntimeId = RUNTIME_ID mintedStoredId = STORED_ID clearSingleFlightSessionResumeState() // Wired exactly as useGatewayBoot: the published active descriptor carries // a registry-backed primary's source identity across a renderer reload. + registryOnEvent = vi.fn() configureGatewayRegistry({ activeConnectionId: () => $connection.get()?.connectionId ?? null, - onEvent: vi.fn() + onEvent: registryOnEvent }) closeSecondaryGateways() installDesktop() @@ -377,7 +416,7 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across /** Boot the exact field state: remote primary on `default`, `homelab` as * the active registry source, then selectProfile("omar") in the rail; mount * the window's real hook stack over the production dispatcher. */ - async function bootProfileRailOmar() { + async function bootProfileRailOmar(startDraft: 'newSessionInProfile' | 'selectProfile' = 'selectProfile') { // Primary / ambient source: a remote gateway on `default`. const primary = makePrimary() setPrimaryGateway(primary as never, 'default') @@ -387,8 +426,13 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across await ensureGatewayAgent(SOURCE_ID, 'default') expect(activeGatewayConnectionId()).toBe(SOURCE_ID) - // The profile rail: selectProfile("omar"). - selectProfile('omar') + // Both profile-rail entry points must capture the same concrete source. + if (startDraft === 'newSessionInProfile') { + newSessionInProfile('omar') + } else { + selectProfile('omar') + } + expect($newChatProfile.get()).toBe('omar') expect($newChatRoute.get()).toBeNull() await waitFor(() => expect(activeGatewayProfileKey()).toBe('omar')) @@ -405,21 +449,25 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across expect(activeGateway()).toBe(omarSocket as never) // Ambient dispatcher = whatever socket is active, as useGatewayRequest does. - const ambientRequest = vi.fn(async (method: string, params?: Record) => - (activeGateway() as unknown as MockGateway).request(method, params) - ) + const ambientRequest = vi.fn(async (method: string, params?: Record) => { + if (method === 'session.create' || sessionScoped(params)) { + throw new Error(`session traffic must not use ambient dispatcher: ${method}`) + } + + return (activeGateway() as unknown as MockGateway).request(method, params) + }) let handle: HarnessHandle | null = null render( (handle = h)} />) await waitFor(() => expect(handle).not.toBeNull()) - return { handle: handle!, omarSocket: omarSocket!, primary } + return { ambientRequest, handle: handle!, omarSocket: omarSocket!, primary } } /** What the gateway's stream end does: the turn settles. */ async function settleTurn(handle: HarnessHandle) { await act(async () => { - handle.updateSessionState(RUNTIME_ID, state => ({ + handle.updateSessionState(mintedRuntimeId, state => ({ ...state, awaitingResponse: false, busy: false, @@ -478,8 +526,31 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across expect($newChatConnectionId.get()).toBeNull() }) + function expectUninterruptedOwner({ + ambientRequest, + omarSocket, + primary + }: { + ambientRequest: GatewayRequestMock + omarSocket: MockGateway + primary: MockGateway + }) { + expect(runtimeOwner).toBe(omarSocket) + expect(ambientRequest.mock.calls.filter(call => call[0] === 'session.create' || sessionScoped(call[1]))).toEqual([]) + + for (const socket of [primary, ...sockets]) { + expect(socket.close).not.toHaveBeenCalled() + expect(socket.connectionState).toBe('open') + expect( + calls(socket).filter(method => ['session.activate', 'session.close', 'session.resume'].includes(method)) + ).toEqual([]) + } + + expect(registryOnEvent.mock.calls.filter(([event]) => event.type === 'session.reclaimed')).toEqual([]) + } + it('session.create and both prompt.submit calls ride the SAME conn:homelab::omar socket', async () => { - const { handle, omarSocket, primary } = await bootProfileRailOmar() + const { ambientRequest, handle, omarSocket, primary } = await bootProfileRailOmar() // Turn one: no session yet → createBackendSessionForSend → prompt.submit. await expect(handle.submitText('first prompt')).resolves.toBe(true) @@ -546,6 +617,7 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across } expect(omarCalls.filter(method => method === 'session.create')).toHaveLength(1) + expectUninterruptedOwner({ ambientRequest: ambientRequest as GatewayRequestMock, omarSocket, primary }) expect(getSessionOwnerHint(STORED_ID)).toEqual({ connectionId: SOURCE_ID, profile: 'omar' }) expect($sessions.get().find(session => sessionMatchesStoredId(session, STORED_ID))).toMatchObject({ connection_id: SOURCE_ID, @@ -555,7 +627,7 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across }) it('turn two still rides conn:homelab::omar after the transient hint is evicted AND a refresh returned the row untagged', async () => { - const { handle, omarSocket, primary } = await bootProfileRailOmar() + const { ambientRequest, handle, omarSocket, primary } = await bootProfileRailOmar('newSessionInProfile') await expect(handle.submitText('first prompt')).resolves.toBe(true) await waitFor(() => expect($activeSessionId.get()).toBe(RUNTIME_ID)) @@ -612,6 +684,7 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across } expect(calls(omarSocket).filter(method => method === 'session.create')).toHaveLength(1) + expectUninterruptedOwner({ ambientRequest: ambientRequest as GatewayRequestMock, omarSocket, primary }) }) it('a pick on the explicit `local` source is a legacy profile pick: create and both turns ride the ONE v1 omar socket', async () => { @@ -649,6 +722,8 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across expect(activeGateway()).toBe(v1Socket as never) expect(sockets.some(socket => socket.connectUrl?.includes(`:${OMAR_PORT}`))).toBe(false) + // The legacy door's create legitimately rides the ambient dispatcher: the + // active socket IS the owner (no route, no registry entry to name). const ambientRequest = vi.fn(async (method: string, params?: Record) => (activeGateway() as unknown as MockGateway).request(method, params) ) @@ -661,23 +736,14 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across await waitFor(() => expect($activeSessionId.get()).toBe(LEGACY_RUNTIME_ID)) expect(handle!.bindings()).toEqual({ runtimeForStored: LEGACY_RUNTIME_ID, storedForRuntime: LEGACY_STORED_ID }) - await act(async () => { - handle!.updateSessionState(LEGACY_RUNTIME_ID, state => ({ - ...state, - awaitingResponse: false, - busy: false, - streamId: null, - turnStartedAt: null - })) - handle!.busyRef.current = false - setBusy(false) - setAwaitingResponse(false) - }) + await settleTurn(handle!) await expect(handle!.submitText('second prompt')).resolves.toBe(true) // The legacy owner is the bare profile: no registry route, no hint — the - // row's profile names the same v1 pool entry that minted the runtime. + // row's profile names the same v1 pool entry that minted the runtime, and + // that socket was never closed, resumed or replaced. + expect(runtimeOwner).toBe(v1Socket) expect(calls(v1Socket!).filter(method => method === 'session.create')).toHaveLength(1) expect( v1Socket!.request.mock.calls @@ -698,9 +764,13 @@ describe('profile rail: a fresh Omar chat keeps its exact registry owner across expect(vi.mocked(getSession)).not.toHaveBeenCalled() for (const socket of [primary, ...sockets]) { - expect(calls(socket).filter(method => method === 'session.close')).toEqual([]) + expect(socket.close).not.toHaveBeenCalled() + expect( + calls(socket).filter(method => ['session.activate', 'session.close', 'session.resume'].includes(method)) + ).toEqual([]) } + expect(registryOnEvent.mock.calls.filter(([event]) => event.type === 'session.reclaimed')).toEqual([]) expect(getSessionOwnerHint(LEGACY_STORED_ID)).toBeUndefined() expect($sessions.get().find(session => sessionMatchesStoredId(session, LEGACY_STORED_ID))).toMatchObject({ profile: 'omar'