test(desktop): prove fresh chat socket continuity
This commit is contained in:
@@ -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<void>>
|
||||
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<string, unknown>) {
|
||||
const isOmar = socket.connectUrl?.includes(`:${ownerPort}`) ?? false
|
||||
|
||||
@@ -119,11 +124,19 @@ function answer(socket: MockGateway, method: string, params: Record<string, unkn
|
||||
throw new Error(`session.create landed on the wrong socket: ${socket.connectUrl}`)
|
||||
}
|
||||
|
||||
if (runtimeOwner) {
|
||||
throw new Error(`session.create attempted to replace runtime owner: ${socket.connectUrl}`)
|
||||
}
|
||||
|
||||
runtimeOwner = socket
|
||||
|
||||
return { info: {}, session_id: mintedRuntimeId, stored_session_id: mintedStoredId }
|
||||
}
|
||||
|
||||
if (sessionScoped(params) && !isOmar) {
|
||||
throw new Error(`Session not found: ${String(params.session_id)} (socket ${socket.connectUrl}, ${method})`)
|
||||
if (sessionScoped(params) && socket !== runtimeOwner) {
|
||||
throw new Error(
|
||||
`Session not found on concrete owner: ${String(params.session_id)} (socket ${socket.connectUrl}, ${method})`
|
||||
)
|
||||
}
|
||||
|
||||
if (method === 'prompt.submit') {
|
||||
@@ -131,16 +144,7 @@ function answer(socket: MockGateway, method: string, params: Record<string, unkn
|
||||
}
|
||||
|
||||
if (method === 'session.resume' || method === 'session.activate') {
|
||||
// The runtime is alive on this socket: a resume re-binds the SAME id.
|
||||
return {
|
||||
info: {},
|
||||
message_count: 1,
|
||||
messages: [],
|
||||
resumed: mintedStoredId,
|
||||
running: false,
|
||||
session_id: mintedRuntimeId,
|
||||
session_key: mintedStoredId
|
||||
}
|
||||
throw new Error(`${method} is recovery, not uninterrupted continuity`)
|
||||
}
|
||||
|
||||
return {}
|
||||
@@ -151,6 +155,8 @@ vi.mock('@/hermes', async importOriginal => ({
|
||||
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<string, unknown> = {}) => 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<string, unknown> = {}) => 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<string, unknown>) =>
|
||||
(activeGateway() as unknown as MockGateway).request(method, params)
|
||||
)
|
||||
const ambientRequest = vi.fn(async (method: string, params?: Record<string, unknown>) => {
|
||||
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(<Harness ambientRequest={ambientRequest as never} onReady={h => (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<string, unknown>) =>
|
||||
(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'
|
||||
|
||||
Reference in New Issue
Block a user