From fbbc141be05b86a6b9ddfeef1e32f398afadba50 Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 18:23:50 +0800 Subject: [PATCH] fix(desktop): restore retained shared sessions Generated-by: Codex --- apps/desktop/e2e/streaming-remount.spec.ts | 36 + apps/desktop/renderer-architecture.json | 10 +- .../desktop-session-projection.test.ts | 37 +- .../desktop-transcript-range-store.test.ts | 39 ++ .../main/__tests__/live-content-seed.test.ts | 26 + .../runtime-host-desktop-candidate.test.ts | 17 +- .../runtime-host-desktop-manager.test.ts | 6 +- .../runtime-host-guest-session-mounts.test.ts | 663 +++++++++++++++++- ...time-host-session-catalog-ipc-main.test.ts | 16 +- ...ntime-host-session-catalog-preload.test.ts | 369 ++++++++-- ...host-session-catalog-running-turns.test.ts | 23 +- .../session-collaboration-join-dialog.test.ts | 69 ++ .../main/__tests__/session-read-state.test.ts | 32 + .../session-settings-controller.test.ts | 6 +- .../session-turn-request-composer.test.ts | 2 + .../__tests__/use-onboarding-snapshot.test.ts | 43 -- apps/desktop/src/main/runtime-host-boot.ts | 45 +- .../main/runtime-host-desktop-candidate.ts | 22 +- .../src/main/runtime-host-desktop-manager.ts | 13 +- .../main/runtime-host-guest-session-mounts.ts | 497 +++++++++++-- .../runtime-host-session-catalog-ipc-main.ts | 50 -- apps/desktop/src/preload/bridge-contract.d.ts | 1 + apps/desktop/src/preload/preload.ts | 257 +++++-- .../preload/runtime-host-session-catalog.ts | 182 +++-- .../desktop/src/renderer/app-shell-effects.ts | 84 +-- apps/desktop/src/renderer/app-shell.tsx | 114 ++- .../desktop/src/renderer/astryx-theme/maka.js | 2 +- .../desktop-transcript-range-store.ts | 95 +++ .../features/session-collaboration/ports.ts | 1 + .../ui/session-collaboration-join-dialog.tsx | 104 ++- .../desktop/src/renderer/live-content-seed.ts | 35 + .../locales/session-collaboration-copy.ts | 12 + .../create-session-collaboration-services.ts | 7 +- .../src/renderer/session-read-state.ts | 40 +- .../runtime-host-profiles-section.tsx | 17 +- .../src/renderer/use-onboarding-snapshot.ts | 28 +- .../src/shared/desktop-session-projection.ts | 6 +- .../src/shared/runtime-host-identity.ts | 4 +- .../src/shared/session-collaboration.d.ts | 7 + .../shared-session-catalog-projection.ts | 61 ++ .../stories/shared-session-guest.stories.tsx | 1 + docs/astryx-surface-file-inventory.md | 2 +- scripts/build-astryx-theme.mjs | 5 +- 43 files changed, 2467 insertions(+), 619 deletions(-) create mode 100644 apps/desktop/src/shared/shared-session-catalog-projection.ts diff --git a/apps/desktop/e2e/streaming-remount.spec.ts b/apps/desktop/e2e/streaming-remount.spec.ts index d0f238d988..f72efb1f81 100644 --- a/apps/desktop/e2e/streaming-remount.spec.ts +++ b/apps/desktop/e2e/streaming-remount.spec.ts @@ -35,6 +35,7 @@ interface SessionObservationLatchWindow extends Window { /** E2E-only preload affordance; see the MAKA_E2E block in preload.ts. */ makaE2eLatch?: { rejectNextSessionObservation(message: string): void; + rejectNextTranscriptOpen(message: string): void; }; } @@ -73,6 +74,41 @@ test('a failed first observation seed reconnects to the live Turn', async ({ win }); }); +test('a failed transcript open recovers when its Session observation becomes ready', async ({ + window: page, +}) => { + const originalPrompt = 'transcript recovery source'; + const composer = page.locator(COMPOSER_INPUT); + await composer.fill(originalPrompt); + await awaitSendReady(page); + await composer.press('Enter'); + await expect(page.getByRole('log')).toContainText(`Fake backend received: ${originalPrompt}`, { + timeout: 20_000, + }); + + const sidebar = page.getByRole('navigation', { name: '任务列表' }); + await ensureSidebarExpanded(page); + const originalSessionId = await sidebar + .locator('[data-session-id]:has([aria-current="page"])') + .getAttribute('data-session-id'); + expect(originalSessionId).toBeTruthy(); + + await sidebar.getByRole('button', { name: '新任务', exact: true }).click(); + + const latchInstalled = await page.evaluate(() => { + const latch = (window as SessionObservationLatchWindow).makaE2eLatch; + if (!latch) return false; + latch.rejectNextTranscriptOpen('forced first transcript failure'); + return true; + }); + expect(latchInstalled, 'the preload E2E latch is installed').toBe(true); + + await sessionRow(sidebar, originalSessionId!).click(); + await expect(page.getByRole('log')).toContainText(`Fake backend received: ${originalPrompt}`, { + timeout: 20_000, + }); +}); + test('remounting a live surface leaves accumulated output settled', async ({ window: page, }) => { diff --git a/apps/desktop/renderer-architecture.json b/apps/desktop/renderer-architecture.json index d05fbe5535..8fea02ce9f 100644 --- a/apps/desktop/renderer-architecture.json +++ b/apps/desktop/renderer-architecture.json @@ -546,8 +546,8 @@ "@maka/ui": 1, "react": 1 }, - "importSpecifiers": 39, - "nonTriviaTokens": 3836 + "importSpecifiers": 37, + "nonTriviaTokens": 3823 }, "src/renderer/app-shell-overlays.tsx": { "importDeclarations": 14, @@ -851,7 +851,7 @@ "useNewTaskChoice": 1, "useOnboardingSnapshot": 1, "usePlanModeState": 1, - "useRef": 25, + "useRef": 24, "useSessionCollaborationDialog": 1, "useSessionEventHealthPolling": 1, "useSessionNavigationReads": 1, @@ -979,8 +979,8 @@ "@maka/ui/icons": 1, "react": 1 }, - "importSpecifiers": 184, - "nonTriviaTokens": 15686 + "importSpecifiers": 180, + "nonTriviaTokens": 15620 }, "src/renderer/use-app-shell-composer-quotes.ts": { "importDeclarations": 3, diff --git a/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts b/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts index d0d96bfae0..37be37826e 100644 --- a/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts @@ -30,7 +30,8 @@ import { projectDesktopTurnRecord, projectDesktopUsageStats, } from '../../shared/desktop-session-projection.js'; -import { runtimeHostChangeRetiresSession } from '../../shared/runtime-host-identity.js'; +import { projectDesktopSharedSessionSummary } from '../../shared/shared-session-catalog-projection.js'; +import { sessionCatalogRetiresSession } from '../../shared/runtime-host-identity.js'; test('keeps equal raw Session ids distinct across Runtime Hosts', () => { const raw = summary('same-session'); @@ -60,7 +61,22 @@ test('keeps equal raw Session ids distinct across Runtime Hosts', () => { assert.equal(remote.profileName, 'Office'); }); -test('retires an active Session only after it leaves the refreshed Host catalog', () => { +test('preserves the authenticated shared Session revision', () => { + assert.equal( + projectDesktopSharedSessionSummary({ + kind: 'shared_session', + id: 'shared-session', + revision: 7, + createdAt: 1, + activityAt: 2, + name: 'Shared', + status: 'active', + }).revision, + 7, + ); +}); + +test('retires an active Session only after it leaves the refreshed catalog', () => { const owner = projectDesktopSessionSummary( { hostId: 'shared-root', @@ -79,20 +95,9 @@ test('retires an active Session only after it leaves the refreshed Host catalog' }, summary('shared-session'), ); - const removedGuest = { - epoch: 'guest-epoch', - profileId: 'guest', - profileName: 'Guest', - profileKind: 'remote', - profileAccess: 'session_guest', - readiness: 'unavailable', - hostId: 'shared-root', - isDefault: false, - removed: true, - } as const; - - assert.equal(runtimeHostChangeRetiresSession(removedGuest, guest.id, [owner]), false); - assert.equal(runtimeHostChangeRetiresSession(removedGuest, guest.id, []), true); + assert.equal(sessionCatalogRetiresSession(guest.id, [owner]), false); + assert.equal(sessionCatalogRetiresSession(guest.id, []), true); + assert.equal(sessionCatalogRetiresSession(undefined, []), false); }); test('projects typed linked Session ids without rewriting opaque tool data', () => { diff --git a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts index ba076c2fc0..c1ee17d385 100644 --- a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts @@ -31,6 +31,7 @@ import { DESKTOP_TRANSCRIPT_RANGE_MAX_BYTES, } from '../../preload/transcript-contract.js'; import { + createDesktopTranscriptReconnectRecovery, createDesktopTranscriptRangeController, DesktopTranscriptRangeStore, } from '../../renderer/desktop-transcript-range-store.js'; @@ -1324,6 +1325,44 @@ test('reopens a failed transcript range with a fresh generation', async () => { await controller.close(); }); +test('retries a failed transcript recovery after a newer observation becomes ready', async () => { + let rejectFirstReload!: (error: Error) => void; + const firstReload = new Promise((_resolve, reject) => { + rejectFirstReload = reject; + }); + let resolveSecondReload!: () => void; + const secondReload = new Promise((resolve) => { + resolveSecondReload = resolve; + }); + const reloads: Promise[] = [firstReload, secondReload]; + const errors: string[] = []; + const recovery = createDesktopTranscriptReconnectRecovery({ + reload: () => { + const reload = reloads.shift(); + if (!reload) throw new Error('unexpected transcript reload'); + return reload; + }, + onError(error) { + errors.push(error instanceof Error ? error.message : String(error)); + }, + }); + + recovery.transcriptFailed(new Error('initial open failed')); + recovery.observationChanged('ready'); + await Promise.resolve(); + recovery.observationChanged('pending'); + recovery.observationChanged('ready'); + rejectFirstReload(new Error('replaced transcript failed')); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(reloads.length, 0, 'the newer ready signal starts one trailing reload'); + resolveSecondReload(); + await new Promise((resolve) => setImmediate(resolve)); + + assert.deepEqual(errors, ['initial open failed', 'replaced transcript failed']); + recovery.close(); +}); + test('forwards a larger logical history range without changing batch size', async () => { const store = transcriptStore(); for (const batch of encodeDesktopTranscriptSnapshot({ diff --git a/apps/desktop/src/main/__tests__/live-content-seed.test.ts b/apps/desktop/src/main/__tests__/live-content-seed.test.ts index 8322a8a43d..92fa5eaf57 100644 --- a/apps/desktop/src/main/__tests__/live-content-seed.test.ts +++ b/apps/desktop/src/main/__tests__/live-content-seed.test.ts @@ -21,11 +21,37 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; import { EMPTY_LIVE_CONTENT_SEED, + EMPTY_SESSION_OBSERVATION_AUTHORITY, + advanceSessionObservationAuthority, beginLiveContentSeed, completeLiveContentSeed, liveContentSeedRevision, } from '../../renderer/live-content-seed.js'; +test('catalog hydration does not replace an already-bound Session observation', () => { + const selected = advanceSessionObservationAuthority( + EMPTY_SESSION_OBSERVATION_AUTHORITY, + 'session-a', + undefined, + ); + const hydrated = advanceSessionObservationAuthority(selected, 'session-a', 'profile-a'); + + assert.equal(hydrated.profileId, 'profile-a'); + assert.equal(hydrated.revision, selected.revision); +}); + +test('a real Session observation authority handoff advances the revision', () => { + const selected = advanceSessionObservationAuthority( + EMPTY_SESSION_OBSERVATION_AUTHORITY, + 'session-a', + 'profile-a', + ); + const handedOff = advanceSessionObservationAuthority(selected, 'session-a', 'profile-b'); + + assert.equal(handedOff.profileId, 'profile-b'); + assert.equal(handedOff.revision, selected.revision + 1); +}); + test('withholds live content until the current observation generation is ready', () => { const first = beginLiveContentSeed(EMPTY_LIVE_CONTENT_SEED, 'session-a'); assert.equal(liveContentSeedRevision(first, 'session-a'), 0); diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts index 294ed24592..b6be2f9cd4 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts @@ -185,11 +185,12 @@ test('owns one complete Desktop candidate generation and can restart cleanly', a assert.equal(ipc.size, 0); }); -test('registers only shared observation IPC and consumes scoped catalog changes for a Guest', async () => { +test('routes Guest catalog changes through the mount projection authority', async () => { const ipc = ipcHarness(); const sharedResource = sharedShellRunUpdate('session-guest'); const host = connectionHarness('guest', { runtimeResourceUpdate: sharedResource }); const changes: Array<{ reason: string; sessionId?: string }> = []; + let catalogChanges = 0; const rendererEvents: Array<{ channel: string; payload: unknown }> = []; const candidate = await createCandidate( host.connection, @@ -198,6 +199,9 @@ test('registers only shared observation IPC and consumes scoped catalog changes emitSessionsChanged: (_scope, reason, sessionId) => { changes.push({ reason, ...(sessionId === undefined ? {} : { sessionId }) }); }, + onGuestSessionCatalogChanged: () => { + catalogChanges += 1; + }, renderer: { send(channel, _scope, payload) { rendererEvents.push({ channel, payload }); @@ -210,10 +214,7 @@ test('registers only shared observation IPC and consumes scoped catalog changes 'session_guest', ); - assert.deepEqual( - ((await ipc.invoke('sessions:list')) as SessionCatalogProjection[]).map(({ id }) => id), - ['session-guest'], - ); + assert.equal(ipc.channels.includes('sessions:list'), false); assert.equal(ipc.channels.includes('sessions:observe'), true); assert.equal(ipc.channels.includes('sessions:transcript:open'), true); assert.equal(ipc.channels.includes('sessions:send'), false); @@ -242,7 +243,8 @@ test('registers only shared observation IPC and consumes scoped catalog changes ), ); host.publishSessionCatalogChange('session-guest'); - assert.deepEqual(changes, [{ reason: 'updated', sessionId: 'session-guest' }]); + assert.equal(catalogChanges, 1); + assert.deepEqual(changes, []); await candidate.close(); }); @@ -859,7 +861,7 @@ test('drops a stale shared Session observation when Guest access is gone', async }); const firstCandidate = await createCandidate( firstHost.connection, - deps(firstIpc), + { ...deps(firstIpc), onGuestSessionCatalogChanged: () => undefined }, observations, 'external', 'remote', @@ -879,6 +881,7 @@ test('drops a stale shared Session observation when Guest access is gone', async emitSessionsChanged: (_scope, reason, sessionId) => { changes.push({ reason, ...(sessionId === undefined ? {} : { sessionId }) }); }, + onGuestSessionCatalogChanged: () => undefined, }, observations, 'external', diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts index 7320f3e868..d4f432b498 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts @@ -605,8 +605,8 @@ test('keeps independent shared-session credentials active for the same Host', as { startCandidate: async () => ready(candidates.shift()!) }, ); - await manager.mountGuest(remoteTarget('shared-one', 'shared', 'session_guest')); - await manager.mountGuest(remoteTarget('shared-two', 'shared', 'session_guest')); + await manager.mountGuest(remoteTarget('shared-one', 'shared', 'session_guest'), () => undefined); + await manager.mountGuest(remoteTarget('shared-two', 'shared', 'session_guest'), () => undefined); await manager.enable(remoteTarget('owner', 'shared')); assert.deepEqual(manager.entries().map(({ target }) => target.profile.id), [ @@ -642,6 +642,7 @@ test('aborts an in-flight Guest mount without publishing a late target', async ( const abort = new AbortController(); const mounting = manager.mountGuest( remoteTarget('shared-cancelled', 'shared', 'session_guest'), + () => undefined, abort.signal, ); await started; @@ -746,6 +747,7 @@ test('completes Guest import at credential activation while reconnect continues' ); await manager.mountGuest( peerGuestTarget('shared-session'), + () => undefined, undefined, (phase) => phases.push(phase), ); diff --git a/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts b/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts index 08a02fec32..ce367beb73 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts @@ -21,15 +21,18 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { RuntimeHostPermanentReconnectError, + RuntimeHostProfileConnectionError, type ResolvedRuntimeHostProfile, } from '@maka/runtime-host/client'; import { encodeCollaborationInvitationCode, type HostPeerEndpoint, + type SharedSessionCatalogProjection, } from '@maka/runtime-host/protocol'; import { encodeDesktopCollaborationInvitation } from '../runtime-host-collaboration-invitation.js'; import { createDesktopGuestSessionMountService, + createGuestSessionMountStore, type GuestSessionMount, type GuestSessionMountStore, registerDesktopGuestSessionMountIpc, @@ -39,7 +42,7 @@ import { RuntimeHostPairingFinalizationInterruptedError } from '../runtime-host- const ROOT_ID = 'a'.repeat(64); test('retains a successful Guest mount and rehydrates the same authority after restart', async () => { - const store = memoryStore(); + const store = serializedStore(); const activated: string[] = []; const first = service(store, { mount: async (target) => { @@ -61,9 +64,112 @@ test('retains a successful Guest mount and rehydrates the same authority after r }); assert.equal(await rehydrated, `${result.mountId}:guest-one`); assert.deepEqual(activated, [`${result.mountId}:guest-one`]); + assert.deepEqual((await second!.list())[0]?.session, sharedSession()); await second!.close(); }); +test('keeps live Guest state in memory without persisting ephemeral run identities', async () => { + const store = serializedStore(); + const live = { + ...sharedSession(), + liveRunState: { schemaVersion: 1 as const, runningTurnIds: ['turn-live'] }, + }; + const mounts = service(store, { getSharedSession: async () => live }); + + assert.equal( + (await mounts.importInvitation(invitation('guest-live'), false, 'live')).kind, + 'connected', + ); + assert.deepEqual((await mounts.list())[0]?.session?.liveRunState, live.liveRunState); + assert.equal((await store.read())[0]?.session?.liveRunState, undefined); + await mounts.close(); +}); + +test('collapses fresh credentials for the same authenticated shared Session', async () => { + const store = memoryStore(); + const unmounted: string[] = []; + const mounts = service(store, { + unmount: async (mountId) => { + unmounted.push(mountId); + }, + }); + + const first = await mounts.importInvitation(invitation('guest-first'), false, 'first'); + const second = await mounts.importInvitation(invitation('guest-second'), false, 'second'); + assert.equal(first.kind, 'connected'); + assert.equal(second.kind, 'connected'); + if (first.kind !== 'connected' || second.kind !== 'connected') return; + + const retained = await mounts.list(); + assert.deepEqual( + retained.map(({ mountId }) => mountId), + [second.mountId], + ); + assert.deepEqual( + (await store.read()).map(({ mountId }) => mountId), + [second.mountId], + ); + assert.deepEqual(unmounted, [first.mountId]); + await mounts.close(); +}); + +test('unmounts a superseded Guest when its replacement is rejected during projection commit', async () => { + const superseded = { + ...retainedMount('shared-superseded'), + session: sharedSession(), + }; + let durable: readonly GuestSessionMount[] = [superseded]; + let replacementId: string | undefined; + let markProjectionWrite!: () => void; + let releaseProjectionWrite!: () => void; + const projectionWrite = new Promise((resolve) => { + markProjectionWrite = resolve; + }); + const projectionWriteReleased = new Promise((resolve) => { + releaseProjectionWrite = resolve; + }); + const store: GuestSessionMountStore = { + read: async () => durable, + write: async (next) => { + const replacement = next.find((mount) => mount.mountId !== superseded.mountId); + if (next.length === 1 && replacement?.session) { + replacementId = replacement.mountId; + markProjectionWrite(); + await projectionWriteReleased; + } + durable = next; + }, + }; + const unmounted: string[] = []; + const mounts = service(store, { + finalizeAccess: async (_mountId, _signal, onAccessActivated) => { + onAccessActivated?.(); + return 'ready'; + }, + unmount: async (mountId) => { + unmounted.push(mountId); + }, + }); + + const importing = mounts.importInvitation(invitation('guest-replacement'), false, 'replace'); + await projectionWrite; + assert.ok(replacementId); + const rejecting = mounts.connectionChanged( + replacementId, + new RuntimeHostProfileConnectionError( + 'credential_rejected', + 'Shared Session access was revoked', + ), + ); + releaseProjectionWrite(); + + assert.equal((await importing).kind, 'error'); + await rejecting; + assert.deepEqual(durable, []); + assert.deepEqual(new Set(unmounted), new Set([superseded.mountId, replacementId])); + await mounts.close(); +}); + test('reports activated Guest access as recovering while reauthentication continues', async () => { const progress: string[] = []; const mounts = service(memoryStore(), { @@ -106,13 +212,10 @@ test('persists authenticated route rotation for reconnect and restart', async () let observePeerEndpoint!: (endpoint: HostPeerEndpoint) => void; const first = service(store, { mount: async (target, _signal, _onConnectionPhase, onPeerEndpoint) => { - assert.deepEqual( - target.profile.kind === 'remote' ? target.profile.transport : undefined, - { - kind: 'libp2p-direct', - reachability: guestPeerReachability(), - }, - ); + assert.deepEqual(target.profile.kind === 'remote' ? target.profile.transport : undefined, { + kind: 'libp2p-direct', + reachability: guestPeerReachability(), + }); assert.ok(onPeerEndpoint); observePeerEndpoint = onPeerEndpoint; }, @@ -141,6 +244,72 @@ test('persists authenticated route rotation for reconnect and restart', async () await restarted.close(); }); +test('retries authenticated route rotation after transient persistence failure', async () => { + let durable: readonly GuestSessionMount[] = []; + let writesBlocked = false; + const store: GuestSessionMountStore = { + read: async () => durable, + write: async (next) => { + if (writesBlocked) throw new Error('credential store is locked'); + durable = next; + }, + }; + let observePeerEndpoint!: (endpoint: HostPeerEndpoint) => void; + let markFailure!: () => void; + const failureReported = new Promise((resolve) => { + markFailure = resolve; + }); + const first = service(store, { + mount: async (_target, _signal, _onConnectionPhase, onPeerEndpoint) => { + assert.ok(onPeerEndpoint); + observePeerEndpoint = onPeerEndpoint; + }, + onError: () => markFailure(), + }); + const imported = await first.importInvitation(peerInvitation('guest-routes'), false, 'routes'); + assert.equal(imported.kind, 'connected'); + + const rotated = guestPeerReachability( + 2, + ['/ip4/198.51.100.2/udp/42000/quic-v1'], + ['/memory/fresh-relay'], + ); + writesBlocked = true; + observePeerEndpoint(rotated); + await failureReported; + + writesBlocked = false; + await first.close(); + + let restarted!: ReturnType; + const restartedTarget = new Promise((resolve) => { + restarted = service(store, { mount: async (target) => resolve(target) }); + void restarted.start(); + }); + const target = await restartedTarget; + assert.deepEqual(target.profile.kind === 'remote' ? target.profile.transport : undefined, { + kind: 'libp2p-direct', + reachability: rotated, + }); + await restarted.close(); +}); + +test('derives retained mount readiness from the Runtime Host connection owner', async () => { + const store = memoryStore(); + await store.write([{ ...retainedMount('shared-readiness'), session: sharedSession() }]); + let connectionReadiness: 'ready' | 'reconnecting' | 'unavailable' = 'reconnecting'; + const mounts = service(store, { + inspect: () => ({ readiness: connectionReadiness }), + }); + + assert.equal((await mounts.list())[0]?.readiness, 'reconnecting'); + connectionReadiness = 'ready'; + assert.equal((await mounts.list())[0]?.readiness, 'ready'); + connectionReadiness = 'unavailable'; + assert.equal((await mounts.list())[0]?.readiness, 'unavailable'); + await mounts.close(); +}); + test('removes failed activation desire instead of creating recoverable profile state', async () => { const store = memoryStore(); const unmounted: string[] = []; @@ -163,14 +332,19 @@ test('removes failed activation desire instead of creating recoverable profile s test('does not retry a startup mount whose reachability recovery is exhausted', async () => { const store = memoryStore(); - await store.write([retainedMount('shared-needs-repair')]); + await store.write([{ ...retainedMount('shared-needs-repair'), session: sharedSession() }]); let attempts = 0; let waits = 0; let reportFailure!: () => void; + let reportUnavailable!: () => void; const failureReported = new Promise((resolve) => { reportFailure = resolve; }); - const mounts = service(store, { + const unavailableReported = new Promise((resolve) => { + reportUnavailable = resolve; + }); + let mounts!: ReturnType; + mounts = service(store, { mount: async () => { attempts += 1; throw new RuntimeHostPermanentReconnectError('reachability recovery exhausted'); @@ -179,13 +353,322 @@ test('does not retry a startup mount whose reachability recovery is exhausted', waits += 1; }, onError: () => reportFailure(), + onMountsChanged: () => { + void mounts.list().then(([mount]) => { + if (mount?.readiness === 'unavailable') reportUnavailable(); + }); + }, + inspect: () => ({ readiness: 'unavailable' }), }); await mounts.start(); await failureReported; - await new Promise((resolve) => setImmediate(resolve)); + await unavailableReported; assert.equal(attempts, 1); assert.equal(waits, 0); + assert.equal((await mounts.list())[0]?.readiness, 'unavailable'); + assert.deepEqual((await store.read())[0]?.session, sharedSession()); + await mounts.close(); +}); + +test('retires a retained Session projection only after explicit access rejection', async () => { + const store = memoryStore(); + const retained = { + ...retainedMount('shared-revoked'), + session: sharedSession(), + }; + await store.write([retained]); + let mountChanges = 0; + const mounts = service(store, { + onMountsChanged: () => { + mountChanges += 1; + }, + }); + + await mounts.connectionChanged( + retained.mountId, + new RuntimeHostProfileConnectionError( + 'credential_rejected', + 'Shared Session access was revoked', + ), + ); + + const [visible] = await mounts.list(); + assert.equal(visible?.readiness, 'unavailable'); + assert.equal(visible?.session, undefined); + assert.equal((await store.read())[0]?.session, undefined); + assert.equal(mountChanges, 1); + await mounts.close(); +}); + +test('fails closed when a revoked Session projection cannot be persisted immediately', async () => { + const retained = { + ...retainedMount('shared-write-locked'), + session: sharedSession(), + }; + let durable: readonly GuestSessionMount[] = [retained]; + let writesBlocked = true; + const store: GuestSessionMountStore = { + read: async () => durable, + write: async (next) => { + if (writesBlocked) throw new Error('credential store is locked'); + durable = next; + }, + }; + let mountChanges = 0; + const mounts = service(store, { + onMountsChanged: () => { + mountChanges += 1; + }, + }); + + await assert.rejects( + mounts.connectionChanged( + retained.mountId, + new RuntimeHostProfileConnectionError( + 'credential_rejected', + 'Shared Session access was revoked', + ), + ), + /credential store is locked/u, + ); + + const [visible] = await mounts.list(); + assert.equal(visible?.readiness, 'unavailable'); + assert.equal(visible?.session, undefined); + assert.ok(durable[0]?.session); + assert.equal(mountChanges, 1); + + writesBlocked = false; + await mounts.close(); + assert.equal(durable[0]?.session, undefined); +}); + +test('publishes a refreshed Guest projection while persistence is unavailable', async () => { + let durable: readonly GuestSessionMount[] = []; + let writesBlocked = false; + let markPersisted!: () => void; + const persisted = new Promise((resolve) => { + markPersisted = resolve; + }); + const store: GuestSessionMountStore = { + read: async () => durable, + write: async (next) => { + if (writesBlocked) throw new Error('credential store is locked'); + durable = next; + if (durable[0]?.session?.revision === 2) markPersisted(); + }, + }; + let markRetryScheduled!: () => void; + let releaseRetry!: () => void; + const retryScheduled = new Promise((resolve) => { + markRetryScheduled = resolve; + }); + const retryReleased = new Promise((resolve) => { + releaseRetry = resolve; + }); + let catalogChanged!: () => void; + let projection = sharedSession(); + let mountChanges = 0; + const errors: Error[] = []; + let markErrorReported!: () => void; + const errorReported = new Promise((resolve) => { + markErrorReported = resolve; + }); + let markRefreshed: (() => void) | undefined; + const mounts = service(store, { + mount: async (_target, _signal, _onConnectionPhase, _onPeerEndpoint, onChanged) => { + assert.ok(onChanged); + catalogChanged = onChanged; + }, + getSharedSession: async () => projection, + onMountsChanged: () => { + mountChanges += 1; + markRefreshed?.(); + }, + wait: async () => { + markRetryScheduled(); + await retryReleased; + }, + onError: (error) => { + errors.push(error); + markErrorReported(); + }, + }); + const joined = await mounts.importInvitation(invitation('guest-refresh'), false, 'refresh'); + assert.equal(joined.kind, 'connected'); + if (joined.kind !== 'connected') return; + mountChanges = 0; + + projection = { + ...projection, + revision: 2, + activityAt: 3, + name: 'Fresh task', + }; + writesBlocked = true; + const refreshed = new Promise((resolve) => { + markRefreshed = resolve; + }); + catalogChanged(); + assert.equal(mountChanges, 0); + await refreshed; + + const [visible] = await mounts.list(); + assert.equal(visible?.session?.revision, 2); + assert.equal(durable[0]?.session?.revision, 1); + await errorReported; + assert.equal(errors.length, 1); + assert.equal(mountChanges, 1); + + await retryScheduled; + writesBlocked = false; + releaseRetry(); + await persisted; + assert.equal(durable[0]?.session?.revision, 2); + await mounts.close(); +}); + +test('serves a published Guest projection while its durable write is pending', async () => { + let durable: readonly GuestSessionMount[] = []; + let projection = sharedSession(); + let blockFreshProjection = false; + let markWriteStarted!: () => void; + let releaseWrite!: () => void; + const writeStarted = new Promise((resolve) => { + markWriteStarted = resolve; + }); + const writeReleased = new Promise((resolve) => { + releaseWrite = resolve; + }); + const store: GuestSessionMountStore = { + read: async () => durable, + write: async (next) => { + if (blockFreshProjection && next[0]?.session?.revision === 2) { + markWriteStarted(); + await writeReleased; + } + durable = next; + }, + }; + const mounts = service(store, { + getSharedSession: async () => projection, + }); + const joined = await mounts.importInvitation(invitation('guest-pending'), false, 'pending'); + assert.equal(joined.kind, 'connected'); + if (joined.kind !== 'connected') return; + + projection = { ...projection, revision: 2, activityAt: 3, name: 'Fresh task' }; + blockFreshProjection = true; + const refreshing = mounts.connectionChanged(joined.mountId); + await writeStarted; + + assert.equal((await mounts.list())[0]?.session?.revision, 2); + assert.equal(durable[0]?.session?.revision, 1); + + releaseWrite(); + await refreshing; + assert.equal(durable[0]?.session?.revision, 2); + await mounts.close(); +}); + +test('cancels an admitted Guest projection refresh during shutdown', async () => { + const store = memoryStore(); + const retained = { ...retainedMount('shared-close-refresh'), session: sharedSession() }; + await store.write([retained]); + let markReadStarted!: () => void; + const readStarted = new Promise((resolve) => { + markReadStarted = resolve; + }); + const mounts = service(store, { + getSharedSession: async () => { + markReadStarted(); + return new Promise(() => undefined); + }, + }); + + const refreshOutcome = mounts.connectionChanged(retained.mountId).catch((error: unknown) => error); + await readStarted; + await settlePromptly(mounts.close()); + + assert.match(String(await refreshOutcome), /closed/); + assert.deepEqual((await store.read())[0]?.session, sharedSession()); +}); + +test('retains activated Guest access when shutdown cancels initial projection hydration', async () => { + const store = memoryStore(); + let markReadStarted!: () => void; + const readStarted = new Promise((resolve) => { + markReadStarted = resolve; + }); + const mounts = service(store, { + getSharedSession: async () => { + markReadStarted(); + return new Promise(() => undefined); + }, + }); + + const importing = mounts.importInvitation(invitation('guest-close'), false, 'close'); + await readStarted; + const [result] = await Promise.all([ + settlePromptly(importing), + settlePromptly(mounts.close()), + ]); + + assert.equal(result.kind, 'recovering'); + assert.equal((await store.read()).length, 1); +}); + +test('does not lose a catalog invalidation that races Guest activation', async () => { + const store = memoryStore(); + let catalogChanged!: () => void; + let releaseInitialRead!: () => void; + let markInitialRead!: () => void; + let markProjectionCleared!: () => void; + const initialRead = new Promise((resolve) => { + markInitialRead = resolve; + }); + const initialReadReleased = new Promise((resolve) => { + releaseInitialRead = resolve; + }); + const projectionCleared = new Promise((resolve) => { + markProjectionCleared = resolve; + }); + let reads = 0; + const mounts = service(store, { + mount: async (_target, _signal, _onConnectionPhase, _onPeerEndpoint, onChanged) => { + assert.ok(onChanged); + catalogChanged = onChanged; + }, + getSharedSession: async () => { + reads += 1; + if (reads === 1) { + markInitialRead(); + await initialReadReleased; + return sharedSession(); + } + return null; + }, + inspect: () => ({ readiness: 'ready' }), + onMountsChanged: () => { + void mounts.list().then(([mount]) => { + if (reads >= 2 && mount && mount.session === undefined) markProjectionCleared(); + }); + }, + }); + + const importing = mounts.importInvitation(invitation('guest-removed'), false, 'removed'); + await initialRead; + catalogChanged(); + releaseInitialRead(); + assert.equal((await importing).kind, 'connected'); + await projectionCleared; + + const [visible] = await mounts.list(); + assert.equal(visible?.readiness, 'unavailable'); + assert.equal(visible?.session, undefined); + assert.equal((await store.read())[0]?.session, undefined); + assert.equal(reads, 2); await mounts.close(); }); @@ -340,6 +823,38 @@ test('settles admitted finalization before closing and retains the mount', async assert.equal((await store.read()).length, 1); }); +test('does not enter startup retry backoff after closing during finalization', async () => { + const store = memoryStore(); + await store.write([retainedMount('shared-closing-startup')]); + let markFinalizing!: () => void; + let finishFinalizing!: () => void; + const finalizing = new Promise((resolve) => { + markFinalizing = resolve; + }); + const finalizationReleased = new Promise((resolve) => { + finishFinalizing = resolve; + }); + let waits = 0; + const mounts = service(store, { + finalizeAccess: async () => { + markFinalizing(); + await finalizationReleased; + return 'reconnecting'; + }, + wait: async () => { + waits += 1; + }, + }); + + await mounts.start(); + await finalizing; + const closing = mounts.close(); + finishFinalizing(); + await closing; + + assert.equal(waits, 0); +}); + test('retains and reconciles a mount when finalization outcome is unknown', async () => { const store = memoryStore(); let attempts = 0; @@ -356,7 +871,11 @@ test('retains and reconciles a mount when finalization outcome is unknown', asyn }, }); - const result = await mounts.importInvitation(invitation('guest-unknown'), false, 'import-unknown'); + const result = await mounts.importInvitation( + invitation('guest-unknown'), + false, + 'import-unknown', + ); assert.equal(result.kind, 'recovering'); assert.equal((await store.read()).length, 1); await reconciled; @@ -365,6 +884,67 @@ test('retains and reconciles a mount when finalization outcome is unknown', asyn await mounts.close(); }); +test('finishes a committed credential reconnect and records its Session projection', async () => { + const store = memoryStore(); + let attempts = 0; + let markAvailable!: () => void; + const available = new Promise((resolve) => { + markAvailable = resolve; + }); + const mounts = service(store, { + finalizeAccess: async () => { + attempts += 1; + return attempts === 1 ? 'reconnecting' : 'ready'; + }, + onMountsChanged: () => { + void store.read().then(([mount]) => { + if (mount?.session) markAvailable(); + }); + }, + wait: async () => undefined, + }); + + const result = await mounts.importInvitation(invitation('guest-rotated'), false, 'rotated'); + assert.equal(result.kind, 'recovering'); + await available; + + assert.equal(attempts, 2); + assert.equal((await mounts.list())[0]?.readiness, 'ready'); + assert.deepEqual((await store.read())[0]?.session, sharedSession()); + await mounts.close(); +}); + +test('retains a finalized mount when its first Session projection read is interrupted', async () => { + const store = memoryStore(); + let reads = 0; + let markAvailable!: () => void; + const available = new Promise((resolve) => { + markAvailable = resolve; + }); + const mounts = service(store, { + getSharedSession: async () => { + reads += 1; + if (reads === 1) throw new Error('connection changed after credential finalization'); + return sharedSession(); + }, + onMountsChanged: () => { + void store.read().then(([mount]) => { + if (mount?.session) markAvailable(); + }); + }, + wait: async () => undefined, + }); + + const result = await mounts.importInvitation(invitation('guest-finalized'), false, 'finalized'); + assert.equal(result.kind, 'recovering'); + assert.equal((await store.read()).length, 1); + await available; + + assert.equal(reads, 2); + assert.deepEqual((await store.read())[0]?.session, sharedSession()); + await mounts.close(); +}); + test('cancels an in-flight import and removes its durable mount desire', async () => { const store = memoryStore(); let connecting!: () => void; @@ -375,7 +955,9 @@ test('cancels an in-flight import and removes its durable mount desire', async ( mount: async (_target, signal) => { connecting(); await new Promise((_resolve, reject) => { - signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + signal.addEventListener('abort', () => reject(signal.reason), { + once: true, + }); }); }, }); @@ -443,7 +1025,16 @@ function service( store: GuestSessionMountStore, overrides: { readonly mount?: Parameters[0]['mount']; - readonly finalizeAccess?: Parameters[0]['finalizeAccess']; + readonly finalizeAccess?: Parameters< + typeof createDesktopGuestSessionMountService + >[0]['finalizeAccess']; + readonly getSharedSession?: Parameters< + typeof createDesktopGuestSessionMountService + >[0]['getSharedSession']; + readonly inspect?: Parameters[0]['inspect']; + readonly onMountsChanged?: Parameters< + typeof createDesktopGuestSessionMountService + >[0]['onMountsChanged']; readonly unmount?: Parameters[0]['unmount']; readonly wait?: Parameters[0]['wait']; readonly onError?: Parameters[0]['onError']; @@ -453,12 +1044,27 @@ function service( store, mount: overrides.mount ?? (async () => undefined), finalizeAccess: overrides.finalizeAccess ?? (async () => 'ready'), + getSharedSession: overrides.getSharedSession ?? (async () => sharedSession()), + inspect: overrides.inspect ?? (() => ({ readiness: 'ready' })), + onMountsChanged: overrides.onMountsChanged ?? (() => undefined), unmount: overrides.unmount ?? (async () => undefined), ...(overrides.wait ? { wait: overrides.wait } : {}), onError: overrides.onError ?? (() => undefined), }); } +function sharedSession(id = 'session-shared'): SharedSessionCatalogProjection { + return { + kind: 'shared_session', + id, + revision: 1, + createdAt: 1, + activityAt: 2, + name: 'Shared task', + status: 'active', + }; +} + function memoryStore(): GuestSessionMountStore { let mounts: readonly GuestSessionMount[] = []; return { @@ -469,6 +1075,35 @@ function memoryStore(): GuestSessionMountStore { }; } +function serializedStore(): GuestSessionMountStore { + let secret: string | null = null; + return createGuestSessionMountStore({ + getSecret: async () => secret, + setSecret: async (_slug, _kind, value) => { + secret = value; + }, + }); +} + +function settlePromptly(operation: Promise): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error('Operation did not settle after cancellation')), + 1_000, + ); + void operation.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (error: unknown) => { + clearTimeout(timer); + reject(error); + }, + ); + }); +} + function invitation(credential: string): string { return encodeDesktopCollaborationInvitation({ invitationCode: encodeCollaborationInvitationCode({ diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-ipc-main.test.ts index fcc68ecb55..9cb9e8b6ec 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-ipc-main.test.ts @@ -23,7 +23,6 @@ import type { IpcMain } from 'electron'; import type { SessionCatalogProjection, SessionCreateInput } from '@maka/runtime-host/protocol'; import { registerRuntimeHostSessionCatalogIpc, - toDesktopHostSharedSessionSummary, toDesktopHostSessionSummary, type RuntimeHostSessionCatalogIpcDeps, } from '../runtime-host-session-catalog-ipc-main.js'; @@ -42,9 +41,8 @@ test('maps Runtime Host live run state without collapsing unknown and known-empt assert.deepEqual(running.runningTurnIds, ['turn-live']); }); -test('preserves the Session revision in Owner and Shared Desktop Host summaries', () => { +test('preserves the Session revision in Owner Desktop Host summaries', () => { assert.equal(toDesktopHostSessionSummary(projection({ revision: 7 })).revision, 7); - assert.equal(toDesktopHostSharedSessionSummary(sharedProjection()).revision, 7); }); test('session creation forwards the caller name for a mode that carries none', async () => { @@ -127,15 +125,3 @@ function projection(overrides: Partial = {}): SessionC ...overrides, }; } - -function sharedProjection() { - return { - kind: 'shared_session' as const, - id: 'session-1', - revision: 7, - createdAt: 1, - activityAt: 2, - name: 'Session', - status: 'active' as const, - }; -} diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-preload.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-preload.test.ts index bed0ef61e3..059c4efcd6 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-preload.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-preload.test.ts @@ -22,6 +22,8 @@ import test from 'node:test'; import type { DesktopSessionSummary } from '../../preload/bridge-contract.js'; import { collectRuntimeHostSessionCatalogsWithCoverage, + createRuntimeHostSessionCatalogRefresher, + recordObservedRuntimeHostSessionAuthority, reconcileRuntimeHostSessionCatalog, resolveRuntimeHostSessionCatalog, } from '../../preload/runtime-host-session-catalog.js'; @@ -30,118 +32,333 @@ function session(id: string, activityAt: number): DesktopSessionSummary { return { id, activityAt } as DesktopSessionSummary; } -test('keeps healthy Host catalogs when another Host rejects', async () => { - const catalog = await collectRuntimeHostSessionCatalogsWithCoverage([ - { hostId: 'older', access: 'owner', sessions: Promise.resolve([session('older', 1)]) }, - { - hostId: 'unavailable', - access: 'owner', - sessions: Promise.reject(new Error('remote unavailable')), +function ownerSession( + id: string, + activityAt: number, + hostId = 'owner-host', + profileId = 'owner-profile', +): DesktopSessionSummary { + return { + ...session(id, activityAt), + runtimeHostId: hostId, + profileId, + }; +} + +function guestSession( + id: string, + activityAt: number, + profileId = 'guest-profile', +): DesktopSessionSummary { + return { + ...session(id, activityAt), + runtimeHostId: 'guest-host', + profileId, + shared: true, + }; +} + +test('starts a fresh catalog read after the previous refresh settles', async () => { + let resolveFirst!: (sessions: DesktopSessionSummary[]) => void; + const firstRead = new Promise((resolve) => { + resolveFirst = resolve; + }); + const stale = session('stale', 1); + const fresh = session('fresh', 2); + let reads = 0; + let current: DesktopSessionSummary[] = []; + const refresher = createRuntimeHostSessionCatalogRefresher({ + listCatalog: async () => ({ + sessions: await (++reads === 1 ? firstRead : Promise.resolve([fresh])), + completeHostIds: [], + }), + currentCatalog: () => ({ sessions: current, completeHostIds: [] }), + commitCatalog: (catalog) => { + current = catalog.sessions; }, - { hostId: 'newer', access: 'owner', sessions: Promise.resolve([session('newer', 2)]) }, - ]); + }); - assert.deepEqual(catalog.sessions.map(({ id }) => id), ['newer', 'older']); + const first = refresher.refresh(); + resolveFirst([stale]); + assert.deepEqual((await first).sessions, [stale]); + const second = refresher.refresh(); + assert.deepEqual((await second).sessions, [fresh]); + assert.equal(reads, 2); }); -test('reports exactly which Host catalogs are complete', async () => { - const catalog = await collectRuntimeHostSessionCatalogsWithCoverage([ - { hostId: 'local', access: 'owner', sessions: Promise.resolve([session('local-session', 1)]) }, - { - hostId: 'remote', - access: 'owner', - sessions: Promise.reject(new Error('remote unavailable')), +test('does not commit a catalog read superseded while it is in flight', async () => { + let resolveFirst!: (sessions: DesktopSessionSummary[]) => void; + const firstRead = new Promise((resolve) => { + resolveFirst = resolve; + }); + const stale = session('stale', 1); + const fresh = session('fresh', 2); + const commits: DesktopSessionSummary[][] = []; + let reads = 0; + let current: DesktopSessionSummary[] = []; + const refresher = createRuntimeHostSessionCatalogRefresher({ + listCatalog: async () => ({ + sessions: await (++reads === 1 ? firstRead : Promise.resolve([fresh])), + completeHostIds: [], + }), + currentCatalog: () => ({ sessions: current, completeHostIds: [] }), + commitCatalog: (catalog) => { + current = catalog.sessions; + commits.push(catalog.sessions); }, - ]); + }); - assert.deepEqual(catalog.sessions.map(({ id }) => id), ['local-session']); - assert.deepEqual(catalog.completeHostIds, ['local']); + const first = refresher.refresh(); + const second = refresher.refresh(); + resolveFirst([stale]); + + assert.deepEqual((await first).sessions, [fresh]); + assert.deepEqual((await second).sessions, [fresh]); + assert.deepEqual(commits, [[fresh]]); + assert.equal(reads, 2); }); -test('collapses overlapping Guest catalogs in favor of the Owner authority', async () => { - const owner = session('shared-session', 2); - const guest = { ...owner, shared: true as const }; +test('continues to an admitted trailing read after a superseded read fails', async () => { + let rejectFirst!: (error: Error) => void; + const firstRead = new Promise((_resolve, reject) => { + rejectFirst = reject; + }); + const fresh = session('fresh', 2); + let reads = 0; + let current: DesktopSessionSummary[] = []; + const refresher = createRuntimeHostSessionCatalogRefresher({ + listCatalog: async () => ({ + sessions: await (++reads === 1 ? firstRead : Promise.resolve([fresh])), + completeHostIds: [], + }), + currentCatalog: () => ({ sessions: current, completeHostIds: [] }), + commitCatalog: (catalog) => { + current = catalog.sessions; + }, + }); - const catalog = await collectRuntimeHostSessionCatalogsWithCoverage([ - { hostId: 'shared', access: 'session_guest', sessions: Promise.resolve([guest]) }, - { hostId: 'shared', access: 'owner', sessions: Promise.resolve([owner]) }, - ]); + const first = refresher.refresh(); + const second = refresher.refresh(); + rejectFirst(new Error('superseded')); - assert.deepEqual(catalog.sessions, [owner]); + assert.deepEqual((await first).sessions, [fresh]); + assert.deepEqual((await second).sessions, [fresh]); + assert.equal(reads, 2); }); -test('retains a Guest catalog row only while its Runtime Host profile remains known', () => { - const shared = { - ...session('shared-session', 2), - runtimeHostId: 'host-guest', - profileId: 'guest-profile', - shared: true as const, +test('keeps a newly created Session when the superseding catalog read fails', async () => { + let resolveStale!: (catalog: { + sessions: DesktopSessionSummary[]; + completeHostIds: string[]; + }) => void; + const staleRead = new Promise<{ + sessions: DesktopSessionSummary[]; + completeHostIds: string[]; + }>((resolve) => { + resolveStale = resolve; + }); + const created = ownerSession('created', 2); + let reads = 0; + let current = { + sessions: [ownerSession('existing', 1)], + completeHostIds: ['owner-host'], }; + const refresher = createRuntimeHostSessionCatalogRefresher({ + listCatalog: () => { + reads += 1; + return reads === 1 + ? staleRead + : Promise.reject(new Error('Owner catalog temporarily unavailable')); + }, + currentCatalog: () => current, + commitCatalog: (catalog) => { + current = catalog; + }, + }); - const reconnecting = reconcileRuntimeHostSessionCatalog([shared], { - sessions: [], - completeHostIds: [], - knownProfileIds: ['guest-profile'], + const refresh = refresher.refresh(); + refresher.admit(created); + resolveStale({ sessions: [], completeHostIds: ['owner-host'] }); + + await assert.rejects(refresh, /temporarily unavailable/); + assert.deepEqual(current.sessions.map(({ id }) => id), ['created', 'existing']); + assert.equal(reads, 2); +}); + +test('rejects a delayed bootstrap seed after a Session is created', async () => { + const created = ownerSession('created', 2); + let current = { + sessions: [ownerSession('existing', 1)], + completeHostIds: ['owner-host'], + }; + const refresher = createRuntimeHostSessionCatalogRefresher({ + listCatalog: () => Promise.reject(new Error('Owner catalog temporarily unavailable')), + currentCatalog: () => current, + commitCatalog: (catalog) => { + current = catalog; + }, }); - assert.deepEqual(reconnecting, [shared]); + + const delayedBootstrap = refresher.beginSeed(); + refresher.admit(created); + + assert.equal( + delayedBootstrap.commit({ sessions: [], completeHostIds: ['owner-host'] }), + false, + ); + await assert.rejects(refresher.refresh(), /temporarily unavailable/); + assert.deepEqual(current.sessions.map(({ id }) => id), ['created', 'existing']); +}); + +test('collects every healthy Owner catalog and reports only complete Hosts', async () => { + const catalog = await collectRuntimeHostSessionCatalogsWithCoverage([ + { hostId: 'older', sessions: Promise.resolve([session('older', 1)]) }, + { + hostId: 'unavailable', + sessions: Promise.reject(new Error('remote unavailable')), + }, + { hostId: 'newer', sessions: Promise.resolve([session('newer', 2)]) }, + ]); + + assert.deepEqual(catalog.sessions.map(({ id }) => id), ['newer', 'older']); + assert.deepEqual(catalog.completeHostIds, ['older', 'newer']); +}); + +test('keeps incomplete coverage when every Owner catalog rejects', async () => { + assert.deepEqual( + await collectRuntimeHostSessionCatalogsWithCoverage([ + { hostId: 'first', sessions: Promise.reject(new Error('first unavailable')) }, + { hostId: 'second', sessions: Promise.reject(new Error('second unavailable')) }, + ]), + { sessions: [], completeHostIds: [] }, + ); +}); + +test('reconciles Owner catalogs per Host and retires removed Owner profiles', () => { + const first = ownerSession('first', 1, 'first-host', 'first-owner'); + const second = ownerSession('second', 2, 'second-host', 'second-owner'); assert.deepEqual( - reconcileRuntimeHostSessionCatalog(reconnecting, { - sessions: [{ ...shared, activityAt: 3 }], - completeHostIds: ['host-guest'], - knownProfileIds: ['guest-profile'], + reconcileRuntimeHostSessionCatalog([first, second], { + sessions: [], + completeHostIds: ['first-host'], + knownOwnerProfileIds: ['first-owner', 'second-owner'], + guestSessions: [], + }), + [second], + ); + assert.deepEqual( + reconcileRuntimeHostSessionCatalog([first, second], { + sessions: [], + completeHostIds: [], + knownOwnerProfileIds: ['first-owner'], + guestSessions: [], }), - [{ ...shared, activityAt: 3 }], + [first], ); +}); + +test('uses the mount service as the complete Guest catalog authority', () => { + const stale = guestSession('shared', 1); + const fresh = { ...stale, activityAt: 2, name: 'fresh' }; + const owner = ownerSession('shared', 3, 'guest-host'); assert.deepEqual( - reconcileRuntimeHostSessionCatalog(reconnecting, { + reconcileRuntimeHostSessionCatalog([stale], { sessions: [], completeHostIds: [], - knownProfileIds: [], + knownOwnerProfileIds: [], + guestSessions: [fresh], + }), + [fresh], + ); + assert.deepEqual( + reconcileRuntimeHostSessionCatalog([stale], { + sessions: [], + completeHostIds: [], + knownOwnerProfileIds: [], + guestSessions: [], }), [], ); + assert.deepEqual( + reconcileRuntimeHostSessionCatalog([], { + sessions: [owner], + completeHostIds: ['guest-host'], + knownOwnerProfileIds: ['owner-profile'], + guestSessions: [fresh], + }), + [owner], + ); }); -test('keeps healthy Owner catalogs when Guest mount inventory is unavailable', async () => { - const owner = { - ...session('owner-session', 3), - runtimeHostId: 'owner-host', - profileId: 'owner-profile', - }; - const shared = { - ...session('shared-session', 2), - runtimeHostId: 'guest-host', - profileId: 'guest-profile', - shared: true as const, +test('prefers a live Guest projection over an unavailable Owner fallback', () => { + const retainedOwner = ownerSession('shared', 1); + const liveGuest = { + ...guestSession('shared', 2), + runtimeHostId: retainedOwner.runtimeHostId, }; + assert.deepEqual( + reconcileRuntimeHostSessionCatalog([retainedOwner], { + sessions: [], + completeHostIds: [], + knownOwnerProfileIds: [retainedOwner.profileId], + guestSessions: [liveGuest], + }), + [liveGuest], + ); +}); + +test('retains the last Guest projection only when mount inventory is unavailable', async () => { + const guest = guestSession('shared', 2); + assert.deepEqual( + await resolveRuntimeHostSessionCatalog( + [guest], + Promise.resolve({ sessions: [], completeHostIds: [] }), + () => [], + Promise.reject(new Error('mount inventory unavailable')), + ), + { sessions: [guest], completeHostIds: [] }, + ); +}); + +test('preserves an authenticated onboarding Owner seed across a failed catalog refresh', async () => { + const owner = ownerSession('owner', 3); + const guest = guestSession('shared', 2); + const seeded = reconcileRuntimeHostSessionCatalog([], { + sessions: [owner], + completeHostIds: ['owner-host'], + knownOwnerProfileIds: ['owner-profile'], + }); + assert.deepEqual( await resolveRuntimeHostSessionCatalog( - [shared], - Promise.resolve({ sessions: [owner], completeHostIds: ['owner-host'] }), + seeded, + collectRuntimeHostSessionCatalogsWithCoverage([ + { + hostId: 'owner-host', + sessions: Promise.reject(new Error('catalog temporarily unavailable')), + }, + ]), () => ['owner-profile'], - Promise.reject(new Error('Guest mount store is unreadable')), + Promise.resolve([guest]), ), - [owner, shared], + { sessions: [owner, guest], completeHostIds: [] }, ); }); -test('fails when every Host catalog rejects', async () => { - await assert.rejects( - collectRuntimeHostSessionCatalogsWithCoverage([ - { - hostId: 'first', - access: 'owner', - sessions: Promise.reject(new Error('first unavailable')), - }, - { - hostId: 'second', - access: 'owner', - sessions: Promise.reject(new Error('second unavailable')), - }, - ]), - /Every Runtime Host Session Catalog request failed/, +test('an observed Guest event cannot replace an accepted Owner authority', () => { + const authorities = new Map([['shared-session', 'owner-profile']]); + + assert.equal( + recordObservedRuntimeHostSessionAuthority(authorities, 'shared-session', 'guest-profile'), + false, + ); + assert.equal(authorities.get('shared-session'), 'owner-profile'); + assert.equal( + recordObservedRuntimeHostSessionAuthority(authorities, 'new-session', 'guest-profile'), + true, ); + assert.equal(authorities.get('new-session'), 'guest-profile'); }); diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts index 957a930072..d2c3400c29 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts @@ -21,28 +21,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import type { SessionCatalogProjection } from '@maka/runtime-host/protocol'; import type { IpcHandler } from '../ipc-reconnect-policy.js'; -import { - registerRuntimeHostSessionCatalogIpc, - registerRuntimeHostSharedSessionCatalogIpc, -} from '../runtime-host-session-catalog-ipc-main.js'; - -test('registers a read-only Session catalog for shared access', async () => { - const handlers = new Map(); - registerRuntimeHostSharedSessionCatalogIpc( - { getSession: async () => ({ id: 'shared' }) as never }, - { - handle: (channel, listener) => handlers.set(channel, listener), - handleReconnectableRead: (channel, listener) => handlers.set(channel, listener), - }, - ); - - assert.deepEqual([...handlers.keys()], ['sessions:list']); - assert.deepEqual(await handlers.get('sessions:list')!({} as never), [{ id: 'shared' }]); - assert.deepEqual( - await handlers.get('sessions:list')!({} as never, { subagentParentSessionId: 'parent' }), - [], - ); -}); +import { registerRuntimeHostSessionCatalogIpc } from '../runtime-host-session-catalog-ipc-main.js'; test('projects observed running Turn identities into renderer Session lists', async () => { const handlers = new Map(); diff --git a/apps/desktop/src/main/__tests__/session-collaboration-join-dialog.test.ts b/apps/desktop/src/main/__tests__/session-collaboration-join-dialog.test.ts index d66c187749..95afd6e693 100644 --- a/apps/desktop/src/main/__tests__/session-collaboration-join-dialog.test.ts +++ b/apps/desktop/src/main/__tests__/session-collaboration-join-dialog.test.ts @@ -67,6 +67,7 @@ test('keeps loading progress visible while an irreversible import settles', asyn cancelImport: async () => 'settling', readInvitationClipboard: async () => '', listMounts: async () => [], + subscribeMountChanges: () => () => undefined, removeMount: async () => undefined, requestTurn: async () => { throw new Error('unused'); @@ -124,6 +125,7 @@ test('closes as a retained background recovery instead of reporting a failed joi cancelImport: async () => 'settling', readInvitationClipboard: async () => '', listMounts: async () => [], + subscribeMountChanges: () => () => undefined, removeMount: async () => undefined, requestTurn: async () => { throw new Error('unused'); @@ -173,6 +175,73 @@ test('closes as a retained background recovery instead of reporting a failed joi assert.doesNotMatch(document.body.textContent, /connectionFailed/u); }); +test('identifies a retained shared task and its selected peer transport', async () => { + const services: SessionCollaborationServices = { + importInvitation: async () => { + throw new Error('unused'); + }, + cancelImport: async () => 'settling', + readInvitationClipboard: async () => '', + listMounts: async () => [{ + mountId: 'shared-1', + name: 'Shared Host', + hostId: 'a'.repeat(64), + readiness: 'ready', + peerPath: { kind: 'direct', transport: 'webrtc' }, + session: { + kind: 'shared_session', + id: 'session-1', + revision: 1, + createdAt: 1, + activityAt: 2, + name: 'Shared task', + status: 'active', + }, + }], + subscribeMountChanges: () => () => undefined, + removeMount: async () => undefined, + requestTurn: async () => { + throw new Error('unused'); + }, + getTurnRequests: async () => ({ canRequestTurns: false, requests: [] }), + acknowledgeTurnRequest: async () => ({ acknowledged: false }), + withdrawTurnRequest: async () => ({ withdrawn: false }), + getPendingTurnRequests: async () => [], + decideTurnRequest: async () => { + throw new Error('unused'); + }, + createOperationId: () => 'operation-1', + }; + const { document } = installDom(); + const container = document.querySelector('#root'); + assert.ok(container); + mountedRoot = createRoot(container); + await act(async () => { + mountedRoot?.render( + createElement(LocaleProvider, { + locale: 'en', + children: createElement(AstryxLocaleProvider, { + children: createElement(ToastProvider, { + children: createElement(SessionCollaborationServicesProvider, { + services, + children: createElement(SessionCollaborationJoinDialog, { + copy: testCopy(), + onImported: assert.fail, + onClose: assert.fail, + }), + }), + }), + }), + }), + ); + await Promise.resolve(); + }); + + assert.match(document.body.textContent, /Shared task/u); + assert.match(document.body.textContent, /Shared Host · mountConnected/u); + assert.match(document.body.textContent, /WebRTC/u); +}); + function installDom(): { document: Document } { const parsed = parseHTML('
'); const { document, window } = parsed; diff --git a/apps/desktop/src/main/__tests__/session-read-state.test.ts b/apps/desktop/src/main/__tests__/session-read-state.test.ts index 492abafb50..36927e339d 100644 --- a/apps/desktop/src/main/__tests__/session-read-state.test.ts +++ b/apps/desktop/src/main/__tests__/session-read-state.test.ts @@ -107,6 +107,38 @@ describe('renderer session read state', () => { assert.equal(committedContext, 'before'); }); + + it('does not lose a refresh admitted while the previous task is settling', async () => { + const firstList = deferred(); + let listCalls = 0; + let currentSessions: SessionSummary[] = []; + const current = session({ id: 'current', lastMessageAt: 2 }); + const refresher = createSessionListRefresher({ + captureRequestContext: () => undefined, + listSessions: () => { + listCalls += 1; + return listCalls === 1 ? firstList.promise : Promise.resolve([current]); + }, + currentSessions: () => currentSessions, + commitSessions: (next) => { + currentSessions = next; + }, + onError: () => {}, + }); + + const firstRefresh = refresher.refresh(); + firstList.resolve([session({ id: 'stale', lastMessageAt: 1 })]); + let settlementRefresh: Promise | undefined; + queueMicrotask(() => { + settlementRefresh = refresher.refresh(); + }); + + assert.deepEqual((await firstRefresh).map(({ id }) => id), ['stale']); + await Promise.resolve(); + assert.ok(settlementRefresh); + assert.deepEqual((await settlementRefresh).map(({ id }) => id), ['current']); + assert.equal(listCalls, 2); + }); }); function session(overrides: Partial & { id: string }): SessionSummary { return { diff --git a/apps/desktop/src/main/__tests__/session-settings-controller.test.ts b/apps/desktop/src/main/__tests__/session-settings-controller.test.ts index 7c81e38720..bb9f477fe7 100644 --- a/apps/desktop/src/main/__tests__/session-settings-controller.test.ts +++ b/apps/desktop/src/main/__tests__/session-settings-controller.test.ts @@ -265,7 +265,7 @@ test('retains the Model overlay while a partial Host catalog still has the prior { sessions: [{ ...otherHostSession, revision: 2, activityAt: 20 }], completeHostIds: ['host-b'], - knownProfileIds: ['profile-a', 'profile-b'], + knownOwnerProfileIds: ['profile-a', 'profile-b'], }, ); assert.equal(partialCatalog.find((session) => session.id === 'session-a')?.revision, 1); @@ -279,7 +279,7 @@ test('retains the Model overlay while a partial Host catalog still has the prior { ...targetAfterWrite, activityAt: 21 }, ], completeHostIds: ['host-a', 'host-b'], - knownProfileIds: ['profile-a', 'profile-b'], + knownOwnerProfileIds: ['profile-a', 'profile-b'], }); assert.equal(caughtUpCatalog.find((session) => session.id === 'session-a')?.revision, 2); await render(2, caughtUpCatalog); @@ -365,7 +365,7 @@ test('retires Permission and Orchestration overlays by their committed Session r { sessions: [{ ...otherHostSession, revision: 2, activityAt: 20 }], completeHostIds: ['host-b'], - knownProfileIds: ['profile-a', 'profile-b'], + knownOwnerProfileIds: ['profile-a', 'profile-b'], }, ); await render(1, partialCatalog); diff --git a/apps/desktop/src/main/__tests__/session-turn-request-composer.test.ts b/apps/desktop/src/main/__tests__/session-turn-request-composer.test.ts index 68bf76d55d..b86f184310 100644 --- a/apps/desktop/src/main/__tests__/session-turn-request-composer.test.ts +++ b/apps/desktop/src/main/__tests__/session-turn-request-composer.test.ts @@ -83,6 +83,7 @@ test('keeps a newer Guest draft across remount when an old request settles later cancelImport: async () => 'cancelled', readInvitationClipboard: async () => '', listMounts: async () => [], + subscribeMountChanges: () => () => undefined, removeMount: async () => undefined, requestTurn: async () => { throw new Error('connection lost after dispatch'); @@ -174,6 +175,7 @@ test('resumes an in-flight Guest request across remount without submitting it tw cancelImport: async () => 'cancelled', readInvitationClipboard: async () => '', listMounts: async () => [], + subscribeMountChanges: () => () => undefined, removeMount: async () => undefined, requestTurn: async () => { requestCount += 1; diff --git a/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts b/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts index 5a5a317b1f..d60c58c767 100644 --- a/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts +++ b/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts @@ -21,9 +21,7 @@ import { strict as assert } from 'node:assert'; import { describe, it } from 'node:test'; import type { OnboardingState } from '@maka/core/onboarding'; import { - advanceOnboardingSnapshotState, createOnboardingSnapshotPoller, - createOnboardingSnapshotState, getOnboardingActivationCandidate, } from '../../renderer/use-onboarding-snapshot.js'; import type { OnboardingSnapshot } from '../../preload/bridge-contract.js'; @@ -252,44 +250,3 @@ describe('createOnboardingSnapshotPoller', () => { assert.deepEqual(events, [{ type: 'snap', payload: READY_SNAPSHOT }]); }); }); - -describe('onboarding mounted snapshot handoff', () => { - it('keeps a session created while mounted snapshots wait for React to commit', () => { - const snapshotA = READY_SNAPSHOT; - const snapshotB = { ...READY_SNAPSHOT }; - const snapshotC: OnboardingSnapshot = { - ...READY_SNAPSHOT, - sessions: [ - { - runtimeHostId: 'host-1', - profileId: 'local', - profileName: 'Local', - profileKind: 'local', - revision: 1, - id: 'created-during-bootstrap', - name: 'New session', - isFlagged: false, - isArchived: false, - labels: [], - hasUnread: false, - status: 'active' as const, - backend: 'fake', - llmConnectionSlug: 'default', - connectionLocked: false, - model: 'fake-model', - permissionMode: 'ask' as const, - projectId: null, - }, - ], - }; - - const afterB = advanceOnboardingSnapshotState(createOnboardingSnapshotState(snapshotA), snapshotB); - const afterC = advanceOnboardingSnapshotState(afterB, snapshotC); - - assert.equal(afterC.snapshot, snapshotC); - assert.deepEqual( - afterC.mountedSnapshotHandoff?.sessions.map(({ id }) => id), - ['created-during-bootstrap'], - ); - }); -}); diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index 69c98b496c..609bb909ac 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -601,15 +601,25 @@ const runtimeHostProfileService = createDesktopRuntimeHostProfileService({ runtimeHostManager.setDefaultProfile(profileId); }, }); +const notifyGuestSessionMountsChanged = (): void => { + mainWindowController.send('session-collaboration:mounts:changed'); +}; const guestSessionMountService = createDesktopGuestSessionMountService({ store: createGuestSessionMountStore(runtimeHostCredentialStore), - mount: async (target, signal, onConnectionPhase, onPeerEndpoint) => { + mount: async ( + target, + signal, + onConnectionPhase, + onPeerEndpoint, + onSessionCatalogChanged, + ) => { if (target.profile.kind !== 'remote' || !target.credential) { throw new Error('A shared Session requires a remote Guest target'); } if (!runtimeHostManager) throw new Error('Runtime Host manager is unavailable'); await runtimeHostManager.mountGuest( { profile: target.profile, credential: target.credential }, + onSessionCatalogChanged, signal, onConnectionPhase, (status) => { @@ -621,6 +631,26 @@ const guestSessionMountService = createDesktopGuestSessionMountService({ if (!runtimeHostManager) throw new Error('Runtime Host manager is unavailable'); return runtimeHostManager.finalizeGuestAccess(mountId, signal, onAccessActivated); }, + getSharedSession: async (mountId) => { + const current = runtimeHostManager?.current(mountId); + if (!current?.candidate) { + throw new Error('Shared Session Runtime Host is reconnecting'); + } + return current.candidate.client.getSharedSession(); + }, + inspect: (mountId) => { + const state = runtimeHostManager?.entries().find( + (candidate) => candidate.target.profile.id === mountId, + ); + if (!state) return undefined; + return { + readiness: state.readiness, + ...(state.readiness === 'ready' && state.candidate.client.peerPath + ? { peerPath: state.candidate.client.peerPath } + : {}), + }; + }, + onMountsChanged: notifyGuestSessionMountsChanged, unmount: async (mountId) => { if (!runtimeHostManager) return; await runtimeHostManager.unmountGuest(mountId); @@ -1095,6 +1125,7 @@ const startLocalRuntimeHostManager = () => startRuntimeHostDesktopManager( showStartupDiagnosticDialog, ), onTargetStateChanged: (state) => { + const profileAccess = runtimeHostProfileAccess(state.target.profile); const hostId = state.readiness === "ready" ? state.candidate.client.hostId : state.hostId; @@ -1103,13 +1134,23 @@ const startLocalRuntimeHostManager = () => startRuntimeHostDesktopManager( profileId: state.target.profile.id, profileName: state.target.profile.name, profileKind: state.target.profile.kind, - profileAccess: runtimeHostProfileAccess(state.target.profile), + profileAccess, ...(hostId ? { hostId } : {}), readiness: state.readiness, isDefault: (runtimeHostManager?.defaultProfileId() ?? runtimeHostStartup.preferences.defaultProfileId) === state.target.profile.id, }); + if (profileAccess === 'session_guest') { + void guestSessionMountService + .connectionChanged( + state.target.profile.id, + state.readiness === 'unavailable' ? state.error : undefined, + ) + .catch((error: unknown) => + console.warn('[runtime-host] shared Session connection update failed:', error), + ); + } if (state.readiness === "unavailable" && state.hostId) { void browserIpc.retireTarget({ hostId: state.hostId, diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index 22e0ddfa9b..f622008f5a 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -75,9 +75,7 @@ import { type DesktopNativeCapabilityProviderInput, } from "./runtime-host-native-capabilities.js"; import { - registerRuntimeHostSharedSessionCatalogIpc, registerRuntimeHostSessionCatalogIpc, - toDesktopHostSharedSessionSummary, } from "./runtime-host-session-catalog-ipc-main.js"; import { registerRuntimeHostWorkHubIpc } from "./runtime-host-workhub-ipc-main.js"; import { registerRuntimeHostExternalSessionsIpc } from "./runtime-host-external-sessions-ipc-main.js"; @@ -147,6 +145,7 @@ export interface DesktopRuntimeHostCandidateDeps { readonly onError?: RuntimeHostSessionDomainsIpcDeps["onError"]; readonly isTargetActive?: () => boolean; readonly isTargetValid?: () => boolean; + readonly onGuestSessionCatalogChanged?: () => void; readonly newId?: () => string; readonly now?: () => number; readonly openSshTunnel?: ( @@ -593,6 +592,12 @@ export async function createDesktopRuntimeHostCandidate( let observationsAttached = false; let capabilitiesRegistered = false; try { + const onGuestSessionCatalogChanged = target.access === 'session_guest' + ? deps.onGuestSessionCatalogChanged + : undefined; + if (target.access === 'session_guest' && !onGuestSessionCatalogChanged) { + throw new Error('A Session Guest candidate requires a catalog-change authority'); + } let domains: RuntimeHostSessionDomainsIpcHandle | undefined; const emitActiveInteractionsChanged = ( sessionId: string, @@ -801,23 +806,12 @@ export async function createDesktopRuntimeHostCandidate( ) : undefined; disposeClientIpc = target.access === 'session_guest' - ? client.subscribeSessionCatalogChanges(({ sessionId }) => - emitSessionsChanged('updated', sessionId), - ) + ? client.subscribeSessionCatalogChanges(() => onGuestSessionCatalogChanged!()) : typeof registeredClientIpc === 'function' ? registeredClientIpc : undefined; if (target.access === 'session_guest') { registerRuntimeHostAttachmentPreviewIpc({ ipcMain: ipc, client }); - registerRuntimeHostSharedSessionCatalogIpc( - { - getSession: async () => { - const session = await client.getSharedSession(); - return session ? toDesktopHostSharedSessionSummary(session) : null; - }, - }, - ipc, - ); } else { if (!sessionCopyCleanup) throw new Error('Owner Session copy authority is unavailable'); registerRuntimeHostSessionCatalogIpc( diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts index 150e760454..b514afaec1 100644 --- a/apps/desktop/src/main/runtime-host-desktop-manager.ts +++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts @@ -75,6 +75,7 @@ export interface RuntimeHostDesktopManager { ): Promise; mountGuest( profileTarget: NonNullable, + onSessionCatalogChanged: () => void, signal?: AbortSignal, onConnectionPhase?: (phase: RuntimeHostConnectionPhase) => void, onHostStatus?: (status: HostStatusResult) => void, @@ -547,6 +548,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { mountGuest( profileTarget: NonNullable, + onSessionCatalogChanged: () => void, signal?: AbortSignal, onConnectionPhase?: (phase: RuntimeHostConnectionPhase) => void, onHostStatus?: (status: HostStatusResult) => void, @@ -555,7 +557,14 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { return Promise.reject(new Error('A Session Guest target is required')); } return this.#mutateTarget(profileTarget.profile.id, () => - this.#enable(profileTarget, true, signal, onConnectionPhase, onHostStatus), + this.#enable( + profileTarget, + true, + signal, + onConnectionPhase, + onHostStatus, + onSessionCatalogChanged, + ), ); } @@ -565,6 +574,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { signal?: AbortSignal, onConnectionPhase?: (phase: RuntimeHostConnectionPhase) => void, onHostStatus?: (status: HostStatusResult) => void, + onGuestSessionCatalogChanged?: () => void, ): Promise { signal?.throwIfAborted(); if (this.#closed) throw new Error('Desktop Runtime Host manager is closed'); @@ -596,6 +606,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { ...withRuntimeHostTarget(this.#baseInput, profileTarget), ...(onConnectionPhase ? { onConnectionPhase } : {}), ...(onHostStatus ? { onHostStatus } : {}), + ...(onGuestSessionCatalogChanged ? { onGuestSessionCatalogChanged } : {}), }); this.#targets.set(profileId, target); this.#publishState(target, { diff --git a/apps/desktop/src/main/runtime-host-guest-session-mounts.ts b/apps/desktop/src/main/runtime-host-guest-session-mounts.ts index 39b231d080..a592272943 100644 --- a/apps/desktop/src/main/runtime-host-guest-session-mounts.ts +++ b/apps/desktop/src/main/runtime-host-guest-session-mounts.ts @@ -19,16 +19,21 @@ import { randomUUID } from 'node:crypto'; import { + abortable, decodeRemoteRuntimeHostProfile, RUNTIME_HOST_ACCESS_CREDENTIAL_MAX_BYTES, RuntimeHostPermanentReconnectError, + RuntimeHostProfileConnectionError, type ResolvedRuntimeHostProfile, type RuntimeHostConnectionPhase, + type RuntimeHostPeerConnectionPath, type RuntimeHostRemoteTransport, } from '@maka/runtime-host/client'; import { decodeCollaborationInvitationCode, + decodeSharedSessionCatalogProjection, type HostPeerEndpoint, + type SharedSessionCatalogProjection, } from '@maka/runtime-host/protocol'; import type { CredentialStore } from '@maka/storage/credential-store'; import type { @@ -50,6 +55,7 @@ const STORE_SCHEMA_VERSION = 1; const STORE_SLOT = 'desktop-guest-session-mounts'; const MAX_MOUNTS = 128; const STARTUP_RETRY_MAX_MS = 30_000; +const DEFERRED_WRITE_RETRY_MAX_MS = 30_000; export interface GuestSessionMount { readonly mountId: string; @@ -57,8 +63,11 @@ export interface GuestSessionMount { readonly rootId: string; readonly transport: RuntimeHostRemoteTransport; readonly credential: string; + readonly session?: SharedSessionCatalogProjection; } +type GuestSessionMountReadiness = SessionCollaborationMountSummary['readiness']; + interface GuestSessionMountDocument { readonly schemaVersion: typeof STORE_SCHEMA_VERSION; readonly mounts: readonly GuestSessionMount[]; @@ -66,7 +75,8 @@ interface GuestSessionMountDocument { interface LiveGuestActivationBase { readonly controller: AbortController; - stage: 'connecting' | 'finalizing'; + stage: 'connecting' | 'finalizing' | 'hydrating'; + accessActivated: boolean; finalization?: Promise; task: Promise; } @@ -85,6 +95,16 @@ interface LiveGuestStartupActivation extends LiveGuestActivationBase { type LiveGuestActivation = LiveGuestImportActivation | LiveGuestStartupActivation; +interface LiveGuestRefresh { + dirty: boolean; + task: Promise; +} + +interface DeferredWriteFailure { + readonly mount: GuestSessionMount; + readonly error: Error; +} + export interface GuestSessionMountStore { read(): Promise; write(mounts: readonly GuestSessionMount[]): Promise; @@ -93,6 +113,7 @@ export interface GuestSessionMountStore { export interface DesktopGuestSessionMountService { start(): Promise; list(): Promise; + connectionChanged(mountId: string, error?: Error): Promise; importInvitation( code: string, allowInsecure: boolean, @@ -119,7 +140,11 @@ export function createGuestSessionMountStore( } const document: GuestSessionMountDocument = { schemaVersion: STORE_SCHEMA_VERSION, - mounts: [...mounts].sort((left, right) => left.mountId.localeCompare(right.mountId)), + mounts: mounts + .map((mount) => + mount.session ? { ...mount, session: retainedSession(mount.session) } : mount, + ) + .sort((left, right) => left.mountId.localeCompare(right.mountId)), }; decodeDocument(document); await credentials.setSecret( @@ -136,28 +161,64 @@ export function createDesktopGuestSessionMountService(input: { readonly mount: ( target: ResolvedRuntimeHostProfile, signal: AbortSignal, - onConnectionPhase?: (phase: RuntimeHostConnectionPhase) => void, - onPeerEndpoint?: (endpoint: HostPeerEndpoint) => void, + onConnectionPhase: ((phase: RuntimeHostConnectionPhase) => void) | undefined, + onPeerEndpoint: ((endpoint: HostPeerEndpoint) => void) | undefined, + onSessionCatalogChanged: () => void, ) => Promise; readonly finalizeAccess: ( mountId: string, signal: AbortSignal, onAccessActivated?: () => void, ) => Promise; + readonly getSharedSession: (mountId: string) => Promise; + readonly inspect: (mountId: string) => + | { + readonly readiness: GuestSessionMountReadiness; + readonly peerPath?: RuntimeHostPeerConnectionPath; + } + | undefined; + readonly onMountsChanged: () => void; readonly unmount: (mountId: string) => Promise; readonly wait?: (delayMs: number, signal: AbortSignal) => Promise; readonly onError?: (error: Error, mount: GuestSessionMount) => void; }): DesktopGuestSessionMountService { const wait = input.wait ?? waitForDelay; - const onError = input.onError ?? ((error, mount) => { - console.warn(`[runtime-host] shared Session ${mount.mountId} is unavailable:`, error); - }); + const onError = + input.onError ?? + ((error, mount) => { + console.warn(`[runtime-host] shared Session ${mount.mountId} is unavailable:`, error); + }); const activations = new Set(); const removingMounts = new Set(); + const invalidatedAccessMounts = new Set(); + const refreshes = new Map(); let mounts: Map | undefined; let mutationTail = Promise.resolve(); + let deferredWriteFailure: DeferredWriteFailure | undefined; + let deferredWriteRetry: Promise | undefined; + const deferredWriteRetryController = new AbortController(); + const projectionLifetime = new AbortController(); let closed = false; + const readSharedSession = ( + mountId: string, + signal?: AbortSignal, + ): Promise => + abortable( + () => input.getSharedSession(mountId), + signal + ? AbortSignal.any([signal, projectionLifetime.signal]) + : projectionLifetime.signal, + ); + + const notifyMountsChanged = (): void => { + try { + input.onMountsChanged(); + } catch { + // Renderer invalidation cannot control the durable mount lifecycle. + } + }; + const mutate = (operation: () => Promise): Promise => { const pending = mutationTail.then(operation); mutationTail = pending.then( @@ -168,13 +229,90 @@ export function createDesktopGuestSessionMountService(input: { }; const load = async (): Promise> => { - if (!mounts) mounts = new Map((await input.store.read()).map((mount) => [mount.mountId, mount])); + if (!mounts) + mounts = new Map((await input.store.read()).map((mount) => [mount.mountId, mount])); return mounts; }; + const persistDeferredState = async ( + current: Map, + changedMount?: GuestSessionMount, + ): Promise => { + const failureMount = changedMount ?? deferredWriteFailure?.mount; + if (!failureMount) return undefined; + try { + await input.store.write([...current.values()]); + deferredWriteFailure = undefined; + return undefined; + } catch (error) { + const failure = asError(error); + deferredWriteFailure = { mount: failureMount, error: failure }; + return failure; + } + }; + + const flushDeferredWrite = (): Promise => + mutate(async () => persistDeferredState(await load())); + + const scheduleDeferredWriteRetry = (): void => { + if (closed || deferredWriteRetry) return; + deferredWriteRetry = (async () => { + let delayMs = 1_000; + while (!closed && deferredWriteFailure) { + try { + await wait(delayMs, deferredWriteRetryController.signal); + } catch { + return; + } + if (!(await flushDeferredWrite())) return; + delayMs = Math.min(delayMs * 2, DEFERRED_WRITE_RETRY_MAX_MS); + } + })().finally(() => { + deferredWriteRetry = undefined; + // Do not lose a failure installed while the prior task settled. + if (!closed && deferredWriteFailure) scheduleDeferredWriteRetry(); + }); + }; + const persist = async (next: Map): Promise => { await input.store.write([...next.values()]); mounts = next; + deferredWriteFailure = undefined; + }; + + const clearSessionProjection = async (mountId: string): Promise => { + // This fence is installed before the durable mutation is queued so a + // concurrent refresh cannot restore a projection after authority loss. + invalidatedAccessMounts.add(mountId); + await mutate(async () => { + const current = await load(); + const mount = current.get(mountId); + if (!mount) { + invalidatedAccessMounts.delete(mountId); + return; + } + let next = current; + if (mount.session) { + next = new Map(current).set(mountId, { + mountId: mount.mountId, + name: mount.name, + rootId: mount.rootId, + transport: mount.transport, + credential: mount.credential, + }); + // Authority loss takes effect in memory before a fallible credential + // store write. A locked or unavailable store must never keep exposing + // a projection whose credential has already been rejected. + mounts = next; + } + notifyMountsChanged(); + if (!mount.session && !deferredWriteFailure) return; + const error = await persistDeferredState(next, mount.session ? mount : undefined); + if (error) { + scheduleDeferredWriteRetry(); + throw error; + } + }); }; const recordPeerEndpoint = (mount: GuestSessionMount, endpoint: HostPeerEndpoint): void => { @@ -182,7 +320,8 @@ export function createDesktopGuestSessionMountService(input: { closed || mount.transport.kind !== 'libp2p-direct' || endpoint.lease.peerId !== mount.transport.reachability.lease.peerId - ) return; + ) + return; void mutate(async () => { if (removingMounts.has(mount.mountId)) return; const current = await load(); @@ -191,31 +330,152 @@ export function createDesktopGuestSessionMountService(input: { retained?.transport.kind !== 'libp2p-direct' || retained.transport.reachability.lease.peerId !== endpoint.lease.peerId || retained.transport.reachability.lease.revision >= endpoint.lease.revision - ) return; - const updated = decodeMount({ + ) + return; + const updated: GuestSessionMount = { ...retained, transport: { kind: 'libp2p-direct', reachability: endpoint, }, - }); - await persist(new Map(current).set(mount.mountId, updated)); + }; + const next = new Map(current).set(mount.mountId, updated); + // The authenticated endpoint is immediately useful to reconnect logic; + // a transient credential-store lock must not discard it. + mounts = next; + const error = await persistDeferredState(next, mount); + if (error) { + scheduleDeferredWriteRetry(); + throw error; + } }).catch((error: unknown) => onError(asError(error), mount)); }; + const recordSharedSession = async ( + mount: GuestSessionMount, + session: SharedSessionCatalogProjection, + ): Promise => { + if (invalidatedAccessMounts.has(mount.mountId)) { + throw new RuntimeHostPermanentReconnectError('Shared Session access is no longer available'); + } + const superseded = await mutate(async () => { + if (invalidatedAccessMounts.has(mount.mountId)) { + throw new RuntimeHostPermanentReconnectError( + 'Shared Session access is no longer available', + ); + } + const current = await load(); + const retained = current.get(mount.mountId); + if (!retained) throw new Error('Shared Session mount was removed while connecting'); + const next = new Map(current); + const duplicates: GuestSessionMount[] = []; + for (const candidate of current.values()) { + if ( + candidate.mountId === mount.mountId || + candidate.rootId !== mount.rootId || + candidate.session?.id !== session.id + ) + continue; + duplicates.push(candidate); + next.delete(candidate.mountId); + } + next.set(mount.mountId, { ...retained, session }); + // The authenticated projection is a live cache, not the access grant. + // Publish it even when the credential store is temporarily locked, then + // retry the durable cache write without holding UI freshness. + mounts = next; + notifyMountsChanged(); + const error = await persistDeferredState(next, mount); + if (error) { + scheduleDeferredWriteRetry(); + onError(error, mount); + } + return duplicates; + }); + for (const duplicate of superseded) { + void input.unmount(duplicate.mountId).catch((error) => onError(asError(error), duplicate)); + } + if (invalidatedAccessMounts.has(mount.mountId)) { + throw new RuntimeHostPermanentReconnectError('Shared Session access is no longer available'); + } + }; + + const refreshOnce = async (mountId: string): Promise => { + if (removingMounts.has(mountId) || invalidatedAccessMounts.has(mountId)) return; + // A catalog change may arrive after activation read its projection but + // before that projection is committed. Wait for the admitted activation + // and read again so the later authoritative state cannot be lost. Once + // admitted, shutdown cancels the cache read without affecting access. + while (true) { + const activation = [...activations].find((candidate) => candidate.mountId === mountId); + if (!activation) break; + await activation.task.catch(() => undefined); + if (removingMounts.has(mountId) || invalidatedAccessMounts.has(mountId)) return; + } + const mount = (await mutate(load)).get(mountId); + if (!mount) return; + const inspected = input.inspect(mountId); + if (inspected && inspected.readiness !== 'ready') return; + const session = await readSharedSession(mountId); + if (removingMounts.has(mountId) || invalidatedAccessMounts.has(mountId)) return; + if (!session) { + await clearSessionProjection(mountId); + return; + } + await recordSharedSession(mount, session); + }; + + const refresh = (mountId: string): Promise => { + if (closed) return Promise.resolve(); + const active = refreshes.get(mountId); + if (active) { + active.dirty = true; + return active.task; + } + const state: LiveGuestRefresh = { dirty: true, task: Promise.resolve() }; + state.task = (async () => { + let failure: unknown; + do { + state.dirty = false; + try { + await refreshOnce(mountId); + failure = undefined; + } catch (error) { + failure = error; + } + } while (state.dirty); + if (failure !== undefined) throw failure; + })().finally(() => { + if (refreshes.get(mountId) === state) refreshes.delete(mountId); + }); + refreshes.set(mountId, state); + return state.task; + }; + const activate = async ( activation: LiveGuestActivation, mount: GuestSessionMount, ): Promise => { activation.stage = 'connecting'; - await input.mount(resolveMountTarget(mount), activation.controller.signal, (phase) => { - if (activation.kind === 'import') { - reportImportProgress( - activation.onProgress, - collaborationProgressForConnectionPhase(phase), - ); - } - }, (endpoint) => recordPeerEndpoint(mount, endpoint)); + await input.mount( + resolveMountTarget(mount), + activation.controller.signal, + (phase) => { + if (activation.kind === 'import') { + reportImportProgress( + activation.onProgress, + collaborationProgressForConnectionPhase(phase), + ); + } + }, + (endpoint) => recordPeerEndpoint(mount, endpoint), + () => { + // The candidate publishes the live Session event. Refresh this + // durable projection independently; its eventual inventory change + // remains useful even when the credential-store write must retry. + void refresh(mount.mountId).catch((error: unknown) => onError(asError(error), mount)); + }, + ); // waitForReady observes host.status before mount resolves. Commit that // authenticated route snapshot before declaring the durable mount ready. await mutationTail; @@ -227,18 +487,37 @@ export function createDesktopGuestSessionMountService(input: { if (activation.kind === 'import') { reportImportProgress(activation.onProgress, 'finalizing_access'); } - const finalization = input.finalizeAccess( - mount.mountId, - activation.controller.signal, - activation.kind === 'import' - ? () => reportImportProgress(activation.onProgress, 'loading_session') - : undefined, - ); - activation.finalization = finalization; - try { - const result = await finalization; + const finalization = (async (): Promise => { + const result = await input.finalizeAccess(mount.mountId, activation.controller.signal, () => { + activation.accessActivated = true; + if (activation.kind === 'import') { + reportImportProgress(activation.onProgress, 'loading_session'); + } + }); + activation.accessActivated = true; activation.controller.signal.throwIfAborted(); + if (result === 'ready') { + activation.stage = 'hydrating'; + if (closed || removingMounts.has(mount.mountId)) { + return result; + } + const session = await readSharedSession(mount.mountId, activation.controller.signal); + activation.controller.signal.throwIfAborted(); + if (!session) { + await clearSessionProjection(mount.mountId); + throw new RuntimeHostPermanentReconnectError( + 'This shared Session is no longer available to the retained Guest access', + ); + } + await recordSharedSession(mount, session); + } else { + activation.stage = 'connecting'; + } return result; + })(); + activation.finalization = finalization; + try { + return await finalization; } finally { if (activation.finalization === finalization) activation.finalization = undefined; } @@ -249,12 +528,14 @@ export function createDesktopGuestSessionMountService(input: { closed || removingMounts.has(mount.mountId) || [...activations].some((activation) => activation.mountId === mount.mountId) - ) return; + ) + return; const activation: LiveGuestActivation = { kind: 'startup', controller: new AbortController(), mountId: mount.mountId, stage: 'connecting', + accessActivated: false, task: Promise.resolve(), }; activations.add(activation); @@ -263,24 +544,32 @@ export function createDesktopGuestSessionMountService(input: { while (!closed && !activation.controller.signal.aborted) { if (!(await load()).has(mount.mountId)) return; try { - await activate(activation, mount); - return; + const result = await activate(activation, mount); + if (result === 'ready') return; + if (closed || activation.controller.signal.aborted || removingMounts.has(mount.mountId)) + return; + await wait(delayMs, activation.controller.signal); + delayMs = Math.min(delayMs * 2, STARTUP_RETRY_MAX_MS); } catch (error) { - if ( - closed || - activation.controller.signal.aborted || - removingMounts.has(mount.mountId) - ) return; + if (closed || activation.controller.signal.aborted || removingMounts.has(mount.mountId)) + return; activation.stage = 'connecting'; const failure = asError(error); onError(failure, mount); - if (failure instanceof RuntimeHostPermanentReconnectError) return; + if (isRejectedAccessFailure(failure)) { + await clearSessionProjection(mount.mountId); + return; + } + if (failure instanceof RuntimeHostPermanentReconnectError) { + return; + } await wait(delayMs, activation.controller.signal); delayMs = Math.min(delayMs * 2, STARTUP_RETRY_MAX_MS); } } })().finally(() => { activations.delete(activation); + notifyMountsChanged(); }); void activation.task.catch((error) => { if (!activation.controller.signal.aborted) onError(asError(error), mount); @@ -292,7 +581,7 @@ export function createDesktopGuestSessionMountService(input: { try { const matching = [...activations].filter((activation) => activation.mountId === mountId); for (const activation of matching) { - if (activation.stage === 'connecting') { + if (activation.stage !== 'finalizing') { activation.controller.abort(new Error('Shared Session mount was removed')); } } @@ -311,6 +600,8 @@ export function createDesktopGuestSessionMountService(input: { return mount; }); if (!removed) return; + invalidatedAccessMounts.delete(mountId); + notifyMountsChanged(); for (const activation of activations) { if (activation.mountId === mountId) { activation.controller.abort(new Error('Shared Session mount was removed')); @@ -370,14 +661,16 @@ export function createDesktopGuestSessionMountService(input: { if (!(await load()).has(mount.mountId)) { throw new Error('Shared Session mount was removed while connecting'); } + reconcile = finalization === 'reconnecting'; return { kind: finalization === 'ready' ? 'connected' : 'recovering', mountId: mount.mountId, }; } catch (error) { if ( - activation.stage === 'finalizing' && - error instanceof RuntimeHostPairingFinalizationInterruptedError + (activation.stage === 'finalizing' && + error instanceof RuntimeHostPairingFinalizationInterruptedError) || + (activation.accessActivated && !(error instanceof RuntimeHostPermanentReconnectError)) ) { reconcile = true; } else { @@ -388,6 +681,7 @@ export function createDesktopGuestSessionMountService(input: { }); activation.controller.abort(new Error('Shared Session mount activation failed')); await input.unmount(mount.mountId).catch(() => undefined); + invalidatedAccessMounts.delete(mount.mountId); } return reconcile ? { kind: 'recovering', mountId: mount.mountId } @@ -399,6 +693,7 @@ export function createDesktopGuestSessionMountService(input: { } finally { activations.delete(activation); if (reconcile) beginStartupReconciliation(mount); + notifyMountsChanged(); } }; @@ -422,12 +717,11 @@ export function createDesktopGuestSessionMountService(input: { ...(onProgress ? { onProgress } : {}), controller: new AbortController(), stage: 'connecting', + accessActivated: false, task: Promise.resolve(), }; activations.add(activation); - const task = runImport(code, allowInsecure, activation).finally(() => { - activations.delete(activation); - }); + const task = runImport(code, allowInsecure, activation); activation.task = task; return task; }; @@ -436,15 +730,57 @@ export function createDesktopGuestSessionMountService(input: { async start() { if (closed) return; const current = await mutate(load); - for (const mount of current.values()) beginStartupReconciliation(mount); + for (const mount of current.values()) { + beginStartupReconciliation(mount); + } }, async list() { - return [...(await mutate(load)).values()] - .map(({ mountId, name }) => ({ mountId, name })) + // Projection updates replace the whole Map before notifying readers, so + // an initialized snapshot is safe to read while its cache write settles. + const current = mounts ?? (await mutate(load)); + return [...current.values()] + .map((mount) => { + const inspected = input.inspect(mount.mountId); + const activation = [...activations].find( + (candidate) => candidate.mountId === mount.mountId, + ); + const currentReadiness = deriveMountReadiness({ + accessInvalidated: invalidatedAccessMounts.has(mount.mountId), + activationKind: activation?.kind, + activationAccessActivated: activation?.accessActivated === true, + connectionReadiness: inspected?.readiness, + hasSessionProjection: mount.session !== undefined, + }); + return { + mountId: mount.mountId, + name: mount.name, + hostId: mount.rootId, + readiness: currentReadiness, + ...(inspected?.peerPath ? { peerPath: inspected.peerPath } : {}), + ...(!invalidatedAccessMounts.has(mount.mountId) && mount.session + ? { session: mount.session } + : {}), + }; + }) .sort((left, right) => left.name.localeCompare(right.name)); }, + async connectionChanged(mountId, error) { + if (closed) return; + if (error && isRejectedAccessFailure(error)) { + await clearSessionProjection(mountId); + return; + } + notifyMountsChanged(); + if (input.inspect(mountId)?.readiness !== 'ready') return; + // Initial activation owns its authenticated projection hydration. A + // later ready transition means a reconnect completed and needs a fresh + // projection from the recovered Host. + if ([...activations].some((activation) => activation.mountId === mountId)) return; + await refresh(mountId); + }, + importInvitation, cancelImport(operationId) { @@ -453,7 +789,7 @@ export function createDesktopGuestSessionMountService(input: { activation.kind === 'import' && activation.operationId === operationId, ); if (!operation) return 'settling'; - if (operation.stage === 'finalizing') return 'settling'; + if (operation.stage !== 'connecting') return 'settling'; operation.controller.abort(new Error('Shared Session import was cancelled')); return 'cancelled'; }, @@ -462,18 +798,54 @@ export function createDesktopGuestSessionMountService(input: { async close() { closed = true; + projectionLifetime.abort(new Error('Shared Session mount service is closed')); + deferredWriteRetryController.abort( + new Error('Shared Session mount service is closed'), + ); for (const activation of activations) { - if (activation.stage === 'connecting') { + if (activation.stage !== 'finalizing') { activation.controller.abort(new Error('Shared Session mount service is closed')); } } await Promise.allSettled([...activations].map((activation) => activation.task)); + await Promise.allSettled([...refreshes.values()].map((refresh) => refresh.task)); + if (deferredWriteRetry) await deferredWriteRetry; await mutationTail; + if (deferredWriteFailure) { + await flushDeferredWrite(); + if (deferredWriteFailure) { + onError(deferredWriteFailure.error, deferredWriteFailure.mount); + } + } activations.clear(); + refreshes.clear(); }, }; } +function deriveMountReadiness(input: { + readonly accessInvalidated: boolean; + readonly activationKind?: LiveGuestActivation['kind']; + readonly activationAccessActivated: boolean; + readonly connectionReadiness?: GuestSessionMountReadiness; + readonly hasSessionProjection: boolean; +}): GuestSessionMountReadiness { + if (input.accessInvalidated || input.connectionReadiness === 'unavailable') { + return 'unavailable'; + } + if ( + input.connectionReadiness === 'ready' && + input.hasSessionProjection && + (!input.activationKind || input.activationAccessActivated) + ) { + return 'ready'; + } + if (input.activationKind === 'import') return 'connecting'; + if (input.activationKind === 'startup') return 'reconnecting'; + if (input.connectionReadiness === 'ready') return 'reconnecting'; + return input.connectionReadiness ?? 'reconnecting'; +} + export function registerDesktopGuestSessionMountIpc( ipcMain: Pick, service: DesktopGuestSessionMountService, @@ -498,7 +870,8 @@ export function registerDesktopGuestSessionMountIpc( }, ); ipcMain.handle(channels[1], (_event, operationIdValue: unknown) => - service.cancelImport(requireOperationId(operationIdValue))); + service.cancelImport(requireOperationId(operationIdValue)), + ); ipcMain.handle(channels[2], () => service.list()); ipcMain.handle(channels[3], (_event, mountId: string) => service.remove(mountId)); ipcMain.handle(channels[4], () => { @@ -547,9 +920,11 @@ function decodeDocument(value: unknown): GuestSessionMountDocument { } function decodeMount(value: unknown): GuestSessionMount { + const keys = ['mountId', 'name', 'rootId', 'transport', 'credential']; + if (isRecord(value) && value.session !== undefined) keys.push('session'); if ( !isRecord(value) || - !hasExactKeys(value, ['mountId', 'name', 'rootId', 'transport', 'credential']) || + !hasExactKeys(value, keys) || typeof value.credential !== 'string' || !value.credential || /\s/u.test(value.credential) || @@ -571,9 +946,25 @@ function decodeMount(value: unknown): GuestSessionMount { rootId: target.rootId, transport: target.transport, credential: value.credential, + ...(value.session === undefined + ? {} + : { + session: retainedSession(decodeSharedSessionCatalogProjection(value.session)), + }), }; } +function retainedSession(session: SharedSessionCatalogProjection): SharedSessionCatalogProjection { + const { liveRunState: _liveRunState, ...retained } = session; + return retained; +} + +function isRejectedAccessFailure(error: Error): boolean { + return ( + error instanceof RuntimeHostProfileConnectionError && error.reason === 'credential_rejected' + ); +} + function isPeerPathUnavailable(error: unknown): boolean { if (!isRecord(error) || typeof error.code !== 'string') return false; return ( diff --git a/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts index a2fed87c34..b1dc078562 100644 --- a/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts @@ -27,7 +27,6 @@ import { type SessionChangedEvent, type SessionChangedReason, type SessionCatalo import { projectSessionCatalogSummary } from '@maka/runtime-host/client'; import type { SessionCatalogProjection, - SharedSessionCatalogProjection, SessionCreateInput, WorkspaceTarget, SessionModelTarget, @@ -82,10 +81,6 @@ export interface RuntimeHostSessionCatalogIpcDeps { newId?: () => string; } -export interface RuntimeHostSharedSessionCatalogIpcDeps { - getSession(): Promise; -} - export function registerRuntimeHostSessionCatalogIpc( deps: RuntimeHostSessionCatalogIpcDeps, ipcMain: ReconnectableReadIpcMain, @@ -238,51 +233,6 @@ export function registerRuntimeHostSessionCatalogIpc( }); } -export function registerRuntimeHostSharedSessionCatalogIpc( - deps: RuntimeHostSharedSessionCatalogIpcDeps, - ipcMain: ReconnectableReadIpcMain, -): void { - handleReconnectableRead(ipcMain, 'sessions:list', async (_event, filter?: unknown) => { - if (normalizeSessionListFilter(filter)?.subagentParentSessionId) return []; - const session = await deps.getSession(); - return session ? [session] : []; - }); -} - -export function toDesktopHostSharedSessionSummary( - session: SharedSessionCatalogProjection, -): DesktopHostSessionSummary { - return { - id: session.id, - revision: session.revision, - name: session.name, - activityAt: session.activityAt, - isFlagged: false, - isArchived: false, - labels: [], - labelsTruncated: false, - hasUnread: false, - ...(session.lastMessageAt === undefined ? {} : { lastMessageAt: session.lastMessageAt }), - ...(session.lastMessagePreview === undefined - ? {} - : { lastMessagePreview: session.lastMessagePreview }), - status: session.status, - ...(session.liveRunState === undefined - ? {} - : { runningTurnIds: [...session.liveRunState.runningTurnIds] }), - ...(session.blockedReason === undefined ? {} : { blockedReason: session.blockedReason }), - ...(session.statusUpdatedAt === undefined - ? {} - : { statusUpdatedAt: session.statusUpdatedAt }), - backend: 'ai-sdk', - llmConnectionSlug: '', - connectionLocked: true, - model: '', - permissionMode: 'ask', - shared: true, - }; -} - /** * Reads the archived premise off the remove options. * diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index fe84b755f4..4ea0d61c26 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -766,6 +766,7 @@ export interface MakaBridge { /** Reads only after the user invokes the invitation paste action. */ readInvitationClipboard(): Promise; listMounts(): Promise; + subscribeMountChanges(handler: () => void): () => void; removeMount(mountId: string): Promise; requestTurn( sessionId: string, diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index ce26162376..1f1662e810 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -123,7 +123,11 @@ import type { BotOnboardingSnapshot, BotOnboardingStartInput } from '@maka/core/ import type { HealthSnapshot } from '@maka/core/health'; import { collectRuntimeHostSessionCatalogsWithCoverage, + createRuntimeHostSessionCatalogRefresher, + recordObservedRuntimeHostSessionAuthority, + reconcileRuntimeHostSessionCatalog, resolveRuntimeHostSessionCatalog, + type RuntimeHostSessionCatalogCoverage, } from './runtime-host-session-catalog.js'; import { collectAvailablePendingTurnRequests } from './runtime-host-turn-request-inbox.js'; import type { ExecutionBoundaryReadModel, SandboxBoundaryResponse } from '@maka/core/sandbox-boundary'; @@ -221,6 +225,7 @@ import type { import type { AttachmentRef, InlineReference, QuoteRef } from '@maka/core/events'; import type { OnboardingMilestoneId } from '@maka/core/onboarding'; import { + decodeSharedSessionCatalogProjection, SCHEDULED_TASK_CATALOG_MAX_ITEMS, type OperationInput, type OperationOutcome, @@ -251,6 +256,7 @@ import { type DesktopSessionSummary, type DesktopSessionSummaryInput, } from '../shared/desktop-session-projection.js'; +import { projectDesktopSharedSessionSummary } from '../shared/shared-session-catalog-projection.js'; let activeRuntimeHost: DesktopTargetScope | undefined; let activeRuntimeHostGeneration = 0; @@ -266,8 +272,11 @@ const runtimeHostMetadata = new Map< readonly profileAccess: 'owner' | 'session_guest'; } >(); -const runtimeHostSessionScopes = new Map(); -let lastDesktopSessionCatalog: DesktopSessionSummary[] = []; +const runtimeHostSessionProfiles = new Map(); +let lastDesktopSessionCatalog: RuntimeHostSessionCatalogCoverage = { + sessions: [], + completeHostIds: [], +}; const newTaskChangeListeners = new Set<() => void>(); let previousMainProcessInterruptionRead: Promise | undefined; @@ -279,10 +288,25 @@ function runtimeHostMetadataFor(scope: DesktopTargetScope) { return runtimeHostMetadata.get(runtimeHostScopeKey(scope)); } -function recordRuntimeHostSessionScope(scope: DesktopTargetScope, sessionId: string): string { +function observeRuntimeHostSessionScope(scope: DesktopTargetScope, sessionId: string): { + readonly sessionId: string; + readonly authorityAccepted: boolean; +} { const projected = desktopSessionKey({ hostId: scope.hostId, sessionId }); - runtimeHostSessionScopes.set(projected, runtimeHostScopeKey(scope)); - return projected; + const profileId = runtimeHostMetadataFor(scope)?.profileId; + if (!profileId) throw new Error('Desktop Runtime Host metadata is unavailable'); + return { + sessionId: projected, + authorityAccepted: recordObservedRuntimeHostSessionAuthority( + runtimeHostSessionProfiles, + projected, + profileId, + ), + }; +} + +function recordRuntimeHostSessionScope(scope: DesktopTargetScope, sessionId: string): string { + return observeRuntimeHostSessionScope(scope, sessionId).sessionId; } type RuntimeHostProfileWireEvent = DesktopRuntimeHostProfileChangedEvent; @@ -299,11 +323,6 @@ ipcRenderer.on( previousScopeKey && (change.removed || (nextScopeKey !== undefined && previousScopeKey !== nextScopeKey)) ) { - for (const [sessionId, scopeKey] of runtimeHostSessionScopes) { - if (scopeKey !== previousScopeKey) continue; - if (nextScopeKey) runtimeHostSessionScopes.set(sessionId, nextScopeKey); - else runtimeHostSessionScopes.delete(sessionId); - } runtimeHostScopes.delete(previousScopeKey); runtimeHostMetadata.delete(previousScopeKey); if (change.removed) { @@ -398,9 +417,15 @@ async function runtimeHostSessionRef(sessionId: string): Promise<{ }> { const ref = parseDesktopSessionKey(sessionId); await runtimeHostScopeList(); - const recordedScopeKey = runtimeHostSessionScopes.get(sessionId); - let scope = recordedScopeKey ? runtimeHostScopes.get(recordedScopeKey) : undefined; - if (!scope) { + const recordedProfileId = runtimeHostSessionProfiles.get(sessionId); + let scope: DesktopTargetScope | undefined; + if (recordedProfileId) { + const scopeKey = runtimeHostProfiles.get(recordedProfileId); + scope = scopeKey ? runtimeHostScopes.get(scopeKey) : undefined; + if (!scope || scope.hostId !== ref.hostId) { + throw new Error('The Runtime Host for this task is unavailable'); + } + } else { const candidates = [...runtimeHostScopes.values()].filter(({ hostId }) => hostId === ref.hostId); if (candidates.length === 1) scope = candidates[0]; } @@ -709,9 +734,9 @@ async function invokeBranchFromTurn( if (!('ok' in result) || result.ok === false) { return result as DesktopSideConversationBranchResult; } - return { ok: true, session: projectSessionSummary(ref.scope, result.session) }; + return { ok: true, session: projectCreatedSessionSummary(ref.scope, result.session) }; } - return projectSessionSummary(ref.scope, result as DesktopSessionSummaryInput); + return projectCreatedSessionSummary(ref.scope, result as DesktopSessionSummaryInput); } async function invokeSessionInput( @@ -733,7 +758,7 @@ function projectSessionSummary( session: DesktopSessionSummaryInput, ): DesktopSessionSummary { const projected = projectSessionCatalogSummary(scope, session); - runtimeHostSessionScopes.set(projected.id, runtimeHostScopeKey(scope)); + runtimeHostSessionProfiles.set(projected.id, projected.profileId); return projected; } @@ -749,17 +774,17 @@ function projectSessionCatalogSummary( ); } -function recordSessionCatalogScopes(sessions: readonly DesktopSessionSummary[]): void { +function recordSessionCatalogAuthorities(sessions: readonly DesktopSessionSummary[]): void { for (const session of sessions) { - const scopeKey = runtimeHostProfiles.get(session.profileId); - const scope = scopeKey ? runtimeHostScopes.get(scopeKey) : undefined; - if (!scopeKey || !scope || scope.hostId !== session.runtimeHostId) { - throw new Error('Desktop Runtime Host Session scope is unavailable'); - } - runtimeHostSessionScopes.set(session.id, scopeKey); + runtimeHostSessionProfiles.set(session.id, session.profileId); } } +function commitDesktopSessionCatalog(catalog: RuntimeHostSessionCatalogCoverage): void { + recordSessionCatalogAuthorities(catalog.sessions); + lastDesktopSessionCatalog = catalog; +} + function projectOnboardingSnapshot( scope: DesktopTargetScope, snapshot: OnboardingSnapshot, @@ -767,16 +792,24 @@ function projectOnboardingSnapshot( return { ...snapshot, sessions: snapshot.sessions.map((session) => projectSessionSummary(scope, session)), - sessionSendOutcomes: Object.fromEntries( - Object.entries(snapshot.sessionSendOutcomes).map(([sessionId, outcome]) => [ - recordRuntimeHostSessionScope(scope, sessionId), - outcome, - ]), - ), + sessionSendOutcomes: projectOnboardingSendOutcomes(scope, snapshot.sessionSendOutcomes), }; } +function projectOnboardingSendOutcomes( + scope: DesktopTargetScope, + outcomes: OnboardingSnapshot['sessionSendOutcomes'], +): OnboardingSnapshot['sessionSendOutcomes'] { + return Object.fromEntries( + Object.entries(outcomes).map(([sessionId, outcome]) => [ + recordRuntimeHostSessionScope(scope, sessionId), + outcome, + ]), + ); +} + async function loadDesktopOnboardingSnapshot(): Promise { + const catalogSeed = desktopSessionCatalogRefresher.beginSeed(); const defaultScope = await activeRuntimeHostRef(); const readyScopes = await runtimeHostScopeList(); const scopes = [ @@ -798,15 +831,35 @@ async function loadDesktopOnboardingSnapshot(): Promise { } const snapshots = results.flatMap((result) => result.status === 'fulfilled' - ? [projectOnboardingSnapshot(result.value.scope, result.value.snapshot)] + ? [result.value] : [], ); + // A successful Owner onboarding snapshot is already an authenticated, + // complete catalog. Seed it before the independent catalog refresh so a + // transient sessions:list failure cannot blank the first renderer commit. + const ownerSnapshots = snapshots.filter( + ({ scope }) => runtimeHostMetadataFor(scope)?.profileAccess === 'owner', + ); + const completeHostIds = [...new Set(ownerSnapshots.map(({ scope }) => scope.hostId))]; + catalogSeed.commit({ + sessions: reconcileRuntimeHostSessionCatalog(lastDesktopSessionCatalog.sessions, { + sessions: ownerSnapshots.flatMap(({ scope, snapshot }) => + snapshot.sessions.map((session) => projectSessionCatalogSummary(scope, session))), + completeHostIds, + knownOwnerProfileIds: [...runtimeHostMetadata.values()].flatMap( + ({ profileId, profileAccess }) => profileAccess === 'owner' ? [profileId] : [], + ), + }), + completeHostIds, + }); + const sessions = await listDesktopSessions(); return { - ...snapshots[0], - sessions: snapshots.flatMap((snapshot) => snapshot.sessions), + ...snapshots[0]!.snapshot, + sessions, sessionSendOutcomes: Object.assign( {}, - ...snapshots.map((snapshot) => snapshot.sessionSendOutcomes), + ...snapshots.map(({ scope, snapshot }) => + projectOnboardingSendOutcomes(scope, snapshot.sessionSendOutcomes)), ), }; } @@ -890,6 +943,43 @@ function subscribeEveryRuntimeHostEvent( }; } +function subscribeGuestSessionMountChanges( + handler: () => void, +): () => void { + const listener = (): void => handler(); + ipcRenderer.on('session-collaboration:mounts:changed', listener); + return () => ipcRenderer.off('session-collaboration:mounts:changed', listener); +} + +const desktopSessionCatalogRefresher = createRuntimeHostSessionCatalogRefresher({ + currentCatalog: () => lastDesktopSessionCatalog, + listCatalog: async () => { + const owners = listDesktopOwnerSessionsWithCoverage(); + return resolveRuntimeHostSessionCatalog( + lastDesktopSessionCatalog.sessions, + owners, + () => [...runtimeHostMetadata.values()].flatMap(({ profileId, profileAccess }) => + profileAccess === 'owner' ? [profileId] : []), + listGuestSessionMountCatalog(), + ); + }, + commitCatalog: commitDesktopSessionCatalog, +}); + +function projectCreatedSessionSummary( + scope: DesktopTargetScope, + session: DesktopSessionSummaryInput, +): DesktopSessionSummary { + const projected = projectSessionSummary(scope, session); + // Guest catalog membership belongs exclusively to the retained mount + // service. A newly created Owner Session, however, must be visible before + // an older catalog read can settle. + if (runtimeHostMetadataFor(scope)?.profileAccess === 'owner') { + desktopSessionCatalogRefresher.admit(projected); + } + return projected; +} + async function listDesktopSessions( filter?: SessionListFilter, ): Promise { @@ -902,55 +992,63 @@ async function listDesktopSessions( ) as DesktopSessionSummaryInput[]; return sessions.map((session) => projectSessionSummary(parent.scope, session)); } - lastDesktopSessionCatalog = await resolveRuntimeHostSessionCatalog( - lastDesktopSessionCatalog, - listDesktopSessionsWithCoverage(), - () => [...runtimeHostMetadata.values()].map(({ profileId }) => profileId), - // Unknown Guest coverage cannot suppress healthy Owner catalogs or prove - // that a previously observed Guest mount was removed. - listKnownGuestMountProfileIds(), - ); - return lastDesktopSessionCatalog; + return (await desktopSessionCatalogRefresher.refresh()).sessions; } -async function listDesktopSessionsWithCoverage(): Promise<{ +async function listDesktopOwnerSessionsWithCoverage(): Promise<{ sessions: DesktopSessionSummary[]; completeHostIds: string[]; }> { const scopes = await runtimeHostScopeList(); - const catalog = await collectRuntimeHostSessionCatalogsWithCoverage( - scopes.map((scope) => { + return collectRuntimeHostSessionCatalogsWithCoverage( + scopes.flatMap((scope) => { const metadata = runtimeHostMetadataFor(scope); if (!metadata) throw new Error('Desktop Runtime Host metadata is unavailable'); - return { + if (metadata.profileAccess !== 'owner') return []; + return [{ hostId: scope.hostId, - access: metadata.profileAccess, sessions: ipcRenderer.invoke('sessions:list', scope) .then((sessions: DesktopSessionSummaryInput[]) => sessions.map((session) => projectSessionCatalogSummary(scope, session))), - }; + }]; }), ); - recordSessionCatalogScopes(catalog.sessions); - return catalog; } -async function listKnownGuestMountProfileIds(): Promise { +async function listGuestSessionMountCatalog(): Promise { const mounts: unknown = await ipcRenderer.invoke('session-collaboration:mount:list'); if (!Array.isArray(mounts)) { throw new Error('Desktop shared Session mounts are unavailable'); } - return mounts.map((mount) => { + const sessions: DesktopSessionSummary[] = []; + for (const mount of mounts) { if ( !mount || typeof mount !== 'object' || !('mountId' in mount) || - typeof mount.mountId !== 'string' + typeof mount.mountId !== 'string' || + !('name' in mount) || + typeof mount.name !== 'string' || + !('hostId' in mount) || + typeof mount.hostId !== 'string' ) { throw new Error('Desktop shared Session mount is invalid'); } - return mount.mountId; - }); + if (!('session' in mount) || mount.session === undefined) continue; + const session = decodeSharedSessionCatalogProjection(mount.session); + const summary = projectDesktopSharedSessionSummary(session); + const projected = projectDesktopSessionSummary( + { + hostId: mount.hostId, + profileId: mount.mountId, + profileName: mount.name, + profileKind: 'remote', + }, + summary, + ); + sessions.push(projected); + } + return sessions; } async function createDesktopSessionOnScope( @@ -958,7 +1056,7 @@ async function createDesktopSessionOnScope( input?: CreateSessionRequestInput, ): Promise { const session = await ipcRenderer.invoke('sessions:create', scope, input) as DesktopSessionSummaryInput; - return projectSessionSummary(scope, session); + return projectCreatedSessionSummary(scope, session); } function sendActiveRuntimeHost(channel: string, ...args: unknown[]): void { @@ -1333,6 +1431,9 @@ const makaBridge = { listMounts() { return ipcRenderer.invoke('session-collaboration:mount:list'); }, + subscribeMountChanges(handler) { + return subscribeGuestSessionMountChanges(() => handler()); + }, removeMount(mountId) { return ipcRenderer.invoke('session-collaboration:mount:remove', mountId); }, @@ -1978,7 +2079,7 @@ const makaBridge = { return listDesktopSessions(filter); }, listWithCoverage() { - return listDesktopSessionsWithCoverage(); + return desktopSessionCatalogRefresher.refresh(); }, /** * The single session-creation channel (#1433). `mode` names a @@ -2121,7 +2222,7 @@ const makaBridge = { const summary = await ipcRenderer.invoke( 'sessions:reviseBeforeTurn', ref.scope, ref.sessionId, input, ) as DesktopSessionSummaryInput; - return projectSessionSummary(ref.scope, summary); + return projectCreatedSessionSummary(ref.scope, summary); }, respondToSandboxBoundary(sessionId: string, response: SandboxBoundaryResponse): Promise { return invokeSessionRuntimeHost('sessions:respondToSandboxBoundary', sessionId, response); @@ -2220,18 +2321,26 @@ const makaBridge = { }; }, subscribeChanges(handler: (event: SessionChangedEvent) => void): () => void { - return subscribeEveryRuntimeHostEvent( + const unsubscribeRuntimeHosts = subscribeEveryRuntimeHostEvent( 'sessions:changed', - (scope, event: SessionChangedEvent) => - handler({ - ...event, - ...(event.sessionId - ? { - sessionId: recordRuntimeHostSessionScope(scope, event.sessionId), - } - : {}), - }), + (scope, event: SessionChangedEvent) => { + if (!event.sessionId) { + handler(event); + return; + } + const observed = observeRuntimeHostSessionScope(scope, event.sessionId); + handler(observed.authorityAccepted + ? { ...event, sessionId: observed.sessionId } + : { reason: 'updated', ts: event.ts }); + }, ); + const unsubscribeMounts = subscribeGuestSessionMountChanges(() => { + handler({ reason: 'updated', ts: Date.now() }); + }); + return () => { + unsubscribeRuntimeHosts(); + unsubscribeMounts(); + }; }, archive(sessionId: string, options?: { revisionFamily?: boolean }): Promise { return invokeSessionRuntimeHost('sessions:archive', sessionId, options); @@ -2479,7 +2588,7 @@ const makaBridge = { 'external-sessions:import', scope, input, ) as ExternalSessionImportIpcResult; return result.ok - ? { ...result, session: projectSessionSummary(scope, result.session as DesktopSessionSummaryInput) } + ? { ...result, session: projectCreatedSessionSummary(scope, result.session as DesktopSessionSummaryInput) } : result; }, }, @@ -3682,6 +3791,7 @@ if (process.env.MAKA_E2E === '1' && process.env.MAKA_E2E_USER_DATA_DIR) { const gates = new Map; oneShot: boolean }>(); const releases = new Map void; reject: (error: Error) => void }>(); let nextSessionObservationError: Error | undefined; + let nextTranscriptOpenError: Error | undefined; const invocableSkillsWaiters = new Map void>>(); const waitForLatch = async (key: LatchKey): Promise => { const gate = gates.get(key); @@ -3731,6 +3841,12 @@ if (process.env.MAKA_E2E === '1' && process.env.MAKA_E2E_USER_DATA_DIR) { disposed = true; }; }; + const openTranscript = makaBridge.transcripts.open.bind(makaBridge.transcripts); + makaBridge.transcripts.open = (...args) => { + const nextError = nextTranscriptOpenError; + nextTranscriptOpenError = undefined; + return nextError ? Promise.reject(nextError) : openTranscript(...args); + }; const listInvocableSkills = makaBridge.skills.listInvocable.bind(makaBridge.skills); makaBridge.skills.listInvocable = async (...args) => { try { @@ -3776,6 +3892,9 @@ if (process.env.MAKA_E2E === '1' && process.env.MAKA_E2E_USER_DATA_DIR) { rejectNextSessionObservation(message: string) { nextSessionObservationError = new Error(message); }, + rejectNextTranscriptOpen(message: string) { + nextTranscriptOpenError = new Error(message); + }, release(key: LatchKey) { releases.get(key)?.resolve(); releases.delete(key); diff --git a/apps/desktop/src/preload/runtime-host-session-catalog.ts b/apps/desktop/src/preload/runtime-host-session-catalog.ts index d8c5955d36..08b506ea97 100644 --- a/apps/desktop/src/preload/runtime-host-session-catalog.ts +++ b/apps/desktop/src/preload/runtime-host-session-catalog.ts @@ -21,86 +21,184 @@ import type { DesktopSessionSummary } from './bridge-contract.js'; export interface RuntimeHostSessionCatalogRequest { readonly hostId: string; - readonly access: 'owner' | 'session_guest'; readonly sessions: Promise; } export interface RuntimeHostSessionCatalogCoverage { readonly sessions: DesktopSessionSummary[]; + /** Hosts whose Owner catalog answered authoritatively. */ readonly completeHostIds: string[]; } export interface RuntimeHostSessionCatalogSnapshot extends RuntimeHostSessionCatalogCoverage { - /** Profiles still retained by Desktop, including unavailable Guest mounts. */ - readonly knownProfileIds: string[]; + /** Owner profiles still retained by Desktop. */ + readonly knownOwnerProfileIds: string[]; + /** The mount service's complete Guest projection, when it answered. */ + readonly guestSessions?: DesktopSessionSummary[]; +} + +export interface RuntimeHostSessionCatalogRefresher { + refresh(): Promise; + /** Commit a newly created Session and fence any catalog read started before it. */ + admit(session: DesktopSessionSummary): void; + /** Begin an asynchronous bootstrap read whose result may later seed the catalog. */ + beginSeed(): { + commit(catalog: RuntimeHostSessionCatalogCoverage): boolean; + }; +} + +export function createRuntimeHostSessionCatalogRefresher(input: { + readonly listCatalog: () => Promise; + readonly currentCatalog: () => RuntimeHostSessionCatalogCoverage; + readonly commitCatalog: (catalog: RuntimeHostSessionCatalogCoverage) => void; +}): RuntimeHostSessionCatalogRefresher { + let dirty = false; + let catalogGeneration = 0; + let active: Promise | undefined; + const commitCatalog = (catalog: RuntimeHostSessionCatalogCoverage): void => { + input.commitCatalog(catalog); + catalogGeneration += 1; + }; + const drain = async (): Promise => { + try { + let catalog = input.currentCatalog(); + do { + dirty = false; + try { + const candidate = await input.listCatalog(); + // A later invalidation supersedes this observation before it can + // mutate the authority map. The trailing read is the one to commit. + if (dirty) continue; + catalog = candidate; + commitCatalog(candidate); + } catch (error) { + // Like a successful stale read, a superseded failure cannot decide + // the drain. Let the already-admitted trailing read decide instead. + if (!dirty) throw error; + catalog = input.currentCatalog(); + } + } while (dirty); + return catalog; + } finally { + active = undefined; + } + }; + return { + admit(session) { + // The creation result is newer than every read already in flight. Mark + // those observations stale before publishing it so none can erase a + // Session that the Host has just committed. + dirty = true; + const current = input.currentCatalog(); + commitCatalog({ + ...current, + sessions: sortSessionCatalogs([ + ...current.sessions.filter(({ id }) => id !== session.id), + session, + ]), + }); + }, + beginSeed() { + const admittedCatalogGeneration = catalogGeneration; + return { + commit(catalog) { + if (catalogGeneration !== admittedCatalogGeneration) return false; + // The bootstrap snapshot is now the newest accepted observation. + // Fence an older catalog read before publishing it. + dirty = true; + commitCatalog(catalog); + return true; + }, + }; + }, + refresh() { + dirty = true; + if (!active) active = drain(); + return active; + }, + }; } export async function resolveRuntimeHostSessionCatalog( current: readonly DesktopSessionSummary[], coverage: Promise, - knownRuntimeProfileIds: () => readonly string[], - guestMountProfileIds: Promise, -): Promise { - const [snapshot, knownGuestProfileIds] = await Promise.all([ + knownOwnerProfileIds: () => readonly string[], + guestSessions: Promise, +): Promise { + const [snapshot, guests] = await Promise.all([ coverage, - guestMountProfileIds.catch(() => - current.flatMap((session) => session.shared === true ? [session.profileId] : []), + guestSessions.then( + (sessions) => ({ available: true as const, sessions }), + () => ({ available: false as const, sessions: [] }), ), ]); - return reconcileRuntimeHostSessionCatalog(current, { - ...snapshot, - knownProfileIds: [...knownRuntimeProfileIds(), ...knownGuestProfileIds], - }); + return { + sessions: reconcileRuntimeHostSessionCatalog(current, { + ...snapshot, + knownOwnerProfileIds: [...knownOwnerProfileIds()], + ...(guests.available ? { guestSessions: guests.sessions } : {}), + }), + completeHostIds: snapshot.completeHostIds, + }; } export async function collectRuntimeHostSessionCatalogsWithCoverage( requests: readonly RuntimeHostSessionCatalogRequest[], ): Promise { const results = await Promise.allSettled(requests.map((request) => request.sessions)); - const fulfilled = results.flatMap((result, index) => result.status === 'fulfilled' - ? [{ ...requests[index]!, sessions: result.value }] - : []); - const fulfilledRequests = new Set( - results.flatMap((result, index) => result.status === 'fulfilled' ? [requests[index]!] : []), + const fulfilled = results.flatMap((result, index) => + result.status === 'fulfilled' ? [{ ...requests[index]!, sessions: result.value }] : [], ); - if (requests.length > 0 && fulfilled.length === 0) { - throw new AggregateError( - results.flatMap((result) => result.status === 'rejected' ? [result.reason] : []), - 'Every Runtime Host Session Catalog request failed', - ); - } - const hostIds = [...new Set(requests.map((request) => request.hostId))]; return { sessions: sortSessionCatalogs(fulfilled.flatMap((entry) => entry.sessions)), - completeHostIds: hostIds.filter((hostId) => { - const hostRequests = requests.filter((request) => request.hostId === hostId); - const ownerRequests = hostRequests.filter((request) => request.access === 'owner'); - return ownerRequests.length > 0 - ? ownerRequests.some((request) => fulfilledRequests.has(request)) - : hostRequests.every((request) => fulfilledRequests.has(request)); - }), + completeHostIds: [...new Set(fulfilled.map(({ hostId }) => hostId))], }; } /** - * Commits complete Host catalogs authoritatively while retaining the last - * accepted rows for a Host that still exists but cannot answer this read. - * An explicitly removed profile is absent from knownProfileIds and therefore - * retires immediately; transport availability alone cannot change access. + * An observation may establish an unknown Session authority, but only an + * accepted catalog may replace one. Returns false when the observation came + * from a different profile and should therefore trigger a generic refresh. + */ +export function recordObservedRuntimeHostSessionAuthority( + authorities: Map, + sessionId: string, + profileId: string, +): boolean { + const accepted = authorities.get(sessionId); + if (accepted === undefined) { + authorities.set(sessionId, profileId); + return true; + } + return accepted === profileId; +} + +/** + * Commits complete Owner catalogs per Host while retaining the last accepted + * Owner rows for an authority that cannot answer. Guest rows come exclusively + * from the mount service: a successful mount read replaces them completely, + * while a failed read retains the last authenticated projection. */ export function reconcileRuntimeHostSessionCatalog( current: readonly DesktopSessionSummary[], snapshot: RuntimeHostSessionCatalogSnapshot, ): DesktopSessionSummary[] { const completeHostIds = new Set(snapshot.completeHostIds); - const knownProfileIds = new Set(snapshot.knownProfileIds); - return sortSessionCatalogs([ + const knownOwnerProfileIds = new Set(snapshot.knownOwnerProfileIds); + const liveSessions = sortSessionCatalogs([ ...snapshot.sessions, - ...current.filter( - (session) => - knownProfileIds.has(session.profileId) && !completeHostIds.has(session.runtimeHostId), - ), + ...(snapshot.guestSessions ?? []), ]); + const liveSessionIds = new Set(liveSessions.map(({ id }) => id)); + const fallbackSessions = current.filter( + (session) => + !liveSessionIds.has(session.id) && + (session.shared === true + ? snapshot.guestSessions === undefined + : knownOwnerProfileIds.has(session.profileId) && + !completeHostIds.has(session.runtimeHostId)), + ); + return sortSessionCatalogs([...liveSessions, ...fallbackSessions]); } function sortSessionCatalogs(sessions: DesktopSessionSummary[]): DesktopSessionSummary[] { diff --git a/apps/desktop/src/renderer/app-shell-effects.ts b/apps/desktop/src/renderer/app-shell-effects.ts index 4a9b383fdf..00d596e466 100644 --- a/apps/desktop/src/renderer/app-shell-effects.ts +++ b/apps/desktop/src/renderer/app-shell-effects.ts @@ -51,12 +51,8 @@ import { ShellRunHydration, type ShellRunUpdatesBySession, } from './shell-run-update-state.js'; -import { runtimeHostChangeRetiresSession } from '../shared/runtime-host-identity.js'; -import { - createDesktopTranscriptRangeController, - DesktopTranscriptRangeStore, - type DesktopTranscriptRangeController, -} from './desktop-transcript-range-store.js'; +import { sessionCatalogRetiresSession } from '../shared/runtime-host-identity.js'; +import * as desktopTranscript from './desktop-transcript-range-store.js'; type RefBox = { current: T }; @@ -154,7 +150,6 @@ export function useAppShellBootstrapSubscriptions(options: { clearPendingTurnActionsForSession: (sessionId: string) => void; /** Releases a send's pending claim once the authority names that turn. */ confirmLiveTurn: (sessionId: string, turnId: string) => void; - clearSessionRendererState: (sessionId: string) => void; createSession: () => Promise | void; handleConnectionEvent: (event: ConnectionEvent) => void; openHelp: () => void; @@ -169,13 +164,12 @@ export function useAppShellBootstrapSubscriptions(options: { refreshShellSettings: () => Promise; refreshSessions: () => Promise; rendererMountedRef: RefBox; - setActiveId: (sessionId: string | undefined) => void; - setMessages: (messages: StoredMessage[]) => void; + retireSession: (sessionId: string) => void; setSessionEventHealthBySession: SessionEventHealthUpdater; toastApi: ToastApi; }) { const runDeferredStartupRefreshes = useEffectEvent(() => { - void options.refreshSessions(); + void options.bootstrapSessions(); void options.applyE2eFixture(); }); const handleConnectionSubscriptionEvent = useEffectEvent((event: ConnectionEvent) => { @@ -184,10 +178,8 @@ export function useAppShellBootstrapSubscriptions(options: { const handleRuntimeHostChange = useEffectEvent((event: DesktopRuntimeHostProfileChangedEvent) => { void options.refreshSessions().then((sessions) => { const activeSessionId = options.activeIdRef.current; - if (!runtimeHostChangeRetiresSession(event, activeSessionId, sessions)) return; - options.setActiveId(undefined); - options.setMessages([]); - options.clearSessionRendererState(activeSessionId); + if (!sessionCatalogRetiresSession(activeSessionId, sessions)) return; + options.retireSession(activeSessionId); }); if (event.readiness !== 'ready') return; if (!event.isDefault) return; @@ -215,7 +207,7 @@ export function useAppShellBootstrapSubscriptions(options: { if (event.sessionId && event.turnId) { options.confirmLiveTurn(event.sessionId, event.turnId); } - void options.refreshSessions(); + const refreshedSessions = options.refreshSessions(); if (event.reason === 'created' || event.reason === 'migrated') { void options.refreshProjects(); } @@ -243,12 +235,11 @@ export function useAppShellBootstrapSubscriptions(options: { const copy = getDesktopConversationCopy(options.uiLocale).actions; options.toastApi.info(copy.modelReboundTitle, copy.modelReboundDescription(event.modelId)); } - if (event.reason === 'deleted' && event.sessionId && event.sessionId === options.activeIdRef.current) { - const deletedSessionId = event.sessionId; - options.setActiveId(undefined); - options.setMessages([]); - options.clearSessionRendererState(deletedSessionId); - } + void refreshedSessions.then((sessions) => { + const activeSessionId = options.activeIdRef.current; + if (!sessionCatalogRetiresSession(activeSessionId, sessions)) return; + options.retireSession(activeSessionId); + }); }, ); // Both shortcuts fire while the composer has focus — they always did, and @@ -328,7 +319,7 @@ export function useAppShellBootstrapSubscriptions(options: { export function useActiveSessionEvents(options: { uiLocale: UiLocale; activeId: string | undefined; - activeProfileId: string | undefined; + observationAuthorityRevision: number; activeIdRef: RefBox; handleEvent: (sessionId: string, event: SessionEvent) => void; beginObservationSeed?: (sessionId: string) => number; @@ -336,20 +327,31 @@ export function useActiveSessionEvents(options: { setMessageLoadErrorBySession: (updater: (current: Record) => Record) => void; setMessageLoadPending: (pending: boolean) => void; setMessages: (messages: StoredMessage[]) => void; - transcriptRangeRef: RefBox; + transcriptRangeRef: RefBox; setSessionEventHealthBySession: SessionEventHealthUpdater; toastApi: Pick; }) { const activeId = options.activeId; + const clearMessageLoadError = useEffectEvent((sessionId: string) => { + options.setMessageLoadErrorBySession((current) => { + if (!current[sessionId]) return current; + const next = { ...current }; + delete next[sessionId]; + return next; + }); + }); const applyTranscript = useEffectEvent(( sessionId: string, - store: DesktopTranscriptRangeStore, + store: desktopTranscript.DesktopTranscriptRangeStore, isDisposed: () => boolean, ) => { if (!isDisposed() && options.activeIdRef.current === sessionId) { const snapshot = store.snapshot(); options.setMessages([...snapshot.messages]); - if (snapshot.ready) options.setMessageLoadPending(false); + if (snapshot.ready) { + clearMessageLoadError(sessionId); + options.setMessageLoadPending(false); + } } }); const applyReadError = useEffectEvent((sessionId: string, error: unknown, isDisposed: () => boolean) => { @@ -407,17 +409,13 @@ export function useActiveSessionEvents(options: { useLayoutEffect(() => { if (!activeId) return; let disposed = false; + const isDisposed = () => disposed; let observationAttempt = 0; let observationFailures = 0; let observationRetryTimer: ReturnType | undefined; let unsubscribeSessionEvents = () => {}; - const transcript = new DesktopTranscriptRangeStore(activeId); - options.setMessageLoadErrorBySession((current) => { - if (!current[activeId]) return current; - const next = { ...current }; - delete next[activeId]; - return next; - }); + const transcript = new desktopTranscript.DesktopTranscriptRangeStore(activeId); + clearMessageLoadError(activeId); options.setSessionEventHealthBySession((current) => ({ ...current, [activeId]: createSessionEventStreamSubscription({ @@ -431,11 +429,9 @@ export function useActiveSessionEvents(options: { (batch) => { if (disposed) return; try { - if (transcript.accept(batch)) { - applyTranscript(activeId, transcript, () => disposed); - } + if (transcript.accept(batch)) applyTranscript(activeId, transcript, isDisposed); } catch (error) { - applyReadError(activeId, error, () => disposed); + applyReadError(activeId, error, isDisposed); } }, (cancel) => { @@ -443,10 +439,13 @@ export function useActiveSessionEvents(options: { else signal.addEventListener('abort', cancel, { once: true }); }, ); - const controller = createDesktopTranscriptRangeController(transcript, openTranscript); - void controller.ready().catch((error) => { - applyReadError(activeId, error, () => disposed); - }); + const controller = desktopTranscript.createRecoveringDesktopTranscriptRangeController( + transcript, + openTranscript, + { + onError: (error) => applyReadError(activeId, error, isDisposed), + }, + ); options.transcriptRangeRef.current = controller; const subscribeSessionEvents = () => { const attempt = ++observationAttempt; @@ -463,16 +462,19 @@ export function useActiveSessionEvents(options: { }, () => { if (attempt !== observationAttempt) return; + controller.observationChanged('ready'); observationFailures = 0; completeObservationSeed(activeId, observationGeneration); }, (phase) => { if (attempt !== observationAttempt) return; + controller.observationChanged(phase); if (phase === 'pending') beginObservationSeed(activeId); else completeObservationSeed(activeId); }, () => { if (disposed || attempt !== observationAttempt) return; + controller.observationChanged('pending'); unsubscribeCurrent(); observationFailures += 1; const retryDelayMs = Math.min(100 * (2 ** (observationFailures - 1)), 2_000); @@ -500,7 +502,7 @@ export function useActiveSessionEvents(options: { unsubscribeSessionEvents(); markSessionEventStreamClosed(activeId); }; - }, [activeId, options.activeProfileId]); + }, [activeId, options.observationAuthorityRevision]); } export function useShellRunUpdates(options: { diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index 3f3a212dd7..78b311a731 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -200,13 +200,7 @@ import { useSessionEventHealthPolling, useShellRunUpdates, } from './app-shell-effects'; -import { - EMPTY_LIVE_CONTENT_SEED, - beginLiveContentSeed, - completeLiveContentSeed, - liveContentSeedRevision, - type LiveContentSeed, -} from './live-content-seed'; +import * as liveContent from './live-content-seed'; import { loadComposerDefaults, saveComposerDefaults } from './composer-defaults'; import { useTurnActionRegistry } from './use-turn-action-registry'; import { useComposerAttachments } from './use-composer-attachments'; @@ -583,8 +577,7 @@ function AppShellContent({ uiLocale, target: { kind: 'session', sessionId: ownerActiveId }, }); - const startupConnectionSnapshot = - initialOnboardingSnapshot ?? onboarding.mountedSnapshotHandoff; + const startupConnectionSnapshot = onboarding.snapshot; const newTaskUsesDefaultHost = taskEntry.selectors.usesDefaultHost; let newTaskConnectionSnapshot = newTaskConnections.snapshot; if (newTaskConnections.projection.status !== 'ready' && newTaskUsesDefaultHost) { @@ -1258,13 +1251,11 @@ function AppShellContent({ // Re-entrancy lock only — a ref, not state, because nothing renders // from it (#1433 removed its last reader with the first-run hero). const sessionStartPendingRef = useRef(false); - // Seed sessions from the onboarding snapshot on first load — the snapshot - // already fetches the session list + connections internally, so separate - // Session and connection snapshot IPCs are redundant. - // This lets the UI show the sidebar + model picker immediately on first load. + // Seed a snapshot captured before React mounted so the sidebar can paint + // immediately. The subscription bootstrap reconciles once through the live + // Session catalog on the next frame; later onboarding pulls still own + // readiness and connection data, but never overwrite that catalog. const initialSnapshotSeededRef = useRef(false); - const mountedSnapshotSeededRef = useRef(false); - const bootstrapFallbackStartedRef = useRef(false); // useLayoutEffect, NOT useEffect: the snapshot render flips // `isOnboardingLoading` off while `sessions` is still []. A passive // effect seeds sessions AFTER the browser paints that frame, so users @@ -1272,54 +1263,29 @@ function AppShellContent({ // "配置页闪了一下" startup flash). Layout effects run before paint, // so the seeded sessions and the un-gated frame commit together. useLayoutEffect(() => { - // Snapshot IPC failed — the seed path will never run, so fall back - // to the classic boot pull or the sidebar stays empty forever. - if ( - onboarding.error && - !initialOnboardingSnapshot && - !onboarding.mountedSnapshotHandoff && - !bootstrapFallbackStartedRef.current - ) { - bootstrapFallbackStartedRef.current = true; - void bootstrapSessions(); - void defaultHostConnections.refreshConnections(); - return; - } - let snapshot: OnboardingSnapshot | null = null; - let releaseSelectionLease = false; - if (!initialSnapshotSeededRef.current && initialOnboardingSnapshot) { - initialSnapshotSeededRef.current = true; - snapshot = initialOnboardingSnapshot; - } else if ( - !bootstrapFallbackStartedRef.current && - !mountedSnapshotSeededRef.current && - onboarding.mountedSnapshotHandoff - ) { - mountedSnapshotSeededRef.current = true; - snapshot = onboarding.mountedSnapshotHandoff; - releaseSelectionLease = true; - } - if (!snapshot) return; - // Seed sessions. Display normalization MUST run here too — this is - // Display normalization prevents legacy blocked/unknown - // sessions flash an 已阻塞 group on first paint until the first - // refreshSessions() overwrites the seed. - const next = seedSessions(snapshot.sessions); + if (initialSnapshotSeededRef.current || !initialOnboardingSnapshot) return; + initialSnapshotSeededRef.current = true; + // This prop settled before React mounted, so it is the only onboarding + // value allowed to seed the catalog. Later snapshots must go through the + // authoritative refresher or they can overwrite a newer Guest-inclusive + // catalog with an older point-in-time view. + const next = seedSessions(initialOnboardingSnapshot.sessions); bootstrapSelectionLease.reconcile(collapseSessionRevisions(next)); - if (releaseSelectionLease) bootstrapSelectionLease.release(); - }, [initialOnboardingSnapshot, onboarding.mountedSnapshotHandoff, onboarding.error]); + }, [initialOnboardingSnapshot]); useEffect(() => { - const snapshot = initialOnboardingSnapshot ?? onboarding.mountedSnapshotHandoff; - if (!snapshot) return; - defaultHostConnections.seedSnapshot({ - connections: snapshot.connections, - defaultConnection: snapshot.defaultSlug, - chatModelChoices: snapshot.chatModelChoices, - }); - }, [ - initialOnboardingSnapshot, - onboarding.mountedSnapshotHandoff, - ]); + const snapshot = onboarding.snapshot; + if (snapshot) { + defaultHostConnections.seedSnapshot({ + connections: snapshot.connections, + defaultConnection: snapshot.defaultSlug, + chatModelChoices: snapshot.chatModelChoices, + }); + } else if (onboarding.error && !initialOnboardingSnapshot) { + // Session bootstrap is independent above. If onboarding itself failed, + // retain the previous connection-specific recovery path as well. + void defaultHostConnections.refreshConnections(); + } + }, [initialOnboardingSnapshot, onboarding.error, onboarding.snapshot]); // PR110c (@kenji review): suppress hero AND the fallback EmptyChatHero // while the initial snapshot is in flight. Otherwise sessions.length===0 // + snapshot===null flashes the prompt-suggestion EmptyChatHero before @@ -2291,7 +2257,6 @@ function AppShellContent({ bootstrapSessions, clearPendingTurnActionsForSession: turnActionRegistry.clearForSession, confirmLiveTurn: sessionUiController.confirmLiveTurn, - clearSessionRendererState, createSession, handleConnectionEvent, openHelp, @@ -2306,8 +2271,11 @@ function AppShellContent({ refreshShellSettings, refreshSessions, rendererMountedRef, - setActiveId, - setMessages, + retireSession: (sessionId) => { + setActiveId(undefined); + setMessages([]); + clearSessionRendererState(sessionId); + }, setSessionEventHealthBySession: sessionUiController.setSessionEventHealthBySession, toastApi, }); @@ -2316,11 +2284,13 @@ function AppShellContent({ themePalette, themePref, }); - const [activeEventSeed, setActiveEventSeed] = useState(EMPTY_LIVE_CONTENT_SEED); + const [activeEventSeed, setActiveEventSeed] = useState( + liveContent.EMPTY_LIVE_CONTENT_SEED, + ); const activeEventSeedRef = useRef(activeEventSeed); activeEventSeedRef.current = activeEventSeed; const beginObservationSeed = (sessionId: string) => { - const next = beginLiveContentSeed(activeEventSeedRef.current, sessionId); + const next = liveContent.beginLiveContentSeed(activeEventSeedRef.current, sessionId); activeEventSeedRef.current = next; markDisplayPending(sessionId); setActiveEventSeed(next); @@ -2332,7 +2302,7 @@ function AppShellContent({ if (current.sessionId !== sessionId || current.generation !== expected) return; flushDisplayEvents(sessionId); markDisplayReady(sessionId); - const next = completeLiveContentSeed(current, sessionId, expected); + const next = liveContent.completeLiveContentSeed(current, sessionId, expected); activeEventSeedRef.current = next; setActiveEventSeed(next); const firstSendWaiter = firstSendObservationWaitersRef.current.get(sessionId); @@ -2343,10 +2313,16 @@ function AppShellContent({ } void retireCancelledTransientMessages(sessionId); }; + const observationAuthorityRef = useRef(liveContent.EMPTY_SESSION_OBSERVATION_AUTHORITY); + observationAuthorityRef.current = liveContent.advanceSessionObservationAuthority( + observationAuthorityRef.current, + activeId, + activeSession?.profileId, + ); useActiveSessionEvents({ uiLocale, activeId, - activeProfileId: activeSession?.profileId, + observationAuthorityRevision: observationAuthorityRef.current.revision, activeIdRef, handleEvent, beginObservationSeed, @@ -3079,7 +3055,7 @@ function AppShellContent({ onLoadEarlierHistory={(anchorTurnId) => loadTranscriptHistory('earlier', anchorTurnId)} onReturnToLatestHistory={() => loadTranscriptHistory('latest')} - liveContentSeedRevision={liveContentSeedRevision(activeEventSeed, activeId)} + liveContentSeedRevision={liveContent.liveContentSeedRevision(activeEventSeed, activeId)} messages={messages} transientMessages={transientMessages} messageLoading={activeMessageLoading} diff --git a/apps/desktop/src/renderer/astryx-theme/maka.js b/apps/desktop/src/renderer/astryx-theme/maka.js index 1711f5e85a..3151aa8009 100644 --- a/apps/desktop/src/renderer/astryx-theme/maka.js +++ b/apps/desktop/src/renderer/astryx-theme/maka.js @@ -8,7 +8,7 @@ import { neutralIconRegistry } from '@astryxdesign/theme-neutral'; /** - * maka theme — built by `npx astryx theme build` + * maka theme — built by `astryx theme build` * Import the CSS file alongside this module: * * import { makaTheme } from './maka'; diff --git a/apps/desktop/src/renderer/desktop-transcript-range-store.ts b/apps/desktop/src/renderer/desktop-transcript-range-store.ts index 6cee40e944..a3d11b0d56 100644 --- a/apps/desktop/src/renderer/desktop-transcript-range-store.ts +++ b/apps/desktop/src/renderer/desktop-transcript-range-store.ts @@ -98,6 +98,101 @@ export function createDesktopTranscriptRangeController( }; } +export interface DesktopTranscriptReconnectRecovery { + transcriptFailed(error: unknown): void; + observationChanged(phase: 'pending' | 'ready'): void; + close(): void; +} + +export function createDesktopTranscriptReconnectRecovery(options: { + reload(): Promise; + onError(error: unknown): void; +}): DesktopTranscriptReconnectRecovery { + let closed = false; + let observationReady = false; + let readinessGeneration = 0; + let needsRecovery = false; + let recoveryTask: Promise | undefined; + + const recover = () => { + if (closed || !observationReady || !needsRecovery || recoveryTask) return; + const admittedReadinessGeneration = readinessGeneration; + needsRecovery = false; + const task = Promise.resolve().then(async () => { + try { + if (closed) return; + await options.reload(); + } catch (error) { + if (closed) return; + needsRecovery = true; + options.onError(error); + } + }); + recoveryTask = task; + const settle = () => { + if (recoveryTask !== task) return; + recoveryTask = undefined; + if ( + needsRecovery + && observationReady + && readinessGeneration > admittedReadinessGeneration + ) recover(); + }; + void task.then(settle, settle); + }; + + return { + transcriptFailed(error) { + if (closed) return; + needsRecovery = true; + options.onError(error); + recover(); + }, + observationChanged(phase) { + if (closed) return; + if (phase === 'pending') { + observationReady = false; + return; + } + if (!observationReady) readinessGeneration += 1; + observationReady = true; + recover(); + }, + close() { + closed = true; + observationReady = false; + }, + }; +} + +export interface RecoveringDesktopTranscriptRangeController + extends DesktopTranscriptRangeController { + observationChanged(phase: 'pending' | 'ready'): void; +} + +export function createRecoveringDesktopTranscriptRangeController( + store: DesktopTranscriptRangeStore, + open: (signal: AbortSignal) => Promise, + options: { + onError(error: unknown): void; + }, +): RecoveringDesktopTranscriptRangeController { + const controller = createDesktopTranscriptRangeController(store, open); + const recovery = createDesktopTranscriptReconnectRecovery({ + reload: controller.reload, + ...options, + }); + void controller.ready().catch(recovery.transcriptFailed); + return { + ...controller, + observationChanged: recovery.observationChanged, + async close() { + recovery.close(); + await controller.close(); + }, + }; +} + interface PendingRecord { readonly source: 'durable' | 'overlay'; readonly identity: number | string; diff --git a/apps/desktop/src/renderer/features/session-collaboration/ports.ts b/apps/desktop/src/renderer/features/session-collaboration/ports.ts index 40675c265a..4684f67474 100644 --- a/apps/desktop/src/renderer/features/session-collaboration/ports.ts +++ b/apps/desktop/src/renderer/features/session-collaboration/ports.ts @@ -47,6 +47,7 @@ export interface SessionCollaborationServices { cancelImport(operationId: string): Promise; readInvitationClipboard(): Promise; listMounts(): Promise; + subscribeMountChanges(handler: () => void): () => void; removeMount(mountId: string): Promise; requestTurn( sessionId: string, diff --git a/apps/desktop/src/renderer/features/session-collaboration/ui/session-collaboration-join-dialog.tsx b/apps/desktop/src/renderer/features/session-collaboration/ui/session-collaboration-join-dialog.tsx index d4c0e27a05..3736b6e9a0 100644 --- a/apps/desktop/src/renderer/features/session-collaboration/ui/session-collaboration-join-dialog.tsx +++ b/apps/desktop/src/renderer/features/session-collaboration/ui/session-collaboration-join-dialog.tsx @@ -21,7 +21,10 @@ import { useEffect, useRef, useState } from 'react'; import { Dialog, DialogHeader } from '@astryxdesign/core/Dialog'; import { Layout, LayoutContent, LayoutFooter } from '@astryxdesign/core/Layout'; import { List, ListItem } from '@astryxdesign/core/List'; +import { HStack } from '@astryxdesign/core/Stack'; +import { Tooltip } from '@astryxdesign/core/Tooltip'; import { + Badge, Banner, Button, FormLayout, @@ -59,6 +62,12 @@ export interface SessionCollaborationJoinCopy { readonly recoveryStarted: string; readonly recoveryStartedBody: string; readonly retainedTasks: string; + readonly mountConnected: string; + readonly mountConnecting: string; + readonly mountReconnecting: string; + readonly mountUnavailable: string; + readonly directConnection: string; + readonly memberTransitConnection: string; readonly disconnect: string; readonly disconnectFailed: string; } @@ -88,14 +97,21 @@ export function SessionCollaborationJoinDialog(props: { useEffect(() => { open.current = true; let disposed = false; - void services.listMounts().then( - (next) => { - if (!disposed) setMounts(next); - }, - () => undefined, - ); + let request = 0; + const refresh = () => { + const currentRequest = ++request; + void services.listMounts().then( + (next) => { + if (!disposed && request === currentRequest) setMounts(next); + }, + () => undefined, + ); + }; + refresh(); + const unsubscribe = services.subscribeMountChanges(refresh); return () => { disposed = true; + unsubscribe(); open.current = false; const operationId = activeOperationId.current; if (operationId) void services.cancelImport(operationId); @@ -290,16 +306,36 @@ export function SessionCollaborationJoinDialog(props: { {mounts.map((mount) => ( void disconnect(mount.mountId)} - /> + + {mount.peerPath ? ( + + + + + + ) : mount.readiness !== 'ready' ? ( + + ) : null} +