diff --git a/apps/desktop/src/app/gateway/hooks/use-gateway-boot.ts b/apps/desktop/src/app/gateway/hooks/use-gateway-boot.ts index 0af663741b..0300430b30 100644 --- a/apps/desktop/src/app/gateway/hooks/use-gateway-boot.ts +++ b/apps/desktop/src/app/gateway/hooks/use-gateway-boot.ts @@ -472,6 +472,23 @@ export function useGatewayBoot({ // profile name (every source has a 'default') can't collide. configureGatewayRegistry({ onActiveConnectionChanged: publish, + // Keep $activeGatewayProfile in lockstep with the registry's OWN record + // of which profile the active socket serves. The registry is the only + // party that sees eviction fallbacks (idle reap, connection removal, + // profile delete → primary); before this mirror those fallbacks moved + // the SOCKET back to the primary while the profile atom kept naming the + // evicted bot. ensureGatewayProfile's "already active" fast path then + // trusted the stale atom and skipped the re-swap, so every + // session-scoped RPC for that bot went out on the primary socket — the + // #89206 "Waking up… → retries gave up" wake failure, while the bot's + // own backend sat healthy and idle. + onActiveRouteChanged: profile => { + const key = normalizeProfileKey(profile) + + if (normalizeProfileKey($activeGatewayProfile.get()) !== key) { + $activeGatewayProfile.set(key) + } + }, onEvent: event => { recordSessionEventScope(event) callbacksRef.current.handleGatewayEvent(event) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts index 25cdbe6ee0..4679f4f144 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts @@ -67,6 +67,7 @@ import { setWorkspaceCwdOwner, setYoloActive } from '@/store/session' +import { requestForSessionProfile } from '@/store/session-request-router' import { $sessionTiles, closeSessionTile, @@ -736,6 +737,21 @@ export function useSessionActions({ await ensureGatewayProfile(sessionProfile) + // Request-time routing guard for every session-scoped RPC below. The + // await above REQUESTS the swap, but by dispatch time the active gateway + // can be back on another profile: a concurrent switch won the + // gatewaySwitch mutex, an eviction path (idle reap, connection edit, + // profile delete) re-pointed the active route at the primary, or the + // target's dial failed and scheduleReconnect left the previous socket + // active. Sending this session's resume/activate on whatever socket + // happens to be active then lands it on a backend that has never heard + // of the session — the backend boots, sits idle, and the renderer burns + // its bounded retries into the "retries gave up" screen while the bot's + // own backend is healthy one port over (#89206: local pool AND SSH). + // requestForSessionProfile re-resolves the route at each call. + const requestForSession = (method: string, params: Record = {}): Promise => + requestForSessionProfile(sessionProfile, requestGateway, method, params) + // Re-check after the profile-resolve / gateway-swap awaits above: the // cache may have changed, and takeWarmCache re-validates belongs-to and // purges a cross-wired mapping before we trust the fast-path. @@ -806,7 +822,7 @@ export function useSessionActions({ let activated: SessionResumeResponse | null = null try { - activated = await requestGateway('session.activate', { + activated = await requestForSession('session.activate', { session_id: cachedRuntimeId, cols: 96, omit_messages: true @@ -819,7 +835,7 @@ export function useSessionActions({ throw error } - const usage = await requestGateway('session.usage', { session_id: cachedRuntimeId }) + const usage = await requestForSession('session.usage', { session_id: cachedRuntimeId }) if (!isCurrentResume()) { return @@ -1047,7 +1063,7 @@ export function useSessionActions({ let resumeRuntimeBaselineMessages: ChatMessage[] = [] - const resumePromise = requestGateway('session.resume', { + const resumePromise = requestForSession('session.resume', { session_id: storedSessionId, cols: 96, source: 'desktop', diff --git a/apps/desktop/src/store/gateway.ts b/apps/desktop/src/store/gateway.ts index e50f20c41e..a3ad8e7834 100644 --- a/apps/desktop/src/store/gateway.ts +++ b/apps/desktop/src/store/gateway.ts @@ -27,6 +27,15 @@ interface RegistryConfig { onEvent: (event: GatewayEvent) => void onActiveConnectionInvalidated?: (fallbackProfile: string, activationEpoch: number) => void onActiveConnectionChanged?: (connection: HermesConnection) => void + /** + * Fires whenever applyActive() moves the active route to a (possibly + * different) profile — including registry-internal eviction fallbacks + * (idle reap, connection removal, profile delete) that no renderer call + * initiated. Consumers mirror this into $activeGatewayProfile so the + * published profile can never diverge from the socket actually selected + * (#89206: the stale-profile split-brain that stranded bot wake-ups). + */ + onActiveRouteChanged?: (profile: string) => void } // ── Secondary (pool) backends ────────────────────────────────────────────── @@ -71,6 +80,7 @@ interface GatewayRegistryState { activationEpoch: number secondaries: Map $gateway: ReturnType> + $activeProfile: ReturnType> } const STATE_KEY = Symbol.for('hermes.desktop.gatewayRegistryState') @@ -86,7 +96,14 @@ function createRegistryState(): GatewayRegistryState { // The active gateway instance, exposed for inline message-stream // components (inline ClarifyTool, model overlays) that call gateway // methods without the instance threaded down through props. - $gateway: atom(null) + $gateway: atom(null), + // The PROFILE the active gateway is routed to (bare profile name, never a + // composite registry scope). Owned exclusively by applyActive() so the + // published profile can never diverge from the socket actually selected — + // the split-brain where an eviction re-pointed activeKey at the primary + // while the profile atom kept naming the evicted bot routed every + // "loki" session.resume to the default backend (#89206 wake failures). + $activeProfile: atom('default') } } @@ -115,6 +132,19 @@ const g = gatewayState() // to. (A fresh `atom()` per reload would orphan existing subscriptions.) export const $gateway = g.$gateway +// The profile the ACTIVE gateway is actually routed to. Registry-owned: the +// only writer is applyActive(), which sets it in the same synchronous step +// that selects the socket — so a consumer that reads this and then calls +// activeGateway() always gets a matching (profile, socket) pair. Renderer +// surfaces (store/profile.ts's $activeGatewayProfile) mirror this atom +// instead of writing their own copy. +export const $activeGatewayRoute = g.$activeProfile + +/** Bare profile name the active gateway serves (never a composite scope). */ +export function activeGatewayProfileKey(): string { + return g.$activeProfile.get() +} + export function configureGatewayRegistry(cfg: RegistryConfig): void { g.config = cfg } @@ -218,6 +248,18 @@ function applyActive(profile: string, activationEpoch: number): boolean { // through the same source of truth every activation path maintains here — // registry-agent activations included, not just profile switches. setApiRequestConnection(activeGatewayConnectionId()) + // Publish the BARE profile this route serves, in the same synchronous step + // as the socket selection. activeKey may be a composite registry scope + // (connectionId::profile); consumers route RPCs by profile, so resolve it + // through the secondary's own record. This atom is the single source of + // truth for "which profile is the active gateway on" — every eviction / + // fallback path funnels through applyActive, so the published profile can + // never linger on a backend that is no longer selected (#89206). + const routeProfile = + g.activeKey === g.primaryProfile ? g.primaryProfile : (g.secondaries.get(g.activeKey)?.profile ?? g.primaryProfile) + + g.$activeProfile.set(routeProfile) + g.config?.onActiveRouteChanged?.(routeProfile) return true } diff --git a/apps/desktop/src/store/profile.ts b/apps/desktop/src/store/profile.ts index 2e67f54cd2..68f0574bfb 100644 --- a/apps/desktop/src/store/profile.ts +++ b/apps/desktop/src/store/profile.ts @@ -181,7 +181,11 @@ export async function switchProfile(name: string): Promise { // A single-profile user never triggers a swap, so their path is unchanged. // The profile the live gateway WebSocket is currently connected to. Initialized -// to the primary (window) backend's profile on boot. +// to the primary (window) backend's profile on boot. The gateway registry +// mirrors its own route into this atom via the onActiveRouteChanged callback +// (wired in use-gateway-boot's configureGatewayRegistry), so registry-internal +// eviction fallbacks (idle reap, connection removal, profile delete) can never +// leave this naming a profile the active socket no longer serves (#89206). export const $activeGatewayProfile = atom('default') // Profile for the NEXT new chat (chosen via the new-chat picker). null = primary diff --git a/apps/desktop/src/store/session-request-router.test.ts b/apps/desktop/src/store/session-request-router.test.ts new file mode 100644 index 0000000000..f6f3ef89a8 --- /dev/null +++ b/apps/desktop/src/store/session-request-router.test.ts @@ -0,0 +1,199 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +// Regression coverage for the #89206 wake-failure class: session-scoped RPCs +// routed to a backend that does not own the session's profile. Three layers: +// 1. The registry publishes the ACTIVE route's profile ($activeGatewayRoute) +// from applyActive itself, so eviction fallbacks move it in lockstep. +// 2. store/profile.ts mirrors that atom into $activeGatewayProfile, so the +// "already active" fast path can never trust a stale profile. +// 3. session-request-router pins session-scoped RPCs to the owning +// profile's socket at REQUEST time when the active route diverges. + +const secondaryGateways: Array<{ + close: ReturnType + connect: ReturnType + connectionState: string + request: ReturnType +}> = [] + +vi.mock('@/hermes', () => ({ + HermesGateway: class { + connectionState = 'closed' + connect = vi.fn(async () => { + this.connectionState = 'open' + }) + request = vi.fn(async (method: string, params: Record) => { + if (this.connectionState !== 'open') { + throw new Error('gateway is not connected') + } + + return { method, params } + }) + close = vi.fn() + onEvent = vi.fn(() => () => {}) + onState = vi.fn(() => () => {}) + + constructor() { + secondaryGateways.push(this) + } + }, + setApiRequestConnection: vi.fn() +})) +vi.mock('@/store/session', () => ({ setConnection: vi.fn(), setGatewayState: vi.fn() })) +vi.mock('@/store/notify-baseline', () => ({ markNativeNotifyBaseline: vi.fn() })) + +const { + $activeGatewayRoute, + activeGatewayProfileKey, + closeSecondaryGateways, + configureGatewayRegistry, + ensureGatewayForProfile, + pruneSecondaryGateways, + retireLocalProfileGateways, + setPrimaryGateway +} = await import('./gateway') + +const { requestForSessionProfile, sessionRpcNeedsProfileRoute } = await import('./session-request-router') + +function installDesktop(): void { + ;(window as unknown as { hermesDesktop: unknown }).hermesDesktop = { + getConnection: vi.fn(async (profile: null | string) => + profile ? { port: 5151, profile, token: 'secondary-token' } : { port: 4242, token: 'primary-token' } + ), + touchBackend: vi.fn(async () => undefined) + } +} + +function makePrimary() { + return { + connectionState: 'open', + request: vi.fn(async (method: string, params: Record) => ({ method, params })) + } +} + +beforeEach(() => { + secondaryGateways.length = 0 + configureGatewayRegistry({ onEvent: vi.fn() }) + closeSecondaryGateways() +}) + +afterEach(() => { + closeSecondaryGateways() + vi.clearAllMocks() + delete (window as unknown as { hermesDesktop?: unknown }).hermesDesktop +}) + +describe('$activeGatewayRoute (registry-owned active profile)', () => { + it('tracks profile activation and eviction fallback in lockstep with the socket', async () => { + const primary = makePrimary() + setPrimaryGateway(primary as never, 'default') + installDesktop() + + await ensureGatewayForProfile('default') + expect(activeGatewayProfileKey()).toBe('default') + + await ensureGatewayForProfile('loki') + expect(activeGatewayProfileKey()).toBe('loki') + expect($activeGatewayRoute.get()).toBe('loki') + + // Idle-reap style eviction of everything but... nothing keeps loki alive. + // The registry must move BOTH the socket and the published profile back + // to the primary — before the fix only the socket moved, and the stale + // profile atom made ensureGatewayProfile skip the re-swap forever. + retireLocalProfileGateways('loki') + expect(activeGatewayProfileKey()).toBe('default') + expect($activeGatewayRoute.get()).toBe('default') + }) + + it('falls back to primary when pruning evicts the active secondary', async () => { + const primary = makePrimary() + setPrimaryGateway(primary as never, 'default') + installDesktop() + + await ensureGatewayForProfile('hulk') + expect(activeGatewayProfileKey()).toBe('hulk') + + // Force-evict the active entry (retention flags off) — the keep-set is + // empty and the active guard is bypassed by retiring first. + retireLocalProfileGateways('hulk') + pruneSecondaryGateways(new Set()) + + expect(activeGatewayProfileKey()).toBe('default') + }) +}) + +describe('sessionRpcNeedsProfileRoute', () => { + it('routes ambient when the owner is unknown or already active', () => { + expect(sessionRpcNeedsProfileRoute(null, 'default')).toBe(false) + expect(sessionRpcNeedsProfileRoute('', 'default')).toBe(false) + expect(sessionRpcNeedsProfileRoute(' ', 'loki')).toBe(false) + expect(sessionRpcNeedsProfileRoute('loki', 'loki')).toBe(false) + expect(sessionRpcNeedsProfileRoute('default', 'default')).toBe(false) + }) + + it('pins to the owning profile when the active route diverges', () => { + expect(sessionRpcNeedsProfileRoute('loki', 'default')).toBe(true) + expect(sessionRpcNeedsProfileRoute('default', 'loki')).toBe(true) + expect(sessionRpcNeedsProfileRoute('loki', 'hulk')).toBe(true) + }) +}) + +describe('requestForSessionProfile', () => { + it("dispatches on the owning profile's own socket when the active route moved off it (#89206)", async () => { + const primary = makePrimary() + setPrimaryGateway(primary as never, 'default') + installDesktop() + await ensureGatewayForProfile('default') + + const ambient = vi.fn(async (method: string, params?: Record) => ({ + ambient: true, + method, + params + })) + + // Active route is 'default'; the session belongs to 'loki'. The failing + // path sent session.resume on the ambient (default) socket — the default + // backend has never heard of the session and the bot never woke. + const result = await requestForSessionProfile<{ method: string; params: Record }>( + 'loki', + ambient as never, + 'session.resume', + { session_id: 'stored-loki-chat' } + ) + + expect(ambient).not.toHaveBeenCalled() + expect(result).toEqual({ method: 'session.resume', params: { session_id: 'stored-loki-chat' } }) + expect(secondaryGateways).toHaveLength(1) + expect(secondaryGateways[0].request).toHaveBeenCalledWith('session.resume', { session_id: 'stored-loki-chat' }) + }) + + it('keeps the ambient dispatcher when the active route already serves the owner', async () => { + const primary = makePrimary() + setPrimaryGateway(primary as never, 'default') + installDesktop() + await ensureGatewayForProfile('loki') + + const ambient = vi.fn(async (method: string, params?: Record) => ({ + ambient: true, + method, + params + })) + + const result = await requestForSessionProfile('loki', ambient as never, 'session.activate', { session_id: 'rt-1' }) + + expect(ambient).toHaveBeenCalledWith('session.activate', { session_id: 'rt-1' }) + expect(result).toEqual({ ambient: true, method: 'session.activate', params: { session_id: 'rt-1' } }) + }) + + it('keeps the ambient dispatcher for sessions with no owning profile', async () => { + const primary = makePrimary() + setPrimaryGateway(primary as never, 'default') + installDesktop() + await ensureGatewayForProfile('default') + + const ambient = vi.fn(async () => ({ ambient: true })) + await requestForSessionProfile(null, ambient as never, 'session.usage', { session_id: 'rt-2' }) + + expect(ambient).toHaveBeenCalledOnce() + }) +}) diff --git a/apps/desktop/src/store/session-request-router.ts b/apps/desktop/src/store/session-request-router.ts new file mode 100644 index 0000000000..8c67a52916 --- /dev/null +++ b/apps/desktop/src/store/session-request-router.ts @@ -0,0 +1,52 @@ +import { activeGatewayProfileKey, requestGatewayForProfile } from '@/store/gateway' + +// ── Session-scoped RPC routing (the #89206 class) ─────────────────────────── +// A session-scoped RPC (session.resume / session.activate / session.usage) +// only means anything on the backend that OWNS the session's profile. The +// ambient "active gateway" is a moving target: between the profile-swap await +// and the RPC dispatch, a concurrent switch, an idle-reap eviction, a failed +// dial, or a connection edit can re-point the active route at another +// backend. Dispatching on it anyway lands the RPC on a backend that has never +// heard of the session — it 404s or times out, the renderer burns its bounded +// retries, and the user sees "retries gave up" while the session's own +// backend is healthy (blank Bot Chats, dead wake-ups; local pool and SSH +// alike). These helpers make the owning profile, resolved at REQUEST time, +// the routing authority. + +const normKey = (profile: null | string | undefined): string => (profile ?? '').trim() || 'default' + +/** + * True when a session-scoped RPC must be pinned to `ownerProfile`'s own + * socket because the active gateway currently serves a different profile. + * A null/empty owner means the session's profile is unknown — route ambient + * (the pre-multi-profile behavior) rather than guessing. + */ +export function sessionRpcNeedsProfileRoute( + ownerProfile: null | string | undefined, + activeProfile: string = activeGatewayProfileKey() +): boolean { + if (ownerProfile == null || !String(ownerProfile).trim()) { + return false + } + + return normKey(ownerProfile) !== normKey(activeProfile) +} + +/** + * Dispatch a session-scoped RPC on the socket that owns `ownerProfile`, + * falling back to the ambient dispatcher when the active gateway already + * serves that profile (keeps the primary's reauth-aware reconnect path). + * The route is decided at CALL time, not at swap time. + */ +export function requestForSessionProfile( + ownerProfile: null | string | undefined, + ambientRequest: (method: string, params?: Record) => Promise, + method: string, + params: Record = {} +): Promise { + if (!sessionRpcNeedsProfileRoute(ownerProfile)) { + return ambientRequest(method, params) + } + + return requestGatewayForProfile(normKey(ownerProfile), method, params) +}