From e21291663f72a45c95cb80078912dfa0f1e57166 Mon Sep 17 00:00:00 2001 From: Mohamad Kanso <91088196+MohamadKanso@users.noreply.github.com> Date: Tue, 22 Sep 2026 12:51:47 +0100 Subject: [PATCH] fix(desktop): aggregate registered gateway sessions --- apps/desktop/electron/main.ts | 73 +++++-- .../electron/profile-session-routing.test.ts | 115 ++++++++++ .../electron/profile-session-routing.ts | 199 ++++++++++++++++-- 3 files changed, 356 insertions(+), 31 deletions(-) diff --git a/apps/desktop/electron/main.ts b/apps/desktop/electron/main.ts index 64f6371591..c13c3c4048 100644 --- a/apps/desktop/electron/main.ts +++ b/apps/desktop/electron/main.ts @@ -439,6 +439,8 @@ import { fetchRegistrySessionRows, fetchRemoteProfileSessions, findRemoteOwnerProfileForSession, + hasPinnedRegistrySessionSource, + isAllProfilesSessionListRequest, mergeProfileSessionWindow, pathWithRemoteOwnerScope, type RegistrySessionSource, @@ -16809,7 +16811,7 @@ ipcMain.handle('hermes:window:readBelow', async event => { // GET /api/sessions/{id}[/messages] → read from remote // DELETE /api/sessions/{id} → delete on remote // PATCH /api/sessions/{id} → rename/archive on remote -async function interceptSessionRequestForRemote(request) { +async function interceptSessionRequestForRemote(request, registryConnectionId = null) { if (typeof request?.path !== 'string') { return undefined } @@ -16828,19 +16830,34 @@ async function interceptSessionRequestForRemote(request) { if (method === 'GET' && pathname === '/api/profiles/sessions') { const remoteProfiles = configuredRemoteProfileNames() - const registrySources = await pooledRegistrySessionSources() + const registrySources = await pooledRegistrySessionSources(Boolean(registryConnectionId)) if (remoteProfiles.length === 0 && registrySources.length === 0) { return undefined // no remote profiles and no connected registry gateways → local fast path } + if ( + !hasPinnedRegistrySessionSource( + registryConnectionId, + request?.profile, + registrySources, + !globalRemoteActive() + ) + ) { + // Do not manufacture a partial all-gateways response while the selected + // registry backend is still dialing or has just gone idle. The caller + // falls back to the direct pinned route, which is slower but complete + // for the gateway the renderer actually selected. + return undefined + } + const requested = (searchParams.get('profile') || 'all').trim() || 'all' if (requested !== 'all') { return profileHasRemoteOverride(requested) ? remoteSessionList(requested, searchParams) : undefined } - return mergeRemoteProfileSessions(searchParams, remoteProfiles) + return mergeRemoteProfileSessions(searchParams, remoteProfiles, registrySources) } // Batched sidebar slices. With no remote profiles the local batched endpoint @@ -16851,18 +16868,29 @@ async function interceptSessionRequestForRemote(request) { // remote correctness is preserved. if (method === 'GET' && pathname === '/api/profiles/sessions/sidebar') { const remoteProfiles = configuredRemoteProfileNames() - const registrySources = await pooledRegistrySessionSources() + const registrySources = await pooledRegistrySessionSources(Boolean(registryConnectionId)) if (remoteProfiles.length === 0 && registrySources.length === 0) { return undefined // local fast path → batched endpoint's single DB open } + if ( + !hasPinnedRegistrySessionSource( + registryConnectionId, + request?.profile, + registrySources, + !globalRemoteActive() + ) + ) { + return undefined + } + const { recents: recentsSp, cron: cronSp, messaging: messagingSp } = buildSidebarSessionSliceParams(searchParams) const [recents, cron, messaging] = await Promise.all([ - fetchProfilesSessionSlice(recentsSp, remoteProfiles), - fetchProfilesSessionSlice(cronSp, remoteProfiles), - fetchProfilesSessionSlice(messagingSp, remoteProfiles) + fetchProfilesSessionSlice(recentsSp, remoteProfiles, registrySources), + fetchProfilesSessionSlice(cronSp, remoteProfiles, registrySources), + fetchProfilesSessionSlice(messagingSp, remoteProfiles, registrySources) ]) return assembleSidebarSessionSlices(recents, cron, messaging) @@ -17003,7 +17031,7 @@ async function remoteOwnerProfileForSession(sessionId: string) { // returns data (never `undefined`) so a batched caller can compose slices. A // specific local profile reads from the local primary; a remote-override profile // reads from its remote; 'all' merges every remote into the primary aggregate. -async function fetchProfilesSessionSlice(searchParams, remoteProfiles) { +async function fetchProfilesSessionSlice(searchParams, remoteProfiles, registrySources = null) { const requested = (searchParams.get('profile') || 'all').trim() || 'all' if (requested !== 'all') { @@ -17014,7 +17042,7 @@ async function fetchProfilesSessionSlice(searchParams, remoteProfiles) { return fetchPrimaryProfileSessions(searchParams, fetchJsonForProfile) } - return mergeRemoteProfileSessions(searchParams, remoteProfiles) + return mergeRemoteProfileSessions(searchParams, remoteProfiles, registrySources) } // Unified list: primary's local aggregate, with each remote profile's stale local @@ -17023,7 +17051,7 @@ async function fetchProfilesSessionSlice(searchParams, remoteProfiles) { // than breaking the sidebar. Connected registry gateways' sessions are spliced // in too (#88880) — the unified Sessions list shows EVERY connected gateway's // chats, tagged with connection_id + profile so opens route correctly. -async function mergeRemoteProfileSessions(searchParams, remoteProfiles) { +async function mergeRemoteProfileSessions(searchParams, remoteProfiles, registrySourcesOverride = null) { const limit = Math.max(1, Number(searchParams.get('limit')) || 20) const offset = Math.max(0, Number(searchParams.get('offset')) || 0) const order = searchParams.get('order') === 'created' ? 'started_at' : 'last_active' @@ -17064,7 +17092,7 @@ async function mergeRemoteProfileSessions(searchParams, remoteProfiles) { // refresh must never dial or spawn a backend (the Bot Mode roster-respawn // trap). Reads omit include_hidden, so Bot Mode's hidden canonical chats // stay out of the global list, same as local sessions. - const registrySources = await pooledRegistrySessionSources() + const registrySources = registrySourcesOverride || (await pooledRegistrySessionSources()) if (registrySources.length) { const registryRows = await fetchRegistrySessionRows(registrySources, remoteParams, (descriptor, path) => @@ -17090,14 +17118,15 @@ async function mergeRemoteProfileSessions(searchParams, remoteProfiles) { // straight from the backend pool, never dialing. SSH sources contribute one // backend per pooled (connection, profile) scope; remote/cloud sources are one // shared host (any pooled scope's descriptor serves the cross-profile read). -// The primary local connection is excluded — the primary aggregate already -// carries local rows. -async function pooledRegistrySessionSources(): Promise { +// The primary local connection is excluded for the legacy unpinned path — the +// primary aggregate carries local rows there. A registry-pinned aggregate opts +// in so forced-local backends remain visible when the legacy primary is remote. +async function pooledRegistrySessionSources(includeLocal = false): Promise { const registry = readDesktopConnectionsRegistry() const sources: RegistrySessionSource[] = [] for (const connection of registry.connections) { - if (connection.kind === 'local') { + if (connection.kind === 'local' && !includeLocal) { continue } @@ -17113,7 +17142,9 @@ async function pooledRegistrySessionSources(): Promise const backends: Array<{ descriptor: unknown; profileLabel: null | string }> = [] - for (const [key, entry] of connection.kind === 'ssh' ? pooled : pooled.slice(0, 1)) { + const perProfile = connection.kind === 'ssh' || (includeLocal && connection.kind === 'local') + + for (const [key, entry] of perProfile ? pooled : pooled.slice(0, 1)) { try { // Already-resolved for a connected backend; a still-dialing entry is // skipped via the timeout guard rather than blocking the sidebar. @@ -17124,7 +17155,7 @@ async function pooledRegistrySessionSources(): Promise backends.push({ descriptor, - profileLabel: connection.kind === 'ssh' ? key.slice(prefix.length) || 'default' : null + profileLabel: perProfile ? key.slice(prefix.length) || 'default' : null }) } catch { // Dead or still-connecting backend — contributes nothing this refresh. @@ -17206,6 +17237,14 @@ async function handleHermesApiRequest(request) { const registryConnectionId = apiRequestRegistryConnectionId(request) if (registryConnectionId) { + if (isAllProfilesSessionListRequest(request?.method, request?.path)) { + const aggregate = await interceptSessionRequestForRemote(request, registryConnectionId) + + if (aggregate !== undefined) { + return aggregate + } + } + return dispatchRegistryApiRequest(request, registryConnectionId) } diff --git a/apps/desktop/electron/profile-session-routing.test.ts b/apps/desktop/electron/profile-session-routing.test.ts index 26d9f0321e..365f170693 100644 --- a/apps/desktop/electron/profile-session-routing.test.ts +++ b/apps/desktop/electron/profile-session-routing.test.ts @@ -9,6 +9,8 @@ import { fetchRegistrySessionRows, fetchRemoteProfileSessions, findRemoteOwnerProfileForSession, + hasPinnedRegistrySessionSource, + isAllProfilesSessionListRequest, mergeProfileSessionWindow, pathWithRemoteOwnerScope, remoteProfileQueryScope, @@ -17,6 +19,54 @@ import { tagRemoteSessionRows } from './profile-session-routing' +test('all-profiles session routing is limited to read-only list endpoints', () => { + assert.equal(isAllProfilesSessionListRequest('GET', '/api/profiles/sessions?profile=all'), true) + assert.equal(isAllProfilesSessionListRequest('GET', '/api/profiles/sessions/sidebar?recents_profile=all'), true) + assert.equal(isAllProfilesSessionListRequest('GET', '/api/profiles/sessions?profile=default'), false) + assert.equal(isAllProfilesSessionListRequest('POST', '/api/profiles/sessions?profile=all'), false) + assert.equal(isAllProfilesSessionListRequest('GET', '/api/sessions?profile=all'), false) +}) + +test('pinned registry aggregation requires the selected gateway to be pooled', () => { + const sources = [{ connectionId: 'gateway-remote' }, { connectionId: 'gateway-ssh' }] + + assert.equal(hasPinnedRegistrySessionSource('gateway-remote', 'default', sources), true) + assert.equal(hasPinnedRegistrySessionSource('gateway-missing', 'default', sources), false) + assert.equal(hasPinnedRegistrySessionSource('local', 'default', sources), true) + assert.equal(hasPinnedRegistrySessionSource('local', 'default', sources, false), false) + assert.equal(hasPinnedRegistrySessionSource('local', 'default', [{ connectionId: 'local' }], false), true) + assert.equal(hasPinnedRegistrySessionSource(null, 'default', sources), true) +}) + +test('pinned registry aggregation requires the selected profile backend for per-profile sources', () => { + const sources = [ + { + connectionId: 'gateway-ssh', + kind: 'ssh', + backends: [{ descriptor: 'research-desc', profileLabel: 'research' }] + }, + { + connectionId: 'local', + kind: 'local', + backends: [ + { descriptor: 'default-desc', profileLabel: 'default' }, + { descriptor: 'work-desc', profileLabel: 'work' } + ] + }, + { + connectionId: 'gateway-remote', + kind: 'remote', + backends: [{ descriptor: 'remote-desc', profileLabel: null }] + } + ] + + assert.equal(hasPinnedRegistrySessionSource('gateway-ssh', 'research', sources), true) + assert.equal(hasPinnedRegistrySessionSource('gateway-ssh', 'default', sources), false) + assert.equal(hasPinnedRegistrySessionSource('local', 'work', sources, false), true) + assert.equal(hasPinnedRegistrySessionSource('local', 'missing', sources, false), false) + assert.equal(hasPinnedRegistrySessionSource('gateway-remote', 'any-profile', sources), true) +}) + test('remote sidebar slices all follow the selected profile', () => { const slices = buildSidebarSessionSliceParams( new URLSearchParams({ @@ -298,6 +348,41 @@ test('registry sources: ssh backends are read natively and rows tagged with conn assert.ok(rows.every(row => (row as any).is_default_profile === false)) }) +test('registry sources: forced-local backends are read per profile', async () => { + const calls: string[] = [] + + const rows = await fetchRegistrySessionRows( + [ + { + connectionId: 'local', + kind: 'local', + backends: [ + { descriptor: 'default-desc', profileLabel: 'default' }, + { descriptor: 'work-desc', profileLabel: 'work' } + ] + } + ], + new URLSearchParams({ profile: 'all', limit: '10' }), + async (descriptor, path) => { + calls.push(`${String(descriptor)} ${path}`) + + return { sessions: [{ id: String(descriptor) }], total: 1 } + } + ) + + assert.deepEqual(calls, [ + 'default-desc /api/sessions?limit=10', + 'work-desc /api/sessions?limit=10' + ]) + assert.deepEqual( + rows.map(row => [(row as any).id, (row as any).profile, (row as any).connection_id]), + [ + ['default-desc', 'default', 'local'], + ['work-desc', 'work', 'local'] + ] + ) +}) + test('registry sources: shared remote hosts read the cross-profile aggregate once', async () => { const calls: string[] = [] @@ -338,6 +423,36 @@ test('registry sources: shared remote hosts read the cross-profile aggregate onc ) }) +test('registry sources: large aggregate reads stay within the backend page cap', async () => { + const calls: string[] = [] + const rows = Array.from({ length: 250 }, (_, index) => ({ id: `r-${index}` })) + + const result = await fetchRegistrySessionRows( + [{ connectionId: 'gw-cloud', kind: 'remote', backends: [{ descriptor: 'cloud-desc', profileLabel: null }] }], + new URLSearchParams({ limit: '250', offset: '0' }), + async (_descriptor, path) => { + calls.push(path) + const url = new URL(path, 'http://desktop.test') + const limit = Number(url.searchParams.get('limit')) + const offset = Number(url.searchParams.get('offset')) + + assert.ok(limit <= 100) + + return { sessions: rows.slice(offset, offset + limit), total: rows.length } + } + ) + + assert.deepEqual( + calls.map(path => [Number(new URL(path, 'http://desktop.test').searchParams.get('limit')), Number(new URL(path, 'http://desktop.test').searchParams.get('offset'))]), + [ + [100, 0], + [100, 100], + [50, 200] + ] + ) + assert.equal(result.length, rows.length) +}) + test('registry-pinned session responses retain their owning connection', () => { const sidebar = tagRegistrySessionResponse( '/api/profiles/sessions/sidebar?recents_profile=default', diff --git a/apps/desktop/electron/profile-session-routing.ts b/apps/desktop/electron/profile-session-routing.ts index d0af80381c..54ebdad3b6 100644 --- a/apps/desktop/electron/profile-session-routing.ts +++ b/apps/desktop/electron/profile-session-routing.ts @@ -27,6 +27,30 @@ type FetchJsonForProfile = (profile: string | null, path: string) => Promise } +type PinnedRegistrySessionSource = Pick & + Partial> + +/** + * A pinned registry request may use the aggregate route only when that + * gateway is part of the already-pooled sources. Otherwise the aggregate + * would look healthy while silently omitting the gateway the renderer asked + * for, so the caller must keep the normal direct route instead. + * + * The legacy local/primary route is represented by the base aggregate rather + * than a registry source and is therefore handled by the caller separately. + */ +export function hasPinnedRegistrySessionSource( + connectionId: string | null | undefined, + profile: string | null | undefined, + sources: readonly PinnedRegistrySessionSource[], + baseCoversLocal = true +): boolean { + const required = String(connectionId ?? '').trim() + const selectedProfile = String(profile ?? '').trim() + + if (!required || (required === 'local' && baseCoversLocal)) { + return true + } + + const source = sources.find(candidate => candidate.connectionId === required) + + if (!source) { + return false + } + + // Shared remote and cloud hosts serve every profile from one backend. An + // absent profile is also allowed because the route can still be an explicit + // all-profiles request without an ambient renderer profile. + if (!selectedProfile || selectedProfile === 'all' || !source.kind || !source.backends) { + return true + } + + if (source.kind !== 'ssh' && source.kind !== 'local') { + return true + } + + return source.backends.some(({ profileLabel }) => (profileLabel || 'default') === selectedProfile) +} + type GetJsonForDescriptor = (descriptor: unknown, path: string) => Promise +/** Read one registry source without sending a page larger than the backend cap. */ +async function fetchSessionRowsInPages( + basePath: string, + searchParams: URLSearchParams, + getPage: (path: string) => Promise +): Promise { + const requestedLimit = Number(searchParams.get('limit')) + const requestedOffset = Number(searchParams.get('offset') || '0') + + const needsPaging = + Number.isInteger(requestedLimit) && + requestedLimit > REMOTE_SESSION_PAGE_LIMIT && + Number.isInteger(requestedOffset) && + requestedOffset >= 0 + + try { + if (!needsPaging) { + return rowsOf(await getPage(`${basePath}?${searchParams}`)) + } + + const sessions: unknown[] = [] + const backfilled: unknown[] = [] + const seenIds = new Set() + const backfilledIds = new Set() + let pageOffset = requestedOffset + let targetOffset = requestedOffset + requestedLimit + + while (pageOffset < targetOffset) { + const pageParams = new URLSearchParams(searchParams) + const pageLimit = Math.min(REMOTE_SESSION_PAGE_LIMIT, targetOffset - pageOffset) + pageParams.set('limit', String(pageLimit)) + pageParams.set('offset', String(pageOffset)) + + const page = await getPage(`${basePath}?${pageParams}`) + const pageRows = rowsOf(page) + + const total = nonNegativeNumber( + page && typeof page === 'object' ? (page as { total?: unknown }).total : null + ) + + const windowedCount = + total !== null ? Math.min(pageLimit, Math.max(0, total - pageOffset)) : Math.min(pageLimit, pageRows.length) + + for (const row of pageRows.slice(0, windowedCount)) { + const id = sessionId(row) + + if (id && seenIds.has(id)) { + continue + } + + if (id) { + seenIds.add(id) + backfilledIds.delete(id) + } + + sessions.push(row) + } + + for (const row of pageRows.slice(windowedCount)) { + const id = sessionId(row) + + if ((id && seenIds.has(id)) || (id && backfilledIds.has(id))) { + continue + } + + if (id) { + backfilledIds.add(id) + } + + backfilled.push(row) + } + + if (total !== null) { + targetOffset = Math.min(targetOffset, total) + } + + pageOffset += pageLimit + } + + for (const row of backfilled) { + const id = sessionId(row) + + if (id && seenIds.has(id)) { + continue + } + + if (id) { + seenIds.add(id) + } + + sessions.push(row) + } + + return sessions + } catch { + return null + } +} + /** * Every connected registry gateway's session rows for the unified Sessions * list (#88880), tagged with the owning `connection_id` + remote `profile` so @@ -275,8 +443,8 @@ export async function fetchRegistrySessionRows( ): Promise { const rows: unknown[] = [] - const tag = (data: unknown, connectionId: string, profileLabel: null | string) => { - for (const row of rowsOf(data)) { + const tag = (sourceRows: unknown[], connectionId: string, profileLabel: null | string) => { + for (const row of sourceRows) { if (!row || typeof row !== 'object') { continue } @@ -297,17 +465,18 @@ export async function fetchRegistrySessionRows( await Promise.all( sources.map(async source => { - if (source.kind === 'ssh') { - // Each ssh-scoped backend serves its own state.db natively. + if (source.kind === 'ssh' || source.kind === 'local') { + // Each ssh-scoped or forced-local backend serves its own state.db + // natively. await Promise.all( source.backends.map(async ({ descriptor, profileLabel }) => { const params = new URLSearchParams(searchParams) params.delete('profile') - const data = await getJson(descriptor, `/api/sessions?${params}`).catch(() => null) + const sourceRows = await fetchSessionRowsInPages('/api/sessions', params, path => getJson(descriptor, path)) - if (data) { - tag(data, source.connectionId, profileLabel || 'default') + if (sourceRows) { + tag(sourceRows, source.connectionId, profileLabel || 'default') } }) ) @@ -326,17 +495,19 @@ export async function fetchRegistrySessionRows( const params = new URLSearchParams(searchParams) params.set('profile', 'all') - let data = await getJson(shared.descriptor, `/api/profiles/sessions?${params}`).catch(() => null) + let sourceRows = await fetchSessionRowsInPages('/api/profiles/sessions', params, path => + getJson(shared.descriptor, path) + ) - if (!data) { + if (!sourceRows) { // Older remote without the aggregator: its own default-profile list. const flat = new URLSearchParams(searchParams) flat.delete('profile') - data = await getJson(shared.descriptor, `/api/sessions?${flat}`).catch(() => null) + sourceRows = await fetchSessionRowsInPages('/api/sessions', flat, path => getJson(shared.descriptor, path)) } - if (data) { - tag(data, source.connectionId, null) + if (sourceRows) { + tag(sourceRows, source.connectionId, null) } }) )